[FINERACT-1678] Add ActiveMQ configuration

fineract-provider/src/main/java/org/apache/fineract/cob/messagehandler/COBJmsManagerConfig.java

@@ -18,7 +18,7 @@
  */
 package org.apache.fineract.cob.messagehandler;
 
-import javax.jms.ConnectionFactory;
+import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.fineract.cob.COBOutputChannelInterceptor;
 import org.apache.fineract.cob.messagehandler.conditions.JmsManagerCondition;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
@@ -27,6 +27,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
@@ -36,6 +37,7 @@ import org.springframework.integration.jms.dsl.Jms;
 @Configuration
 @EnableBatchIntegration
 @Conditional(JmsManagerCondition.class)
+@Import(value = { JmsBrokerConfiguration.class })
 public class COBJmsManagerConfig {
 
     @Autowired
@@ -43,17 +45,15 @@ public class COBJmsManagerConfig {
     @Autowired
     private COBOutputChannelInterceptor outputInterceptor;
     @Autowired
-    private ConnectionFactory connectionFactory;
-    @Autowired
     private FineractProperties fineractProperties;
 
     @Bean
-    public IntegrationFlow outboundFlow() {
+    public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
         return IntegrationFlows.from(outboundRequests) //
                 .intercept(outputInterceptor) //
                 .log(LoggingHandler.Level.DEBUG) //
-                .handle(Jms.outboundGateway(connectionFactory)
+                .handle(Jms.outboundAdapter(connectionFactory)
-                        .requestDestination(fineractProperties.getRemoteJobMessageHandler().getJms().getRequestQueueName()))
+                        .destination(fineractProperties.getRemoteJobMessageHandler().getJms().getRequestQueueName()))
                 .get();
     }
 }

fineract-provider/src/main/java/org/apache/fineract/cob/messagehandler/COBJmsWorkerConfig.java

@@ -18,7 +18,8 @@
  */
 package org.apache.fineract.cob.messagehandler;
 
-import javax.jms.ConnectionFactory;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.fineract.cob.COBInputChannelInterceptor;
 import org.apache.fineract.cob.messagehandler.conditions.JmsWorkerCondition;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
@@ -26,6 +27,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
@@ -35,22 +37,23 @@ import org.springframework.integration.jms.dsl.Jms;
 @Configuration
 @EnableBatchIntegration
 @Conditional(JmsWorkerCondition.class)
+@Import(value = { JmsBrokerConfiguration.class })
 public class COBJmsWorkerConfig {
 
-    @Autowired
-    private ConnectionFactory connectionFactory;
     @Autowired
     private QueueChannel inboundRequests;
     @Autowired
+    private COBInputChannelInterceptor inputInterceptor;
+    @Autowired
     private FineractProperties fineractProperties;
 
     @Bean
-    public IntegrationFlow inboundFlow() {
+    public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
         return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(connectionFactory) //
-                .configureListenerContainer(c -> c.subscriptionDurable(true)) //
+                .configureListenerContainer(c -> c.subscriptionDurable(false)) //
                 .destination(fineractProperties.getRemoteJobMessageHandler().getJms().getRequestQueueName())) //
                 .channel(inboundRequests) //
-                .log(LoggingHandler.Level.DEBUG) //
+                .intercept(inputInterceptor).log(LoggingHandler.Level.DEBUG) //
                 .get();
     }
 }

fineract-provider/src/main/java/org/apache/fineract/cob/messagehandler/JmsBrokerConfiguration.java

@@ -0,0 +1,42 @@
+/**
+ * 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.cob.messagehandler;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@ConditionalOnProperty(value = "fineract.remote-job-message-handler.jms.enabled", havingValue = "true")
+public class JmsBrokerConfiguration {
+
+    @Autowired
+    private FineractProperties fineractProperties;
+
+    @Bean
+    public ActiveMQConnectionFactory connectionFactory() {
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
+        connectionFactory.setBrokerURL(fineractProperties.getRemoteJobMessageHandler().getJms().getBrokerUrl());
+        connectionFactory.setTrustAllPackages(true);
+        return connectionFactory;
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java

@@ -117,5 +117,6 @@ public class FineractProperties {
 
         private boolean enabled;
         private String requestQueueName;
+        private String brokerUrl;
     }
 }

fineract-provider/src/main/resources/application.properties

@@ -51,6 +51,7 @@ fineract.partitioned-job.partitioned-job-properties[0].thread-count=${LOAN_COB_T
 fineract.remote-job-message-handler.spring-events.enabled=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_SPRING_EVENTS_ENABLED:true}
 fineract.remote-job-message-handler.jms.enabled=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_ENABLED:false}
 fineract.remote-job-message-handler.jms.request-queue-name=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_QUEUE_NAME:JMS-request-queue}
+fineract.remote-job-message-handler.jms.broker-url=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_BROKER_URL:tcp://127.0.0.1:61616}
 
 # Logging pattern for the console
 logging.pattern.console=${CONSOLE_LOG_PATTERN:%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(%replace([%X{correlationId}]){'\\[\\]', ''}) %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:%wEx}}