FINERACT-1724: Extended the event framework sinks to be ActiveMQ topics as well in addition to ActiveMQ queues

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/condition/EnableFineractEventListenerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.core.config;
+package org.apache.fineract.infrastructure.core.condition;
 
 import java.util.Optional;
 import org.apache.fineract.infrastructure.instancemode.api.FineractInstanceModeConstants;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/condition/EnableFineractEventsCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.core.config;
+package org.apache.fineract.infrastructure.core.condition;
 
 import java.util.Optional;
 import org.apache.fineract.infrastructure.instancemode.api.FineractInstanceModeConstants;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/condition/FineractModeValidationCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.core.config;
+package org.apache.fineract.infrastructure.core.condition;
 
 import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/condition/FineractPartitionJobConfigValidationCondition.java

@@ -16,11 +16,13 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.core.config;
+package org.apache.fineract.infrastructure.core.condition;
 
 import java.util.List;
 import java.util.function.Predicate;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.core.config.ExplicitConfigurationPropertiesFactory;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
 import org.springframework.context.annotation.Condition;
 import org.springframework.context.annotation.ConditionContext;
 import org.springframework.core.type.AnnotatedTypeMetadata;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/condition/FineractRemoteJobMessageHandlerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.core.config;
+package org.apache.fineract.infrastructure.core.condition;
 
 import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/condition/FineractValidationCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.core.config;
+package org.apache.fineract.infrastructure.core.condition;
 
 import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
 import org.springframework.context.annotation.Conditional;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/condition/PropertiesCondition.java

@@ -0,0 +1,35 @@
+/**
+ * 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.infrastructure.core.condition;
+
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+
+public abstract class PropertiesCondition implements Condition {
+
+    @Override
+    public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
+        FineractProperties properties = SpringPropertiesFactory.get(context.getEnvironment(), FineractProperties.class);
+        return matches(properties);
+    }
+
+    protected abstract boolean matches(FineractProperties properties);
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/condition/SpringPropertiesFactory.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.infrastructure.core.condition;
+
+import static org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.bind.Binder;
+import org.springframework.core.env.Environment;
+
+public abstract class SpringPropertiesFactory {
+
+    private SpringPropertiesFactory() {}
+
+    public static <T> T get(Environment environment, Class<T> clazz) {
+        ConfigurationProperties configurationProperties = clazz.getAnnotation(ConfigurationProperties.class);
+        if (configurationProperties == null) {
+            throw new IllegalArgumentException("Not a class with @ConfigurationProperties annotation");
+        }
+        String prefix = configurationProperties.prefix();
+        String value = configurationProperties.value();
+        if (isBlank(prefix) && isBlank(value)) {
+            throw new IllegalArgumentException("@ConfigurationProperties is missing both prefix and value properties");
+        }
+        String propertyName = isNotBlank(prefix) ? prefix : value;
+        return get(environment, propertyName, clazz);
+    }
+
+    public static <T> T get(Environment environment, String propertyName, Class<T> clazz) {
+        return Binder.get(environment).bind(propertyName, clazz).orElseThrow(
+                () -> new IllegalArgumentException("Couldn't bind " + clazz.getSimpleName() + " to the '" + propertyName + "' property"));
+    }
+}

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

@@ -22,6 +22,7 @@ package org.apache.fineract.infrastructure.core.config;
 import java.util.List;
 import lombok.Getter;
 import lombok.Setter;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
 @Getter
@@ -155,7 +156,14 @@ public class FineractProperties {
 
         private boolean enabled;
         private String eventQueueName;
+        private String eventTopicName;
         private String brokerUrl;
+        private String brokerUsername;
+        private String brokerPassword;
+
+        public boolean isBrokerPasswordProtected() {
+            return StringUtils.isNotBlank(brokerUsername) || StringUtils.isNotBlank(brokerPassword);
+        }
     }
 
     @Getter

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

@@ -20,6 +20,7 @@ package org.apache.fineract.infrastructure.core.config;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.core.condition.FineractValidationCondition;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/EnableExternalEventQueueCondition.java

@@ -0,0 +1,31 @@
+/**
+ * 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.infrastructure.event.external.config;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.fineract.infrastructure.core.condition.PropertiesCondition;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+
+public class EnableExternalEventQueueCondition extends PropertiesCondition {
+
+    @Override
+    protected boolean matches(FineractProperties properties) {
+        return StringUtils.isNotBlank(properties.getEvents().getExternal().getProducer().getJms().getEventQueueName());
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/EnableExternalEventTopicCondition.java

@@ -0,0 +1,31 @@
+/**
+ * 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.infrastructure.event.external.config;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.fineract.infrastructure.core.condition.PropertiesCondition;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+
+public class EnableExternalEventTopicCondition extends PropertiesCondition {
+
+    @Override
+    protected boolean matches(FineractProperties properties) {
+        return StringUtils.isNotBlank(properties.getEvents().getExternal().getProducer().getJms().getEventTopicName());
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventJMSBrokerConfiguration.java

@@ -19,10 +19,14 @@
 package org.apache.fineract.infrastructure.event.external.config;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ActiveMQTopic;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.core.config.FineractProperties.FineractExternalEventsProducerJmsProperties;
 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.Conditional;
 import org.springframework.context.annotation.Configuration;
 
 @Configuration
@@ -37,6 +41,23 @@ public class ExternalEventJMSBrokerConfiguration {
         ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
         connectionFactory.setBrokerURL(fineractProperties.getEvents().getExternal().getProducer().getJms().getBrokerUrl());
         connectionFactory.setTrustAllPackages(true);
+        FineractExternalEventsProducerJmsProperties jmsProps = fineractProperties.getEvents().getExternal().getProducer().getJms();
+        if (jmsProps.isBrokerPasswordProtected()) {
+            connectionFactory.setUserName(jmsProps.getBrokerUsername());
+            connectionFactory.setPassword(jmsProps.getBrokerPassword());
+        }
         return connectionFactory;
     }
+
+    @Conditional(EnableExternalEventTopicCondition.class)
+    @Bean(name = "eventDestination")
+    public ActiveMQTopic activeMqTopic() {
+        return new ActiveMQTopic(fineractProperties.getEvents().getExternal().getProducer().getJms().getEventTopicName());
+    }
+
+    @Conditional(EnableExternalEventQueueCondition.class)
+    @Bean(name = "eventDestination")
+    public ActiveMQQueue activeMqQueue() {
+        return new ActiveMQQueue(fineractProperties.getEvents().getExternal().getProducer().getJms().getEventQueueName());
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventJMSProducerConfiguration.java

@@ -18,9 +18,11 @@
  */
 package org.apache.fineract.infrastructure.event.external.config;
 
-import org.apache.activemq.ActiveMQConnectionFactory;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -29,7 +31,6 @@ import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
-import org.springframework.integration.handler.LoggingHandler;
 import org.springframework.integration.jms.dsl.Jms;
 
 @Configuration
@@ -45,12 +46,10 @@ public class ExternalEventJMSProducerConfiguration {
     private FineractProperties fineractProperties;
 
     @Bean
-    public IntegrationFlow outboundFlowEvents(ActiveMQConnectionFactory connectionFactory) {
+    public IntegrationFlow outboundFlowEvents(ConnectionFactory connectionFactory,
+            @Qualifier("eventDestination") Destination eventDestination) {
         return IntegrationFlows.from(outboundRequestsEvents) //
-                .log(LoggingHandler.Level.DEBUG) //
-                .handle(Jms.outboundAdapter(connectionFactory)
-                        .destination(fineractProperties.getEvents().getExternal().getProducer().getJms().getEventQueueName()))
-                .get();
+                .handle(Jms.outboundAdapter(connectionFactory).destination(eventDestination)).get();
     }
 
 }

fineract-provider/src/main/java/org/apache/fineract/notification/config/MessagingConfiguration.java

@@ -21,7 +21,7 @@ package org.apache.fineract.notification.config;
 import javax.jms.ExceptionListener;
 import javax.jms.JMSException;
 import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.fineract.infrastructure.core.config.EnableFineractEventsCondition;
+import org.apache.fineract.infrastructure.core.condition.EnableFineractEventsCondition;
 import org.apache.fineract.notification.eventandlistener.NotificationEventListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

fineract-provider/src/main/java/org/apache/fineract/notification/eventandlistener/ActiveMQNotificationEventListener.java

@@ -23,7 +23,7 @@ import javax.jms.Message;
 import javax.jms.ObjectMessage;
 import javax.jms.Session;
 import lombok.RequiredArgsConstructor;
-import org.apache.fineract.infrastructure.core.config.EnableFineractEventListenerCondition;
+import org.apache.fineract.infrastructure.core.condition.EnableFineractEventListenerCondition;
 import org.apache.fineract.notification.data.NotificationData;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Profile;

fineract-provider/src/main/java/org/apache/fineract/notification/eventandlistener/ActiveMQNotificationEventPublisher.java

@@ -21,7 +21,7 @@ package org.apache.fineract.notification.eventandlistener;
 import javax.jms.Queue;
 import lombok.RequiredArgsConstructor;
 import org.apache.activemq.command.ActiveMQQueue;
-import org.apache.fineract.infrastructure.core.config.EnableFineractEventsCondition;
+import org.apache.fineract.infrastructure.core.condition.EnableFineractEventsCondition;
 import org.apache.fineract.notification.data.NotificationData;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Profile;

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

@@ -56,8 +56,11 @@ fineract.remote-job-message-handler.jms.broker-url=${FINERACT_REMOTE_JOB_MESSAGE
 fineract.events.external.enabled=${FINERACT_EXTERNAL_EVENTS_ENABLED:false}
 fineract.events.external.producer.read-batch-size=${FINERACT_EXTERNAL_EVENTS_PRODUCER_READ_BATCH_SIZE:1000}
 fineract.events.external.producer.jms.enabled=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_ENABLED:false}
-fineract.events.external.producer.jms.event-queue-name=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_QUEUE_NAME:JMS-event-queue}
+fineract.events.external.producer.jms.event-queue-name=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_QUEUE_NAME:}
+fineract.events.external.producer.jms.event-topic-name=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_TOPIC_NAME:}
 fineract.events.external.producer.jms.broker-url=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_BROKER_URL:tcp://127.0.0.1:61616}
+fineract.events.external.producer.jms.broker-username=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_BROKER_USERNAME:}
+fineract.events.external.producer.jms.broker-password=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_BROKER_PASSWORD:}
 
 fineract.idempotency-key-header-name=${FINERACT_IDEMPOTENCY_KEY_HEADER_NAME:Idempotency-Key}
 

fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/config/EnableFineractEventListenerConditionTest.java

@@ -23,6 +23,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
 
+import org.apache.fineract.infrastructure.core.condition.EnableFineractEventListenerCondition;
 import org.apache.fineract.infrastructure.instancemode.api.FineractInstanceModeConstants;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;

fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/config/FineractPartitionJobConfigValidationConditionTest.java

@@ -24,6 +24,7 @@ import static org.mockito.Mockito.mock;
 
 import java.util.ArrayList;
 import java.util.List;
+import org.apache.fineract.infrastructure.core.condition.FineractPartitionJobConfigValidationCondition;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;