FINERACT-2169: New command processing infrastructure

fineract-command/src/main/java/org/apache/fineract/command/core/Command.java

@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.time.OffsetDateTime;
+import java.util.UUID;
+import lombok.Data;
+import lombok.experimental.FieldNameConstants;
+
+@Data
+@FieldNameConstants
+public class Command<T> implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    private UUID id;
+
+    private OffsetDateTime createdAt;
+
+    private String tenantId;
+
+    private String username;
+
+    private T payload;
+}

fineract-command/src/main/java/org/apache/fineract/command/core/CommandConstants.java

@@ -0,0 +1,27 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public final class CommandConstants {
+
+    public static final String COMMAND_REQUEST_ID = "x-fineract-request-id";
+}

fineract-command/src/main/java/org/apache/fineract/command/core/CommandExecutor.java

@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core;
+
+import java.util.function.Supplier;
+
+public interface CommandExecutor {
+
+    <REQ, RES> Supplier<RES> execute(Command<REQ> command);
+}

fineract-command/src/main/java/org/apache/fineract/command/core/CommandHandler.java

@@ -0,0 +1,32 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core;
+
+import com.google.common.reflect.TypeToken;
+
+public interface CommandHandler<REQ, RES> {
+
+    RES handle(Command<REQ> command);
+
+    default boolean matches(Command<REQ> command) {
+        TypeToken<REQ> handlerType = new TypeToken<>(getClass()) {};
+
+        return handlerType.getRawType().isAssignableFrom(command.getPayload().getClass());
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/core/CommandMiddleware.java

@@ -0,0 +1,25 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core;
+
+@FunctionalInterface
+public interface CommandMiddleware {
+
+    void invoke(Command<?> command);
+}

fineract-command/src/main/java/org/apache/fineract/command/core/CommandPipeline.java

@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core;
+
+import java.util.function.Supplier;
+
+public interface CommandPipeline {
+
+    <REQ, RES> Supplier<RES> send(Command<REQ> command);
+}

fineract-command/src/main/java/org/apache/fineract/command/core/CommandProperties.java

@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core;
+
+import com.lmax.disruptor.dsl.ProducerType;
+import java.io.Serial;
+import java.io.Serializable;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.FieldNameConstants;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@Builder
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@FieldNameConstants
+@ConfigurationProperties(prefix = "fineract.command")
+public final class CommandProperties implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    @Builder.Default
+    private Boolean enabled = true;
+
+    @Builder.Default
+    private CommandExecutorType executor = CommandExecutorType.sync;
+
+    @Builder.Default
+    private Integer ringBufferSize = 1024;
+
+    @Builder.Default
+    private ProducerType producerType = ProducerType.SINGLE;
+
+    public enum CommandExecutorType {
+        sync, async, disruptor
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/core/CommandRouter.java

@@ -0,0 +1,25 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core;
+
+@FunctionalInterface
+public interface CommandRouter {
+
+    <REQ, RES> CommandHandler<REQ, RES> route(Command<REQ> command);
+}

fineract-command/src/main/java/org/apache/fineract/command/core/exception/CommandHandlerNotFoundException.java

@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core.exception;
+
+import org.apache.fineract.command.core.Command;
+
+public class CommandHandlerNotFoundException extends RuntimeException {
+
+    public CommandHandlerNotFoundException(Command<?> command) {
+        super("Cannot find a matching handler for " + (command != null ? command.getClass().getSimpleName() : "UKNOWN") + " command");
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/core/exception/CommandIllegalArgumentException.java

@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core.exception;
+
+import org.apache.fineract.command.core.Command;
+
+public class CommandIllegalArgumentException extends RuntimeException {
+
+    public CommandIllegalArgumentException(Command<?> command, String message) {
+        super("Illegal argument for " + command.getClass().getSimpleName() + " command: " + message);
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/implementation/AsynchronousCommandExecutor.java

@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.implementation;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.command.core.Command;
+import org.apache.fineract.command.core.CommandExecutor;
+import org.apache.fineract.command.core.CommandHandler;
+import org.apache.fineract.command.core.CommandMiddleware;
+import org.apache.fineract.command.core.CommandRouter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@RequiredArgsConstructor
+@Component
+@ConditionalOnProperty(value = "fineract.command.executor", havingValue = "async")
+public class AsynchronousCommandExecutor implements CommandExecutor {
+
+    private final List<CommandMiddleware> middlewares;
+
+    private final CommandRouter router;
+
+    @Override
+    public <REQ, RES> Supplier<RES> execute(Command<REQ> command) {
+        CompletableFuture<RES> future = CompletableFuture.supplyAsync(() -> {
+            for (CommandMiddleware middleware : middlewares) {
+                middleware.invoke(command);
+            }
+
+            CommandHandler<REQ, RES> handler = router.route(command);
+
+            return handler.handle(command);
+        });
+
+        return future::join;
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/implementation/DefaultCommandPipeline.java

@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.implementation;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.function.Supplier;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.command.core.Command;
+import org.apache.fineract.command.core.CommandExecutor;
+import org.apache.fineract.command.core.CommandPipeline;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@RequiredArgsConstructor
+@Component
+@ConditionalOnBean(CommandPipeline.class)
+public class DefaultCommandPipeline implements CommandPipeline {
+
+    private final CommandExecutor executor;
+
+    @Override
+    public <REQ, RES> Supplier<RES> send(final Command<REQ> command) {
+        requireNonNull(command, "Command must not be null");
+
+        return executor.execute(command);
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/implementation/DefaultCommandRouter.java

@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.implementation;
+
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.command.core.Command;
+import org.apache.fineract.command.core.CommandHandler;
+import org.apache.fineract.command.core.CommandRouter;
+import org.apache.fineract.command.core.exception.CommandHandlerNotFoundException;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@RequiredArgsConstructor
+@Component
+@ConditionalOnBean(CommandRouter.class)
+@SuppressWarnings({ "unchecked", "raw" })
+public class DefaultCommandRouter implements CommandRouter {
+
+    private final List<CommandHandler> commandHandlers;
+
+    @Override
+    public <REQ, RES> CommandHandler<REQ, RES> route(Command<REQ> command) {
+        // TODO: make sure there are no duplicate handlers
+        if (command == null) {
+            throw new CommandHandlerNotFoundException(command);
+        }
+
+        return commandHandlers.stream().filter(handler -> handler.matches(command)).findFirst()
+                .orElseThrow(() -> new CommandHandlerNotFoundException(command));
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/implementation/DisruptorCommandExecutor.java

@@ -0,0 +1,104 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.implementation;
+
+import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.dsl.Disruptor;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.command.core.Command;
+import org.apache.fineract.command.core.CommandExecutor;
+import org.apache.fineract.command.core.CommandMiddleware;
+import org.apache.fineract.command.core.CommandRouter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@RequiredArgsConstructor
+@Component
+@ConditionalOnProperty(value = "fineract.command.executor", havingValue = "disruptor")
+@SuppressWarnings({ "unchecked", "rawtypes" })
+public class DisruptorCommandExecutor implements CommandExecutor, Closeable {
+
+    private final Disruptor<CommandEvent> disruptor;
+
+    @Override
+    public <REQ, RES> Supplier<RES> execute(Command<REQ> command) {
+        CommandEvent<REQ, RES> processedEvent = next(command);
+
+        return processedEvent.getFuture()::join;
+    }
+
+    @Override
+    public void close() throws IOException {
+        disruptor.shutdown();
+    }
+
+    @SuppressWarnings({ "unchecked" })
+    private <REQ, RES> CommandEvent<REQ, RES> next(Command<REQ> command) {
+        var ringBuffer = disruptor.getRingBuffer();
+
+        var sequenceId = ringBuffer.next();
+
+        CommandEvent<REQ, RES> event = ringBuffer.get(sequenceId);
+        event.setCommand(command);
+        ringBuffer.publish(sequenceId);
+
+        return event;
+    }
+
+    @Getter
+    @Setter
+    public static class CommandEvent<REQ, RES> {
+
+        private Command<REQ> command;
+        private CompletableFuture<RES> future = new CompletableFuture<>();
+    }
+
+    @RequiredArgsConstructor
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public static class CompleteableCommandEventHandler implements EventHandler<CommandEvent> {
+
+        private final List<CommandMiddleware> middlewares;
+
+        private final CommandRouter router;
+
+        @Override
+        public void onEvent(CommandEvent event, long sequence, boolean endOfBatch) throws Exception {
+            try {
+                for (CommandMiddleware middleware : middlewares) {
+                    middleware.invoke(event.getCommand());
+                }
+
+                var handler = router.route(event.getCommand());
+
+                event.getFuture().complete(handler.handle(event.getCommand()));
+            } catch (Exception e) {
+                event.getFuture().completeExceptionally(e);
+            }
+        }
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/implementation/SynchronousCommandExecutor.java

@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.implementation;
+
+import java.util.List;
+import java.util.function.Supplier;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.command.core.Command;
+import org.apache.fineract.command.core.CommandExecutor;
+import org.apache.fineract.command.core.CommandHandler;
+import org.apache.fineract.command.core.CommandMiddleware;
+import org.apache.fineract.command.core.CommandRouter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@RequiredArgsConstructor
+@Component
+@ConditionalOnProperty(value = "fineract.command.executor", havingValue = "sync")
+@SuppressWarnings({ "unchecked" })
+public class SynchronousCommandExecutor implements CommandExecutor {
+
+    private final List<CommandMiddleware> middlewares;
+
+    private final CommandRouter router;
+
+    @Override
+    public <REQ, RES> Supplier<RES> execute(Command<REQ> command) {
+        for (CommandMiddleware middleware : middlewares) {
+            middleware.invoke(command);
+        }
+
+        CommandHandler<REQ, RES> handler = router.route(command);
+
+        return () -> handler.handle(command);
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/persistence/CommandRepository.java

@@ -0,0 +1,25 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.persistence;
+
+import org.apache.fineract.command.persistence.model.CommandEntity;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
+
+public interface CommandRepository extends JpaRepository<CommandEntity, Long>, JpaSpecificationExecutor<CommandEntity> {}

fineract-command/src/main/java/org/apache/fineract/command/persistence/converter/JsonAttributeConverter.java

@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.persistence.converter;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.persistence.AttributeConverter;
+import jakarta.persistence.Converter;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@RequiredArgsConstructor
+@Converter
+public class JsonAttributeConverter implements AttributeConverter<JsonNode, String> {
+    // TODO: it would be nicer to use a native JSON type on the database side, but not every system supports this;
+    // string/text are the lowest common denominator that should work on every database
+
+    private final ObjectMapper mapper;
+
+    @Override
+    @SneakyThrows
+    public String convertToDatabaseColumn(JsonNode source) {
+        if (source != null) {
+            return mapper.writeValueAsString(source);
+        }
+
+        // TODO: throw exception?
+        return null;
+    }
+
+    @Override
+    @SneakyThrows
+    public JsonNode convertToEntityAttribute(String source) {
+        if (source != null) {
+            return mapper.readTree(source);
+        }
+
+        // TODO: throw exception?
+        return null;
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/persistence/mapping/CommandJsonMapper.java

@@ -0,0 +1,64 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.persistence.mapping;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.node.TextNode;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@RequiredArgsConstructor
+@Component
+public final class CommandJsonMapper {
+
+    private static final String CLASS_ATTRIBUTE = "@class";
+    private final ObjectMapper mapper;
+
+    public <T> T map(JsonNode source) {
+        if (source == null) {
+            return null;
+        }
+
+        var canonicalName = source.get(CLASS_ATTRIBUTE).asText();
+
+        try {
+            return (T) mapper.convertValue(source, Class.forName(canonicalName));
+        } catch (Exception e) {
+            log.error("Error while mapping json node", e);
+        }
+
+        return null;
+    }
+
+    public JsonNode map(Object source) {
+        if (source == null) {
+            return null;
+        }
+
+        var json = mapper.convertValue(source, ObjectNode.class);
+
+        json.set(CLASS_ATTRIBUTE, new TextNode(source.getClass().getCanonicalName()));
+
+        return json;
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/persistence/mapping/CommandMapper.java

@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.persistence.mapping;
+
+import static org.mapstruct.InjectionStrategy.CONSTRUCTOR;
+import static org.mapstruct.MappingConstants.ComponentModel.SPRING;
+
+import org.apache.fineract.command.core.Command;
+import org.apache.fineract.command.persistence.model.CommandEntity;
+import org.mapstruct.InheritInverseConfiguration;
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+
+@Mapper(componentModel = SPRING, injectionStrategy = CONSTRUCTOR, uses = { CommandJsonMapper.class })
+public interface CommandMapper {
+
+    @Mapping(ignore = true, target = "id")
+    @Mapping(source = "id", target = "commandId")
+    @Mapping(source = "createdAt", target = "createdAt")
+    @Mapping(source = "tenantId", target = "tenantId")
+    @Mapping(source = "username", target = "username")
+    @Mapping(source = "payload", target = "payload")
+    CommandEntity map(Command source);
+
+    @InheritInverseConfiguration
+    Command map(CommandEntity source);
+}

fineract-command/src/main/java/org/apache/fineract/command/persistence/model/CommandEntity.java

@@ -0,0 +1,83 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.persistence.model;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import jakarta.persistence.Column;
+import jakarta.persistence.Convert;
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.PostLoad;
+import jakarta.persistence.PrePersist;
+import jakarta.persistence.Table;
+import jakarta.persistence.Transient;
+import java.io.Serial;
+import java.io.Serializable;
+import java.time.OffsetDateTime;
+import java.util.UUID;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+import lombok.experimental.FieldNameConstants;
+import org.apache.fineract.command.persistence.converter.JsonAttributeConverter;
+
+@Getter
+@Setter
+@ToString
+@FieldNameConstants
+@Entity
+@Table(name = "m_command")
+public class CommandEntity implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    @jakarta.persistence.Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    @Column(name = "id")
+    private Long id;
+
+    @Transient
+    @Setter(value = AccessLevel.NONE)
+    private boolean isNew = true;
+
+    @Column(name = "command_id")
+    private UUID commandId;
+
+    @Column(name = "created_at")
+    private OffsetDateTime createdAt;
+
+    @Column(name = "tenant_id")
+    private String tenantId;
+
+    @Column(name = "username")
+    private String username;
+
+    @Column(name = "payload")
+    @Convert(converter = JsonAttributeConverter.class)
+    private JsonNode payload;
+
+    @PrePersist
+    @PostLoad
+    void markNotNew() {
+        this.isNew = false;
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/starter/CommandAutoConfiguration.java

@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.starter;
+
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.context.annotation.Import;
+
+@AutoConfiguration
+@Import({ CommandConfiguration.class, CommandPersistenceConfiguration.class })
+public class CommandAutoConfiguration {}

fineract-command/src/main/java/org/apache/fineract/command/starter/CommandConfiguration.java

@@ -0,0 +1,66 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.starter;
+
+import com.lmax.disruptor.IgnoreExceptionHandler;
+import com.lmax.disruptor.WaitStrategy;
+import com.lmax.disruptor.YieldingWaitStrategy;
+import com.lmax.disruptor.dsl.Disruptor;
+import com.lmax.disruptor.util.DaemonThreadFactory;
+import java.util.List;
+import org.apache.fineract.command.core.CommandMiddleware;
+import org.apache.fineract.command.core.CommandProperties;
+import org.apache.fineract.command.core.CommandRouter;
+import org.apache.fineract.command.implementation.DisruptorCommandExecutor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@EnableConfigurationProperties(CommandProperties.class)
+@ComponentScan("org.apache.fineract.command.core")
+@ComponentScan("org.apache.fineract.command.implementation")
+class CommandConfiguration {
+
+    @Bean
+    @ConditionalOnMissingBean
+    WaitStrategy waitStrategy() {
+        return new YieldingWaitStrategy();
+    }
+
+    @Bean
+    Disruptor<?> disruptor(CommandProperties properties, WaitStrategy waitStrategy, List<CommandMiddleware> middlewares,
+            CommandRouter router) {
+        // TODO: make this more configurable
+
+        // Create the disruptor
+        Disruptor<DisruptorCommandExecutor.CommandEvent> disruptor = new Disruptor<>(DisruptorCommandExecutor.CommandEvent::new,
+                properties.getRingBufferSize(), DaemonThreadFactory.INSTANCE, properties.getProducerType(), waitStrategy);
+
+        disruptor.handleEventsWith(new DisruptorCommandExecutor.CompleteableCommandEventHandler(middlewares, router));
+        disruptor.setDefaultExceptionHandler(new IgnoreExceptionHandler());
+
+        // Start the disruptor
+        disruptor.start();
+
+        return disruptor;
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/starter/CommandPersistenceConfiguration.java

@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.starter;
+
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@ComponentScan("org.apache.fineract.command.persistence")
+class CommandPersistenceConfiguration {}

fineract-command/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+org.apache.fineract.command.starter.CommandAutoConfiguration

fineract-command/src/main/resources/db/custom-changelog/0001_command_init.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.1.xsd">
+    <changeSet author="fineract" id="command-00001" context="postgresql">
+        <createTable tableName="m_command">
+            <column autoIncrement="true" name="id" type="bigint" remarks="Internal ID">
+                <constraints nullable="false" primaryKey="true"/>
+            </column>
+            <column name="created_at" type="timestamp with time zone" />
+            <column name="command_id" type="uuid" />
+            <column name="tenant_id" type="varchar(255)" />
+            <column name="username" type="varchar(255)" />
+            <column name="payload" type="text" />
+        </createTable>
+    </changeSet>
+    <changeSet author="fineract" id="command-00002" context="mysql">
+        <createTable tableName="command">
+            <column autoIncrement="true" name="id" type="bigint" remarks="Internal ID">
+                <constraints nullable="false" primaryKey="true"/>
+            </column>
+            <column name="created_at" type="datetime" />
+            <column name="command_id" type="uuid" />
+            <column name="tenant_id" type="varchar(255)" />
+            <column name="username" type="varchar(255)" />
+            <column name="payload" type="text" />
+        </createTable>
+    </changeSet>
+</databaseChangeLog>