[FINERACT-1678] Spring Batch JMS support

This commit is contained in:
taskain7
2022-08-22 14:04:58 +02:00
committed by Arnold Gálovics
parent 02c19c963e
commit 340516c1a3
15 changed files with 398 additions and 41 deletions
+2
View File
@@ -36,6 +36,8 @@ dependencies {
'org.springframework.batch:spring-batch-integration',
'org.springframework.boot:spring-boot-starter-integration',
'org.springframework.integration:spring-integration-event',
'org.springframework.integration:spring-integration-jms',
'jakarta.ws.rs:jakarta.ws.rs-api',
'org.glassfish.jersey.media:jersey-media-multipart',
@@ -23,26 +23,12 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
import org.springframework.integration.handler.LoggingHandler;
@Configuration
@EnableBatchIntegration
@ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true")
public class COBManagerConfig {
@Bean
public IntegrationFlow outboundFlow() {
ApplicationEventPublishingMessageHandler handler = new ApplicationEventPublishingMessageHandler();
return IntegrationFlows.from(outboundRequests()) //
.intercept(outputInterceptor()) //
.log(LoggingHandler.Level.DEBUG) //
.handle(handler) //
.get(); //
}
@Bean
public DirectChannel outboundRequests() {
return new DirectChannel();
@@ -22,32 +22,11 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.event.core.MessagingEvent;
import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
import org.springframework.integration.handler.LoggingHandler;
@Configuration
@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
public class COBWorkerConfig {
@Bean
public IntegrationFlow inboundFlow() {
return IntegrationFlows.from(eventListener()) //
.log(LoggingHandler.Level.DEBUG) //
.channel(inboundRequests()) //
.intercept(inputInterceptor()) //
.get(); //
}
@Bean
public ApplicationEventListeningMessageProducer eventListener() {
ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer();
producer.setEventTypes(MessagingEvent.class);
return producer;
}
@Bean
public QueueChannel inboundRequests() {
return new QueueChannel();
@@ -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.cob.messagehandler;
import javax.jms.ConnectionFactory;
import org.apache.fineract.cob.COBOutputChannelInterceptor;
import org.apache.fineract.cob.messagehandler.conditions.JmsManagerCondition;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
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.integration.channel.DirectChannel;
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
@EnableBatchIntegration
@Conditional(JmsManagerCondition.class)
public class COBJmsManagerConfig {
@Autowired
private DirectChannel outboundRequests;
@Autowired
private COBOutputChannelInterceptor outputInterceptor;
@Autowired
private ConnectionFactory connectionFactory;
@Autowired
private FineractProperties fineractProperties;
@Bean
public IntegrationFlow outboundFlow() {
return IntegrationFlows.from(outboundRequests) //
.intercept(outputInterceptor) //
.log(LoggingHandler.Level.DEBUG) //
.handle(Jms.outboundGateway(connectionFactory)
.requestDestination(fineractProperties.getRemoteJobMessageHandler().getJms().getRequestQueueName()))
.get();
}
}
@@ -0,0 +1,56 @@
/**
* 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 javax.jms.ConnectionFactory;
import org.apache.fineract.cob.messagehandler.conditions.JmsWorkerCondition;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
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.integration.channel.QueueChannel;
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
@EnableBatchIntegration
@Conditional(JmsWorkerCondition.class)
public class COBJmsWorkerConfig {
@Autowired
private ConnectionFactory connectionFactory;
@Autowired
private QueueChannel inboundRequests;
@Autowired
private FineractProperties fineractProperties;
@Bean
public IntegrationFlow inboundFlow() {
return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(connectionFactory) //
.configureListenerContainer(c -> c.subscriptionDurable(true)) //
.destination(fineractProperties.getRemoteJobMessageHandler().getJms().getRequestQueueName())) //
.channel(inboundRequests) //
.log(LoggingHandler.Level.DEBUG) //
.get();
}
}
@@ -0,0 +1,53 @@
/**
* 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.fineract.cob.COBOutputChannelInterceptor;
import org.apache.fineract.cob.messagehandler.conditions.SpringEventManagerCondition;
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
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.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
import org.springframework.integration.handler.LoggingHandler;
@Configuration
@EnableBatchIntegration
@Conditional(SpringEventManagerCondition.class)
public class COBSpringEventManagerConfig {
@Autowired
private DirectChannel outboundRequests;
@Autowired
private COBOutputChannelInterceptor outputInterceptor;
@Bean
public IntegrationFlow outboundFlow() {
ApplicationEventPublishingMessageHandler handler = new ApplicationEventPublishingMessageHandler();
return IntegrationFlows.from(outboundRequests) //
.intercept(outputInterceptor) //
.log(LoggingHandler.Level.DEBUG) //
.handle(handler) //
.get(); //
}
}
@@ -0,0 +1,60 @@
/**
* 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.fineract.cob.COBInputChannelInterceptor;
import org.apache.fineract.cob.messagehandler.conditions.SpringEventWorkerCondition;
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
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.integration.channel.QueueChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.event.core.MessagingEvent;
import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
import org.springframework.integration.handler.LoggingHandler;
@Configuration
@EnableBatchIntegration
@Conditional(SpringEventWorkerCondition.class)
public class COBSpringEventWorkerConfig {
@Autowired
private QueueChannel inboundRequests;
@Autowired
private COBInputChannelInterceptor inputInterceptor;
@Bean
public IntegrationFlow inboundFlow() {
return IntegrationFlows.from(eventListener()) //
.log(LoggingHandler.Level.DEBUG) //
.channel(inboundRequests) //
.intercept(inputInterceptor) //
.get(); //
}
@Bean
public ApplicationEventListeningMessageProducer eventListener() {
ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer();
producer.setEventTypes(MessagingEvent.class);
return producer;
}
}
@@ -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.cob.messagehandler.conditions;
import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
public class JmsManagerCondition extends AllNestedConditions {
public JmsManagerCondition() {
super(ConfigurationPhase.PARSE_CONFIGURATION);
}
@ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true")
static class ManagerCondition {}
@ConditionalOnProperty(value = "fineract.remote-job-message-handler.jms.enabled", havingValue = "true")
static class JmsCondition {}
}
@@ -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.cob.messagehandler.conditions;
import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
public class JmsWorkerCondition extends AllNestedConditions {
public JmsWorkerCondition() {
super(ConfigurationPhase.PARSE_CONFIGURATION);
}
@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
static class WorkerCondition {}
@ConditionalOnProperty(value = "fineract.remote-job-message-handler.jms.enabled", havingValue = "true")
static class JmsCondition {}
}
@@ -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.cob.messagehandler.conditions;
import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
public class SpringEventManagerCondition extends AllNestedConditions {
public SpringEventManagerCondition() {
super(ConfigurationPhase.PARSE_CONFIGURATION);
}
@ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true")
static class ManagerCondition {}
@ConditionalOnProperty(value = "fineract.remote-job-message-handler.spring-events.enabled", havingValue = "true")
static class SpringEventCondition {}
}
@@ -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.cob.messagehandler.conditions;
import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
public class SpringEventWorkerCondition extends AllNestedConditions {
public SpringEventWorkerCondition() {
super(ConfigurationPhase.PARSE_CONFIGURATION);
}
@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
static class WorkerCondition {}
@ConditionalOnProperty(value = "fineract.remote-job-message-handler.spring-events.enabled", havingValue = "true")
static class SpringEventCondition {}
}
@@ -101,6 +101,7 @@ public class FineractProperties {
public static class FineractRemoteJobMessageHandlerProperties {
private FineractRemoteJobMessageHandlerSpringEventsProperties springEvents;
private FineractRemoteJobMessageHandlerJmsProperties jms;
}
@Getter
@@ -109,4 +110,12 @@ public class FineractProperties {
private boolean enabled;
}
@Getter
@Setter
public static class FineractRemoteJobMessageHandlerJmsProperties {
private boolean enabled;
private String requestQueueName;
}
}
@@ -33,19 +33,26 @@ public class FineractRemoteJobMessageHandlerCondition implements Condition {
.ofNullable(
context.getEnvironment().getProperty("fineract.remote-job-message-handler.spring-events.enabled", Boolean.class))
.orElse(true);
boolean isJmsEnabled = Optional
.ofNullable(context.getEnvironment().getProperty("fineract.remote-job-message-handler.jms.enabled", Boolean.class))
.orElse(true);
boolean isBatchManagerModeEnabled = Optional
.ofNullable(context.getEnvironment().getProperty("fineract.mode.batch-manager-enabled", Boolean.class)).orElse(true);
boolean isBatchWorkerModeEnabled = Optional
.ofNullable(context.getEnvironment().getProperty("fineract.mode.batch-worker-enabled", Boolean.class)).orElse(true);
// TODO extend this expression with other message handlers in the future
boolean isMessageHandlerCountFails = !isSpringEventsEnabled;
boolean isSpringEventValidationFails = !(isSpringEventsEnabled && isBatchManagerModeEnabled && isBatchWorkerModeEnabled);
if (isMessageHandlerCountFails) {
boolean conditionFails = false;
boolean onlyOneMessageHandlerEnabled = isSpringEventsEnabled ^ isJmsEnabled;
if (!onlyOneMessageHandlerEnabled) {
conditionFails = true;
log.error("For remote partitioning jobs exactly one Message Handler must be enabled.");
}
if (isSpringEventValidationFails) {
log.error("If Spring Event Message Handler is enabled, the instance must be Batch Manager and Batch Worker too.");
if (isSpringEventsEnabled) {
if (!(isBatchManagerModeEnabled && isBatchWorkerModeEnabled)) {
conditionFails = true;
log.error("If Spring Event Message Handler is enabled, the instance must be Batch Manager and Batch Worker too.");
}
}
return isMessageHandlerCountFails || isSpringEventValidationFails;
return conditionFails;
}
}
@@ -49,6 +49,8 @@ fineract.partitioned-job.partitioned-job-properties[0].partition-size=${LOAN_COB
fineract.partitioned-job.partitioned-job-properties[0].thread-count=${LOAN_COB_THREAD_COUNT:1}
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}
# 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}}
@@ -42,6 +42,10 @@ fineract.partitioned-job.partitioned-job-properties[0].chunk-size=100
fineract.partitioned-job.partitioned-job-properties[0].partition-size=100
fineract.partitioned-job.partitioned-job-properties[0].thread-count=1
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}
management.health.jms.enabled=false
# FINERACT 1296