From 002382e6477293d67e683eb8d2d6ae046613d6ce Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 6 Nov 2020 13:51:03 -0500 Subject: [PATCH] Rely on `MProducerSupport.active` for `Flux` (#3423) * Rely on `MProducerSupport.active` for `Flux` * Fix `MessageProducerSupport` to extract an `active` flag and set it before `isRunning` - the `Flux` subscription relies on the `takeWhile()` where in case of `autoStartup = false` we will never start consume because it is set to `true` already after `doStart()` * Refactor all the `MessageProducerSupport` implementation with similar `active` state to use already one from the super class **Cherry-pick to 5.3.x** * * Remove `MessageProducerSupport.setActive()` to not let to mutate it from the implementations * Set `active` to `false` in the `destroy()` * Clean up and fix typos in the affected `JmsMessageDrivenEndpoint` * * Pull `active` flag down to the `AbstractEndpoint` * Set `active = true` in the `start()` before calling `doStart()` * Do same for `active = false` in the `stop()` * Clean up `AbstractEndpoint` impls to not call `doStart/doStop` for nothing * Refactor endpoints to rely on the `active` state from the `AbstractEndpoint` not their own --- .../endpoint/AbstractEndpoint.java | 26 +- .../endpoint/AbstractPollingEndpoint.java | 2 +- .../endpoint/MessageProducerSupport.java | 2 +- .../gateway/GatewayProxyFactoryBean.java | 6 + .../ReactiveMessageProducerTests.java | 23 +- ...licationEventListeningMessageProducer.java | 33 +- ...acheCommonsFileTailingMessageProducer.java | 28 +- .../FileTailingMessageProducerSupport.java | 2 - ...ternetProtocolReceivingChannelAdapter.java | 84 ++-- .../integration/ip/tcp/TcpInboundGateway.java | 57 ++- .../ip/tcp/TcpReceivingChannelAdapter.java | 77 ++-- .../integration/jms/JmsInboundGateway.java | 9 +- .../jms/JmsMessageDrivenEndpoint.java | 28 +- .../kafka/inbound/KafkaInboundGateway.java | 401 ++++++++++++++++++ .../MqttPahoMessageDrivenChannelAdapter.java | 2 - .../ReactiveRedisStreamMessageProducer.java | 248 +++++++++++ .../inbound/RedisInboundChannelAdapter.java | 4 +- .../inbound/RedisQueueInboundGateway.java | 18 +- .../RedisQueueMessageDrivenEndpoint.java | 18 +- .../TcpSyslogReceivingChannelAdapter.java | 7 +- .../UdpSyslogReceivingChannelAdapter.java | 11 +- .../WebSocketInboundChannelAdapter.java | 17 +- .../zeromq/inbound/ZeroMqMessageProducer.java | 317 ++++++++++++++ 23 files changed, 1167 insertions(+), 253 deletions(-) create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index 5bdde86d77..ace88bcc05 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -47,14 +47,6 @@ import org.springframework.util.StringUtils; public abstract class AbstractEndpoint extends IntegrationObjectSupport implements SmartLifecycle, DisposableBean { - private boolean autoStartupSetExplicitly; - - private volatile boolean autoStartup = true; - - private volatile int phase = 0; - - private volatile boolean running; - protected final ReentrantLock lifecycleLock = new ReentrantLock(); // NOSONAR protected final Condition lifecycleCondition = this.lifecycleLock.newCondition(); // NOSONAR @@ -63,6 +55,16 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport private SmartLifecycleRoleController roleController; + private boolean autoStartup = true; + + private boolean autoStartupSetExplicitly; + + private int phase = 0; + + private volatile boolean running; + + private volatile boolean active; + public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; this.autoStartupSetExplicitly = true; @@ -120,6 +122,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport @Override public void destroy() { + stop(); if (this.roleController != null) { this.roleController.removeLifecycle(this); } @@ -153,6 +156,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport this.lifecycleLock.lock(); try { if (!this.running) { + this.active = true; doStart(); this.running = true; if (logger.isInfoEnabled()) { @@ -170,6 +174,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport this.lifecycleLock.lock(); try { if (this.running) { + this.active = false; doStop(); this.running = false; if (logger.isInfoEnabled()) { @@ -187,6 +192,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport this.lifecycleLock.lock(); try { if (this.running) { + this.active = false; doStop(callback); this.running = false; if (logger.isInfoEnabled()) { @@ -211,6 +217,10 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport callback.run(); } + public boolean isActive() { + return this.active; + } + /** * Subclasses must implement this method with the start behavior. * This method will be invoked while holding the {@link #lifecycleLock}. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 305b48c218..26411642c3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -368,7 +368,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement triggerContext.lastActualExecutionTime(), new Date()) )), 1) - .repeat(this::isRunning) + .repeat(this::isActive) .doOnSubscribe(subs -> this.subscription = subs); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 90d4c155d7..481c652c65 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -222,7 +222,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements .map(this::trackMessageIfAny) .doOnComplete(this::stop) .doOnCancel(this::stop) - .takeWhile((message) -> isRunning()); + .takeWhile((message) -> isActive()); if (channelForSubscription instanceof ReactiveStreamsSubscribableChannel) { ((ReactiveStreamsSubscribableChannel) channelForSubscription).subscribeTo(messageFlux); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index ef5991fde0..792bb30d87 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -1016,6 +1016,12 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } } + @Override + public void destroy() { + super.destroy(); + this.gatewayMap.values().forEach(MethodInvocationGateway::destroy); + } + private static final class MethodInvocationGateway extends MessagingGatewaySupport { private Expression receiveTimeoutExpression; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveMessageProducerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveMessageProducerTests.java index fff06a4c5d..f9d99f28ab 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveMessageProducerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveMessageProducerTests.java @@ -16,7 +16,7 @@ package org.springframework.integration.endpoint; -import static org.assertj.core.api.Assertions.assertThat; +import java.time.Duration; import org.junit.jupiter.api.Test; @@ -50,17 +50,18 @@ public class ReactiveMessageProducerTests { @Test public void test() { - assertThat(this.producer.isRunning()).isTrue(); + StepVerifier stepVerifier = + StepVerifier.create( + Flux.from(this.fluxMessageChannel) + .map(Message::getPayload) + .cast(String.class)) + .expectNext("test1", "test2") + .thenCancel() + .verifyLater(); - StepVerifier.create( - Flux.from(this.fluxMessageChannel) - .map(Message::getPayload) - .cast(String.class)) - .expectNext("test1", "test2") - .thenCancel() - .verify(); + this.producer.start(); - assertThat(this.producer.isRunning()).isFalse(); + stepVerifier.verify(Duration.ofSeconds(10)); } @Configuration @@ -79,10 +80,12 @@ public class ReactiveMessageProducerTests { @Override protected void doStart() { + super.doStart(); subscribeToPublisher(Flux.just("test1", "test2").map(GenericMessage::new)); } }; + producer.setAutoStartup(false); producer.setOutputChannel(fluxMessageChannel()); return producer; } diff --git a/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java b/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java index 55c512bad0..a25b1067f8 100644 --- a/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java +++ b/spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -41,6 +41,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Artem Bilan * @author Gary Russell + * * @see ApplicationEventMulticaster * @see ExpressionMessageProducerSupport */ @@ -51,8 +52,6 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP private ApplicationEventMulticaster applicationEventMulticaster; - private volatile boolean active; - private volatile long stoppedAt; public ApplicationEventListeningMessageProducer() { @@ -66,14 +65,13 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP * In addition, this method re-registers the current instance as a {@link GenericApplicationListener} * with the {@link ApplicationEventMulticaster} which clears the listener cache. The cache will be * refreshed on the next appropriate {@link ApplicationEvent}. - * * @param eventTypes The event types. * @see ApplicationEventMulticaster#addApplicationListener * @see #supportsEventType */ public final void setEventTypes(Class... eventTypes) { Assert.notNull(eventTypes, "'eventTypes' must not be null"); - Set eventSet = new HashSet(eventTypes.length); + Set eventSet = new HashSet<>(eventTypes.length); for (Class eventType : eventTypes) { if (eventType != null) { eventSet.add(ResolvableType.forClass(eventType)); @@ -94,7 +92,7 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP @Override protected void onInit() { super.onInit(); - this.applicationEventMulticaster = this.getBeanFactory() + this.applicationEventMulticaster = getBeanFactory() .getBean(AbstractApplicationContext.APPLICATION_EVENT_MULTICASTER_BEAN_NAME, ApplicationEventMulticaster.class); Assert.notNull(this.applicationEventMulticaster, @@ -104,21 +102,23 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP @Override public void onApplicationEvent(ApplicationEvent event) { - if (this.active || ((event instanceof ContextStoppedEvent || event instanceof ContextClosedEvent) - && this.stoppedRecently())) { - if (event.getSource() instanceof Message) { - this.sendMessage((Message) event.getSource()); + if (isActive() || ((event instanceof ContextStoppedEvent || event instanceof ContextClosedEvent) + && stoppedRecently())) { + + Object source = event.getSource(); + if (source instanceof Message) { + sendMessage((Message) source); } else { - Message message = null; + Message message; Object result = extractObjectToSend(event); if (result instanceof Message) { message = (Message) result; } else { - message = this.getMessageBuilderFactory().withPayload(result).build(); + message = getMessageBuilderFactory().withPayload(result).build(); } - this.sendMessage(message); + sendMessage(message); } } } @@ -175,15 +175,10 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP return Ordered.LOWEST_PRECEDENCE; } - @Override - protected void doStart() { - this.active = true; - } - @Override protected void doStop() { this.stoppedAt = System.currentTimeMillis(); - this.active = false; + super.doStop(); } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java index df894cd912..da5b374ca2 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,20 +23,22 @@ import org.apache.commons.io.input.TailerListener; * File tailer that delegates to the Apache Commons Tailer. * * @author Gary Russell + * @author Artem Bilan + * * @since 3.0 * */ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageProducerSupport implements TailerListener { + private long pollingDelay = 1000; + + private boolean end = true; + + private boolean reopen = false; + private volatile Tailer tailer; - private volatile long pollingDelay = 1000; - - private volatile boolean end = true; - - private volatile boolean reopen = false; - /** * The delay between checks of the file for new content in milliseconds. * @param pollingDelay The delay. @@ -71,8 +73,8 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP @Override protected void doStart() { super.doStart(); - Tailer theTailer = new Tailer(this.getFile(), this, this.pollingDelay, this.end, this.reopen); - this.getTaskExecutor().execute(theTailer); + Tailer theTailer = new Tailer(getFile(), this, this.pollingDelay, this.end, this.reopen); + getTaskExecutor().execute(theTailer); this.tailer = theTailer; } @@ -88,9 +90,9 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP @Override public void fileNotFound() { - this.publish("File not found: " + this.getFile().getAbsolutePath()); + publish("File not found: " + getFile().getAbsolutePath()); try { - Thread.sleep(this.getMissingFileDelay()); + Thread.sleep(getMissingFileDelay()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -99,7 +101,7 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP @Override public void fileRotated() { - this.publish("File rotated: " + this.getFile().getAbsolutePath()); + publish("File rotated: " + getFile().getAbsolutePath()); } @Override @@ -109,7 +111,7 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP @Override public void handle(Exception ex) { - this.publish(ex.getMessage()); + publish(ex.getMessage()); } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java index 5e23124cb6..cf10b70248 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java @@ -145,7 +145,6 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS @Override protected void doStart() { - super.doStart(); if (this.idleEventInterval > 0) { this.idleEventScheduledFuture = getTaskScheduler().scheduleWithFixedDelay(() -> { long now = System.currentTimeMillis(); @@ -162,7 +161,6 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS @Override protected void doStop() { - super.doStop(); if (this.idleEventScheduledFuture != null) { this.idleEventScheduledFuture.cancel(true); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java index 9bde742687..b4af6c5ea1 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,7 +19,6 @@ package org.springframework.integration.ip; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.ThreadFactory; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; @@ -32,6 +31,8 @@ import org.springframework.util.Assert; * * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public abstract class AbstractInternetProtocolReceivingChannelAdapter @@ -56,8 +57,6 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter private int poolSize = 5; - private volatile boolean active; - private volatile boolean listening; @@ -108,48 +107,6 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter return this.receiveBufferSize; } - /** - * Protected by lifecycleLock - */ - @Override - protected void doStart() { - if (!this.active) { - this.active = true; - String beanName = this.getComponentName(); - checkTaskExecutor((beanName == null ? "" : beanName + "-") + this.getComponentType()); - this.taskExecutor.execute(this); - } - } - - /** - * Creates a default task executor if none was supplied. - * - * @param threadName The thread name. - */ - protected void checkTaskExecutor(final String threadName) { - if (this.active && this.taskExecutor == null) { - Executor executor = Executors.newFixedThreadPool(this.poolSize, new ThreadFactory() { - @Override - public Thread newThread(Runnable runner) { - Thread thread = new Thread(runner); - thread.setName(threadName); - thread.setDaemon(true); - return thread; - } - }); - this.taskExecutor = executor; - } - } - - @Override - protected void doStop() { - this.active = false; - if (!this.taskExecutorSet && this.taskExecutor != null) { - ((ExecutorService) this.taskExecutor).shutdown(); - this.taskExecutor = null; - } - } - public boolean isListening() { return this.listening; } @@ -197,10 +154,39 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter } /** - * @return the active + * Protected by lifecycleLock */ - public boolean isActive() { - return this.active; + @Override + protected void doStart() { + String beanName = getComponentName(); + checkTaskExecutor((beanName == null ? "" : beanName + "-") + getComponentType()); + this.taskExecutor.execute(this); + } + + /** + * Creates a default task executor if none was supplied. + * + * @param threadName The thread name. + */ + protected void checkTaskExecutor(final String threadName) { + if (isActive() && this.taskExecutor == null) { + this.taskExecutor = + Executors.newFixedThreadPool(this.poolSize, + (runner) -> { + Thread thread = new Thread(runner); + thread.setName(threadName); + thread.setDaemon(true); + return thread; + }); + } + } + + @Override + protected void doStop() { + if (!this.taskExecutorSet && this.taskExecutor != null) { + ((ExecutorService) this.taskExecutor).shutdown(); + this.taskExecutor = null; + } } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java index 466729aae8..341c9164cd 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java @@ -75,8 +75,6 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements private long retryInterval = DEFAULT_RETRY_INTERVAL; - private volatile boolean active; - private volatile ClientModeConnectionManager clientModeConnectionManager; private volatile ScheduledFuture scheduledFuture; @@ -210,8 +208,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements protected void onInit() { super.onInit(); if (this.isClientMode) { - Assert.notNull(this.clientConnectionFactory, - "For client-mode, connection factory must be type='client'"); + Assert.notNull(this.clientConnectionFactory, "For client-mode, connection factory must be type='client'"); Assert.isTrue(!this.clientConnectionFactory.isSingleUse(), "For client-mode, connection factory must have single-use='false'"); } @@ -220,40 +217,34 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements @Override // protected by super#lifecycleLock protected void doStart() { super.doStart(); - if (!this.active) { - this.active = true; - this.shuttingDown = false; - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.start(); - } - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.start(); - } - if (this.isClientMode) { - ClientModeConnectionManager manager = - new ClientModeConnectionManager(this.clientConnectionFactory); - this.clientModeConnectionManager = manager; - Assert.state(getTaskScheduler() != null, "Client mode requires a task scheduler"); - this.scheduledFuture = getTaskScheduler().scheduleAtFixedRate(manager, this.retryInterval); - } + this.shuttingDown = false; + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.start(); + } + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.start(); + } + if (this.isClientMode) { + ClientModeConnectionManager manager = + new ClientModeConnectionManager(this.clientConnectionFactory); + this.clientModeConnectionManager = manager; + Assert.state(getTaskScheduler() != null, "Client mode requires a task scheduler"); + this.scheduledFuture = getTaskScheduler().scheduleAtFixedRate(manager, this.retryInterval); } } @Override // protected by super#lifecycleLock protected void doStop() { super.doStop(); - if (this.active) { - this.active = false; - if (this.scheduledFuture != null) { - this.scheduledFuture.cancel(true); - } - this.clientModeConnectionManager = null; - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.stop(); - } - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.stop(); - } + if (this.scheduledFuture != null) { + this.scheduledFuture.cancel(true); + } + this.clientModeConnectionManager = null; + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.stop(); + } + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.stop(); } } @@ -301,7 +292,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements @Override public void retryConnection() { - if (this.active && this.isClientMode && this.clientModeConnectionManager != null) { + if (isActive() && this.isClientMode && this.clientModeConnectionManager != null) { this.clientModeConnectionManager.run(); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java index e8306439ae..843fc3f8ae 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -31,6 +31,7 @@ import org.springframework.integration.ip.tcp.connection.ConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpListener; import org.springframework.messaging.Message; import org.springframework.messaging.support.ErrorMessage; +import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; /** @@ -40,11 +41,13 @@ import org.springframework.util.Assert; * a client factory, the sender owns the connection. * * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 * */ public class TcpReceivingChannelAdapter - extends MessageProducerSupport implements TcpListener, ClientModeCapable, OrderlyShutdownCapable { + extends MessageProducerSupport implements TcpListener, ClientModeCapable, OrderlyShutdownCapable { private AbstractConnectionFactory clientConnectionFactory; @@ -60,8 +63,6 @@ public class TcpReceivingChannelAdapter private volatile ClientModeConnectionManager clientModeConnectionManager; - private volatile boolean active; - private volatile boolean shuttingDown; private final AtomicInteger activeCount = new AtomicInteger(); @@ -71,9 +72,7 @@ public class TcpReceivingChannelAdapter boolean isErrorMessage = message instanceof ErrorMessage; try { if (this.shuttingDown) { - if (logger.isInfoEnabled()) { - logger.info("Inbound message ignored; shutting down; " + message.toString()); - } + logger.info(() -> "Inbound message ignored; shutting down; " + message.toString()); } else { if (isErrorMessage) { @@ -123,41 +122,33 @@ public class TcpReceivingChannelAdapter @Override // protected by super#lifecycleLock protected void doStart() { - super.doStart(); - if (!this.active) { - this.active = true; - this.shuttingDown = false; - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.start(); - } - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.start(); - } - if (this.isClientMode) { - ClientModeConnectionManager manager = new ClientModeConnectionManager( - this.clientConnectionFactory); - this.clientModeConnectionManager = manager; - Assert.state(this.getTaskScheduler() != null, "Client mode requires a task scheduler"); - this.scheduledFuture = this.getTaskScheduler().scheduleAtFixedRate(manager, this.retryInterval); - } + this.shuttingDown = false; + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.start(); + } + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.start(); + } + if (this.isClientMode) { + ClientModeConnectionManager manager = new ClientModeConnectionManager(this.clientConnectionFactory); + this.clientModeConnectionManager = manager; + TaskScheduler taskScheduler = getTaskScheduler(); + Assert.state(taskScheduler != null, "Client mode requires a task scheduler"); + this.scheduledFuture = taskScheduler.scheduleAtFixedRate(manager, this.retryInterval); } } @Override // protected by super#lifecycleLock protected void doStop() { - super.doStop(); - if (this.active) { - this.active = false; - if (this.scheduledFuture != null) { - this.scheduledFuture.cancel(true); - } - this.clientModeConnectionManager = null; - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.stop(); - } - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.stop(); - } + if (this.scheduledFuture != null) { + this.scheduledFuture.cancel(true); + } + this.clientModeConnectionManager = null; + if (this.clientConnectionFactory != null) { + this.clientConnectionFactory.stop(); + } + if (this.serverConnectionFactory != null) { + this.serverConnectionFactory.stop(); } } @@ -165,7 +156,6 @@ public class TcpReceivingChannelAdapter * Sets the client or server connection factory; for this (an inbound adapter), if * the factory is a client connection factory, the sockets are owned by a sending * channel adapter and this adapter is used to receive replies. - * * @param connectionFactory the connectionFactory to set */ public void setConnectionFactory(AbstractConnectionFactory connectionFactory) { @@ -217,8 +207,7 @@ public class TcpReceivingChannelAdapter } /** - * @param isClientMode - * the isClientMode to set + * @param isClientMode the isClientMode to set */ public void setClientMode(boolean isClientMode) { this.isClientMode = isClientMode; @@ -232,8 +221,7 @@ public class TcpReceivingChannelAdapter } /** - * @param retryInterval - * the retryInterval to set + * @param retryInterval the retryInterval to set */ public void setRetryInterval(long retryInterval) { this.retryInterval = retryInterval; @@ -251,7 +239,7 @@ public class TcpReceivingChannelAdapter @Override public void retryConnection() { - if (this.active && this.isClientMode && this.clientModeConnectionManager != null) { + if (isActive() && this.isClientMode && this.clientModeConnectionManager != null) { this.clientModeConnectionManager.run(); } } @@ -264,7 +252,8 @@ public class TcpReceivingChannelAdapter @Override public int afterShutdown() { - this.stop(); + stop(); return this.activeCount.get(); } + } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java index 1117550e7b..2ee091d000 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -159,12 +159,7 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements Orderl @Override public void destroy() { this.endpoint.destroy(); - try { - super.destroy(); - } - catch (Exception e) { - throw new IllegalStateException(e); - } + super.destroy(); } @Override diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java index 3253268304..b2bf2027dd 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -56,6 +56,7 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements */ public JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer, ChannelPublishingJmsMessageListener listener) { + this(listenerContainer, listener, true); } @@ -65,11 +66,12 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements * 'transacted'. * @param listenerContainer the container. * @param listener the listener. - * @param externalContainer true if the container is externally configured and should not have its ackmode + * @param externalContainer true if the container is externally configured and should not have its ack mode * coerced when no sessionAcknowledgeMode was supplied. */ private JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer, ChannelPublishingJmsMessageListener listener, boolean externalContainer) { + Assert.notNull(listenerContainer, "listener container must not be null"); Assert.notNull(listener, "listener must not be null"); if (listenerContainer.getMessageListener() != null) { @@ -171,12 +173,12 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements if (!this.listenerContainer.isActive()) { this.listenerContainer.afterPropertiesSet(); } - String sessionAckeMode = this.sessionAcknowledgeMode; - if (sessionAckeMode == null && !this.externalContainer + String sessionAckMode = this.sessionAcknowledgeMode; + if (sessionAckMode == null && !this.externalContainer && DefaultMessageListenerContainer.class.isAssignableFrom(this.listenerContainer.getClass())) { - sessionAckeMode = JmsAdapterUtils.SESSION_TRANSACTED_STRING; + sessionAckMode = JmsAdapterUtils.SESSION_TRANSACTED_STRING; } - Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(sessionAckeMode); + Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(sessionAckMode); if (acknowledgeMode != null) { if (JmsAdapterUtils.SESSION_TRANSACTED == acknowledgeMode) { this.listenerContainer.setSessionTransacted(true); @@ -185,7 +187,7 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements this.listenerContainer.setSessionAcknowledgeMode(acknowledgeMode); } } - this.listener.setComponentName(this.getComponentName()); + this.listener.setComponentName(getComponentName()); } @Override @@ -212,21 +214,13 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements @Override public void destroy() { - if (this.isRunning()) { - this.stop(); - } + super.destroy(); this.listenerContainer.destroy(); - try { - super.destroy(); - } - catch (Exception e) { - throw new IllegalStateException(e); - } } @Override public int beforeShutdown() { - this.stop(); + stop(); return 0; } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java new file mode 100644 index 0000000000..2feefae122 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -0,0 +1,401 @@ +/* + * Copyright 2018-2020 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.integration.kafka.inbound; + +import java.nio.ByteBuffer; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.header.Header; + +import org.springframework.core.AttributeAccessor; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.context.OrderlyShutdownCapable; +import org.springframework.integration.core.Pausable; +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.integration.support.ErrorMessageUtils; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.ConsumerSeekAware; +import org.springframework.kafka.listener.MessageListener; +import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter; +import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.support.converter.ConversionException; +import org.springframework.kafka.support.converter.KafkaMessageHeaders; +import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.RetryListener; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; + +/** + * Inbound gateway. + * + * @param the key type. + * @param the request value type. + * @param the reply value type. + * + * @author Gary Russell + * @author Artem Bilan + * @author Urs Keller + * + * @since 5.4 + * + */ +public class KafkaInboundGateway extends MessagingGatewaySupport implements Pausable, OrderlyShutdownCapable { + + private static final ThreadLocal ATTRIBUTES_HOLDER = new ThreadLocal<>(); + + private final IntegrationRecordMessageListener listener = new IntegrationRecordMessageListener(); + + private final AbstractMessageListenerContainer messageListenerContainer; + + private final KafkaTemplate kafkaTemplate; + + private RetryTemplate retryTemplate; + + private RecoveryCallback recoveryCallback; + + private BiConsumer, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedSeekCallback; + + private boolean bindSourceRecord; + + private boolean containerDeliveryAttemptPresent; + + /** + * Construct an instance with the provided container. + * @param messageListenerContainer the container. + * @param kafkaTemplate the kafka template. + */ + public KafkaInboundGateway(AbstractMessageListenerContainer messageListenerContainer, + KafkaTemplate kafkaTemplate) { + + Assert.notNull(messageListenerContainer, "messageListenerContainer is required"); + Assert.notNull(kafkaTemplate, "kafkaTemplate is required"); + Assert.isNull(messageListenerContainer.getContainerProperties().getMessageListener(), + "Container must not already have a listener"); + this.messageListenerContainer = messageListenerContainer; + this.messageListenerContainer.setAutoStartup(false); + this.kafkaTemplate = kafkaTemplate; + setErrorMessageStrategy(new RawRecordHeaderErrorMessageStrategy()); + } + + /** + * Set the message converter; must be a {@link RecordMessageConverter} or + * {@link org.springframework.kafka.support.converter.BatchMessageConverter} depending on mode. + * @param messageConverter the converter. + */ + public void setMessageConverter(RecordMessageConverter messageConverter) { + this.listener.setMessageConverter(messageConverter); + } + + /** + * When using a type-aware message converter (such as {@code StringJsonMessageConverter}, + * set the payload type the converter should create. Defaults to {@link Object}. + * @param payloadType the type. + */ + public void setPayloadType(Class payloadType) { + this.listener.setFallbackType(payloadType); + } + + /** + * Specify a {@link RetryTemplate} instance to wrap + * {@link KafkaInboundGateway.IntegrationRecordMessageListener} into + * {@link RetryingMessageListenerAdapter}. + * @param retryTemplate the {@link RetryTemplate} to use. + */ + public void setRetryTemplate(RetryTemplate retryTemplate) { + this.retryTemplate = retryTemplate; + } + + /** + * A {@link RecoveryCallback} instance for retry operation; + * if null, the exception will be thrown to the container after retries are exhausted + * (unless an error channel is configured). + * Does not make sense if {@link #setRetryTemplate(RetryTemplate)} isn't specified. + * @param recoveryCallback the recovery callback. + */ + public void setRecoveryCallback(RecoveryCallback recoveryCallback) { + this.recoveryCallback = recoveryCallback; + } + + /** + * Specify a {@link BiConsumer} for seeks management during + * {@link ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback)} + * call from the {@link org.springframework.kafka.listener.KafkaMessageListenerContainer}. + * This is called from the internal + * {@link org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter} implementation. + * @param onPartitionsAssignedCallback the {@link BiConsumer} to use + * @since 3.0.4 + * @see ConsumerSeekAware#onPartitionsAssigned + */ + public void setOnPartitionsAssignedSeekCallback( + BiConsumer, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) { + this.onPartitionsAssignedSeekCallback = onPartitionsAssignedCallback; + } + + /** + * Set to true to bind the source consumer record in the header named + * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}. + * @param bindSourceRecord true to bind. + * @since 3.1.4 + */ + public void setBindSourceRecord(boolean bindSourceRecord) { + this.bindSourceRecord = bindSourceRecord; + } + + @Override + protected void onInit() { + super.onInit(); + MessageListener kafkaListener = this.listener; + if (this.retryTemplate != null) { + kafkaListener = new RetryingMessageListenerAdapter<>(kafkaListener, this.retryTemplate, + this.recoveryCallback); + this.retryTemplate.registerListener(this.listener); + } + this.messageListenerContainer.getContainerProperties().setMessageListener(kafkaListener); + this.containerDeliveryAttemptPresent = this.messageListenerContainer.getContainerProperties() + .isDeliveryAttemptHeader(); + } + + @Override + protected void doStart() { + super.doStart(); + this.messageListenerContainer.start(); + } + + @Override + protected void doStop() { + super.doStop(); + this.messageListenerContainer.stop(); + } + + @Override + public void pause() { + this.messageListenerContainer.pause(); + } + + @Override + public void resume() { + this.messageListenerContainer.resume(); + } + + @Override + public boolean isPaused() { + return this.messageListenerContainer.isContainerPaused(); + } + + @Override + public String getComponentType() { + return "kafka:inbound-gateway"; + } + + @Override + public int beforeShutdown() { + this.messageListenerContainer.stop(); + return getPhase(); + } + + @Override + public int afterShutdown() { + return getPhase(); + } + + /** + * If there's a retry template, it will set the attributes holder via the listener. If + * there's no retry template, but there's an error channel, we create a new attributes + * holder here. If an attributes holder exists (by either method), we set the + * attributes for use by the {@link org.springframework.integration.support.ErrorMessageStrategy}. + * @param record the record. + * @param message the message. + */ + private void setAttributesIfNecessary(Object record, Message message) { + boolean needHolder = getErrorChannel() != null && this.retryTemplate == null; + boolean needAttributes = needHolder | this.retryTemplate != null; + if (needHolder) { + ATTRIBUTES_HOLDER.set(ErrorMessageUtils.getAttributeAccessor(null, null)); + } + if (needAttributes) { + AttributeAccessor attributes = ATTRIBUTES_HOLDER.get(); + if (attributes != null) { + attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message); + attributes.setAttribute(KafkaHeaders.RAW_DATA, record); + } + } + } + + @Override + protected AttributeAccessor getErrorMessageAttributes(Message message) { + AttributeAccessor attributes = ATTRIBUTES_HOLDER.get(); + if (attributes == null) { + return super.getErrorMessageAttributes(message); + } + else { + return attributes; + } + } + + private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter + implements RetryListener { + + IntegrationRecordMessageListener() { + super(null, null); + } + + @Override + public void onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback) { + if (KafkaInboundGateway.this.onPartitionsAssignedSeekCallback != null) { + KafkaInboundGateway.this.onPartitionsAssignedSeekCallback.accept(assignments, callback); + } + } + + @Override + public void onMessage(ConsumerRecord record, Acknowledgment acknowledgment, Consumer consumer) { + Message message = null; + try { + message = enhanceHeaders(toMessagingMessage(record, acknowledgment, consumer), record); + setAttributesIfNecessary(record, message); + } + catch (RuntimeException e) { + if (getErrorChannel() != null) { + KafkaInboundGateway.this.messagingTemplate.send(getErrorChannel(), buildErrorMessage(null, + new ConversionException("Failed to convert to message for: " + record, e))); + } + } + if (message != null) { + try { + Message reply = sendAndReceiveMessage(message); + if (reply != null) { + reply = enhanceReply(message, reply); + KafkaInboundGateway.this.kafkaTemplate.send(reply); + } + } + finally { + if (KafkaInboundGateway.this.retryTemplate == null) { + ATTRIBUTES_HOLDER.remove(); + } + } + } + else { + KafkaInboundGateway.this.logger.debug("Converter returned a null message for: " + record); + } + } + + private Message enhanceHeaders(Message message, ConsumerRecord record) { + Message messageToReturn = message; + if (message.getHeaders() instanceof KafkaMessageHeaders) { + Map rawHeaders = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders(); + if (KafkaInboundGateway.this.retryTemplate != null) { + AtomicInteger deliveryAttempt = + new AtomicInteger(((RetryContext) ATTRIBUTES_HOLDER.get()).getRetryCount() + 1); + rawHeaders.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt); + } + else if (KafkaInboundGateway.this.containerDeliveryAttemptPresent) { + Header header = record.headers().lastHeader(KafkaHeaders.DELIVERY_ATTEMPT); + rawHeaders.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, + new AtomicInteger(ByteBuffer.wrap(header.value()).getInt())); + } + if (KafkaInboundGateway.this.bindSourceRecord) { + rawHeaders.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, record); + } + } + else { + MessageBuilder builder = MessageBuilder.fromMessage(message); + if (KafkaInboundGateway.this.retryTemplate != null) { + AtomicInteger deliveryAttempt = + new AtomicInteger(((RetryContext) ATTRIBUTES_HOLDER.get()).getRetryCount() + 1); + builder.setHeader(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt); + } + else if (KafkaInboundGateway.this.containerDeliveryAttemptPresent) { + Header header = record.headers().lastHeader(KafkaHeaders.DELIVERY_ATTEMPT); + builder.setHeader(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, + new AtomicInteger(ByteBuffer.wrap(header.value()).getInt())); + } + if (KafkaInboundGateway.this.bindSourceRecord) { + builder.setHeader(IntegrationMessageHeaderAccessor.SOURCE_DATA, record); + } + messageToReturn = builder.build(); + } + return messageToReturn; + } + + private Message enhanceReply(Message message, Message reply) { + AbstractIntegrationMessageBuilder builder = null; + MessageHeaders replyHeaders = reply.getHeaders(); + MessageHeaders requestHeaders = message.getHeaders(); + if (replyHeaders.get(KafkaHeaders.CORRELATION_ID) == null && + requestHeaders.get(KafkaHeaders.CORRELATION_ID) != null) { + builder = getMessageBuilderFactory().fromMessage(reply) + .setHeader(KafkaHeaders.CORRELATION_ID, requestHeaders.get(KafkaHeaders.CORRELATION_ID)); + } + if (replyHeaders.get(KafkaHeaders.TOPIC) == null && + requestHeaders.get(KafkaHeaders.REPLY_TOPIC) != null) { + if (builder == null) { + builder = getMessageBuilderFactory().fromMessage(reply); + } + builder.setHeader(KafkaHeaders.TOPIC, requestHeaders.get(KafkaHeaders.REPLY_TOPIC)); + } + if (replyHeaders.get(KafkaHeaders.PARTITION_ID) == null && + requestHeaders.get(KafkaHeaders.REPLY_PARTITION) != null) { + if (builder == null) { + builder = getMessageBuilderFactory().fromMessage(reply); + } + builder.setHeader(KafkaHeaders.PARTITION_ID, requestHeaders.get(KafkaHeaders.REPLY_PARTITION)); + } + if (builder != null) { + return builder.build(); + } + return reply; + } + + @Override + public boolean open(RetryContext context, RetryCallback callback) { + if (KafkaInboundGateway.this.retryTemplate != null) { + ATTRIBUTES_HOLDER.set(context); + } + return true; + } + + @Override + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { + + ATTRIBUTES_HOLDER.remove(); + } + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty + } + + } + +} diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java index 23e69a6906..d2a7946f4c 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java @@ -177,7 +177,6 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv @Override protected void doStart() { Assert.state(getTaskScheduler() != null, "A 'taskScheduler' is required"); - super.doStart(); try { connectAndSubscribe(); } @@ -190,7 +189,6 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv @Override protected synchronized void doStop() { cancelReconnect(); - super.doStop(); if (this.client != null) { try { if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS) diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java new file mode 100644 index 0000000000..880ee53fa4 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java @@ -0,0 +1,248 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.integration.redis.inbound; + +import java.time.Duration; + +import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; +import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.Record; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.core.ReactiveRedisTemplate; +import org.springframework.data.redis.core.ReactiveStreamOperations; +import org.springframework.data.redis.serializer.RedisSerializationContext; +import org.springframework.data.redis.stream.StreamReceiver; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.acks.SimpleAcknowledgment; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.redis.support.RedisHeaders; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * A {@link MessageProducerSupport} for reading messages from a Redis Stream and publishing them into the provided + * output channel. + * By default this adapter reads message as a standalone client {@code XREAD} (Redis command) but can be switched to a + * Consumer Group feature {@code XREADGROUP} by setting {@link #consumerName} field. + * By default the Consumer Group name is the id of this bean {@link #getBeanName()}. + * + * @author Attoumane Ahamadi + * @author Artem Bilan + * @author Rohan Mukesh + * + * @since 5.4 + */ +public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport { + + private final ReactiveRedisConnectionFactory reactiveConnectionFactory; + + private final String streamKey; + + private ReactiveStreamOperations reactiveStreamOperations; + + private StreamReceiver.StreamReceiverOptions streamReceiverOptions = + StreamReceiver.StreamReceiverOptions.builder() + .pollTimeout(Duration.ZERO) + .build(); + + private StreamReceiver streamReceiver; + + private ReadOffset readOffset = ReadOffset.latest(); + + private boolean extractPayload = true; + + private boolean autoAck = true; + + @Nullable + private String consumerGroup; + + @Nullable + private String consumerName; + + private boolean createConsumerGroup; + + public ReactiveRedisStreamMessageProducer(ReactiveRedisConnectionFactory reactiveConnectionFactory, + String streamKey) { + + Assert.notNull(reactiveConnectionFactory, "'connectionFactory' must not be null"); + Assert.hasText(streamKey, "'streamKey' must be set"); + this.reactiveConnectionFactory = reactiveConnectionFactory; + this.streamKey = streamKey; + } + + /** + * Define the offset from which we want to read message. By default the {@link ReadOffset#latest()} is used. + * {@link ReadOffset#latest()} is equal to '$', which is the Id used with {@code XREAD} to get new data added to + * the stream. Note that when switching to the Consumer Group feature, we set it to + * {@link ReadOffset#lastConsumed()} if it is still equal to {@link ReadOffset#latest()}. + * @param readOffset the desired offset + */ + public void setReadOffset(ReadOffset readOffset) { + this.readOffset = readOffset; + } + + /** + * Configure this channel adapter to extract or not value from the {@link Record}. + * @param extractPayload default true + */ + public void setExtractPayload(boolean extractPayload) { + this.extractPayload = extractPayload; + } + + /** + * Set whether or not acknowledge message read in the Consumer Group. {@code true} by default. + * @param autoAck the acknowledge option. + */ + public void setAutoAck(boolean autoAck) { + this.autoAck = autoAck; + } + + /** + * Set the name of the Consumer Group. It is possible to create that Consumer Group if desired, see: + * {@link #createConsumerGroup}. If not set, the defined bean name {@link #getBeanName()} is used. + * @param consumerGroup the Consumer Group on which this adapter should register to listen messages. + */ + public void setConsumerGroup(@Nullable String consumerGroup) { + this.consumerGroup = consumerGroup; + } + + /** + * Set the name of the consumer. When a consumer name is provided, this adapter is switched to the Consumer Group + * feature. Note that this value should be unique in the group. + * @param consumerName the consumer name in the Consumer Group + */ + public void setConsumerName(@Nullable String consumerName) { + this.consumerName = consumerName; + } + + /** + * Create the Consumer Group if and only if it does not exist. + * During the creation we also create the stream, see {@code MKSTREAM}. + * @param createConsumerGroup specify if we should create the Consumer Group, {@code false} by default + */ + public void setCreateConsumerGroup(boolean createConsumerGroup) { + this.createConsumerGroup = createConsumerGroup; + } + + /** + * Set {@link ReactiveStreamOperations} used to customize the {@link StreamReceiver}. + * It provides a way to set the polling timeout and the serialization context. + * By default the polling timeout is set to infinite and + * {@link org.springframework.data.redis.serializer.StringRedisSerializer} is used. + * @param streamReceiverOptions the desired receiver options + * */ + public void setStreamReceiverOptions( + @Nullable StreamReceiver.StreamReceiverOptions streamReceiverOptions) { + + this.streamReceiverOptions = streamReceiverOptions; + } + + @Override + public String getComponentType() { + return "redis:stream-inbound-channel-adapter"; + } + + @Override + protected void onInit() { + super.onInit(); + this.streamReceiver = StreamReceiver.create(this.reactiveConnectionFactory, this.streamReceiverOptions); + if (StringUtils.hasText(this.consumerName) && !StringUtils.hasText(this.consumerGroup)) { + this.consumerGroup = getBeanName(); + } + ReactiveRedisTemplate reactiveRedisTemplate = + new ReactiveRedisTemplate<>(this.reactiveConnectionFactory, RedisSerializationContext.string()); + this.reactiveStreamOperations = reactiveRedisTemplate.opsForStream(); + } + + @Override + protected void doStart() { + StreamOffset offset = StreamOffset.create(this.streamKey, this.readOffset); + + Flux> events; + + if (!StringUtils.hasText(this.consumerName)) { + events = this.streamReceiver.receive(offset); + } + else { + Mono consumerGroupMono = Mono.empty(); + if (this.createConsumerGroup) { + consumerGroupMono = + this.reactiveStreamOperations.createGroup(this.streamKey, this.consumerGroup) // NOSONAR + .onErrorReturn(this.consumerGroup); + } + + Consumer consumer = Consumer.from(this.consumerGroup, this.consumerName); // NOSONAR + + if (offset.getOffset().equals(ReadOffset.latest())) { + // for consumer group offset id should be equal to '>' + offset = StreamOffset.create(this.streamKey, ReadOffset.lastConsumed()); + } + + events = + this.autoAck + ? this.streamReceiver.receiveAutoAck(consumer, offset) + : this.streamReceiver.receive(consumer, offset); + + events = consumerGroupMono.thenMany(events); + + } + + Flux> messageFlux = + events.map((record) -> buildMessageFromRecord(record, this.extractPayload)) + .onErrorContinue((ex, record) -> { + @SuppressWarnings("unchecked") + Message failedMessage = buildMessageFromRecord((Record) record, false); + MessagingException conversionException = + new MessageConversionException(failedMessage, + "Cannot deserialize Redis Stream Record", ex); + if (!sendErrorMessageIfNecessary(null, conversionException)) { + logger.getLog().error(conversionException); + } + }); + subscribeToPublisher(messageFlux); + } + + private Message buildMessageFromRecord(Record record, boolean extractPayload) { + AbstractIntegrationMessageBuilder builder = + getMessageBuilderFactory() + .withPayload(extractPayload ? record.getValue() : record) + .setHeader(RedisHeaders.STREAM_KEY, record.getStream()) + .setHeader(RedisHeaders.STREAM_MESSAGE_ID, record.getId()) + .setHeader(RedisHeaders.CONSUMER_GROUP, this.consumerGroup) + .setHeader(RedisHeaders.CONSUMER, this.consumerName); + + if (!this.autoAck && this.consumerGroup != null) { + builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, + (SimpleAcknowledgment) () -> + this.reactiveStreamOperations + .acknowledge(this.consumerGroup, record) + .subscribe()); + } + + return builder.build(); + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java index 0b4de82a09..c825e7af09 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2019 the original author or authors. + * Copyright 2007-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -140,14 +140,12 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { @Override protected void doStart() { - super.doStart(); this.container.start(); } @Override protected void doStop() { - super.doStop(); this.container.stop(); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java index 450354e3ec..f5be05518d 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java @@ -81,8 +81,6 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport private boolean extractPayload = true; - private volatile boolean active; - private volatile boolean listening; private volatile Runnable stopCallback; @@ -173,7 +171,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport private void handlePopException(Exception e) { this.listening = false; - if (this.active) { + if (isActive()) { logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval + " milliseconds.", e); publishException(e); @@ -196,7 +194,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport } String uuid = null; if (value != null) { - if (!this.active) { + if (!isActive()) { this.boundListOperations.rightPush(value); return; } @@ -213,7 +211,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport } Message requestMessage = null; if (value != null) { - if (!this.active) { + if (!isActive()) { this.template.boundListOps(uuid).rightPush(value); byte[] serialized = stringSerializer.serialize(uuid); if (serialized != null) { @@ -281,10 +279,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport @Override protected void doStart() { super.doStart(); - if (!this.active) { - this.active = true; - this.restart(); - } + restart(); } /** @@ -327,7 +322,6 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport @Override protected void doStop() { super.doStop(); - this.active = false; this.listening = false; } @@ -369,13 +363,13 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport @Override public void run() { try { - while (RedisQueueInboundGateway.this.active) { + while (isActive()) { RedisQueueInboundGateway.this.listening = true; receiveAndReply(); } } finally { - if (RedisQueueInboundGateway.this.active) { + if (isActive()) { restart(); } else if (RedisQueueInboundGateway.this.stopCallback != null) { diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java index b4adc2e3a4..2dffca0e30 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java @@ -82,8 +82,6 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport private boolean rightPop = true; - private volatile boolean active; - private volatile boolean listening; private volatile Runnable stopCallback; @@ -243,7 +241,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport } catch (Exception e) { this.listening = false; - if (this.active) { + if (isActive()) { logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval + " milliseconds.", e); publishException(e); @@ -258,10 +256,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport @Override protected void doStart() { - if (!this.active) { - this.active = true; - this.restart(); - } + restart(); } /** @@ -304,7 +299,6 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport @Override protected void doStop() { super.doStop(); - this.active = false; this.listening = false; } @@ -346,14 +340,14 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport @Override public void run() { try { - while (RedisQueueMessageDrivenEndpoint.this.active) { + while (isActive()) { RedisQueueMessageDrivenEndpoint.this.listening = true; - RedisQueueMessageDrivenEndpoint.this.popMessageAndSend(); + popMessageAndSend(); } } finally { - if (RedisQueueMessageDrivenEndpoint.this.active) { - RedisQueueMessageDrivenEndpoint.this.restart(); + if (isActive()) { + restart(); } else if (RedisQueueMessageDrivenEndpoint.this.stopCallback != null) { RedisQueueMessageDrivenEndpoint.this.stopCallback.run(); diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/TcpSyslogReceivingChannelAdapter.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/TcpSyslogReceivingChannelAdapter.java index 1fe34de29d..9151f75490 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/TcpSyslogReceivingChannelAdapter.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/TcpSyslogReceivingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -30,6 +30,7 @@ import org.springframework.messaging.Message; * * @author Gary Russell * @author Artem Bilan + * * @since 3.0 * */ @@ -84,19 +85,17 @@ public class TcpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdap @Override protected void doStart() { - super.doStart(); this.connectionFactory.start(); } @Override protected void doStop() { - super.doStop(); this.connectionFactory.stop(); } @Override public boolean onMessage(Message message) { - this.convertAndSend(message); + convertAndSend(message); return false; } diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/UdpSyslogReceivingChannelAdapter.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/UdpSyslogReceivingChannelAdapter.java index 9b347c6ab5..4d7a7cce95 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/UdpSyslogReceivingChannelAdapter.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/UdpSyslogReceivingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -24,6 +24,7 @@ import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter; * * @author Gary Russell * @author Artem Bilan + * * @since 3.0 * */ @@ -33,8 +34,8 @@ public class UdpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdap private volatile boolean udpAdapterSet; - public void setUdpAdapter(UnicastReceivingChannelAdapter udpAdpter) { - this.udpAdapter = udpAdpter; + public void setUdpAdapter(UnicastReceivingChannelAdapter udpAdapter) { + this.udpAdapter = udpAdapter; this.udpAdapterSet = true; } @@ -65,7 +66,7 @@ public class UdpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdap "of the provided 'UnicastReceivingChannelAdapter' to support Syslog conversion " + "for the incoming UDP packets"); } - this.udpAdapter.setOutputChannel(new FixedSubscriberChannel(message -> convertAndSend(message))); + this.udpAdapter.setOutputChannel(new FixedSubscriberChannel(this::convertAndSend)); if (!this.udpAdapterSet) { this.udpAdapter.afterPropertiesSet(); } @@ -73,13 +74,11 @@ public class UdpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdap @Override protected void doStart() { - super.doStart(); this.udpAdapter.start(); } @Override protected void doStop() { - super.doStop(); this.udpAdapter.stop(); } diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java index e02d99a871..f195923463 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -108,8 +108,6 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport private AbstractBrokerMessageHandler brokerHandler; - private volatile boolean active; - public WebSocketInboundChannelAdapter(IntegrationWebSocketContainer webSocketContainer) { this(webSocketContainer, new SubProtocolHandlerRegistry(new PassThruSubProtocolHandler())); } @@ -127,10 +125,10 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport try { handleMessageAndSend(message); } - catch (Exception e) { + catch (Exception ex) { throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, () -> "Failed to handle and process message in the [" - + WebSocketInboundChannelAdapter.this + ']', e); + + WebSocketInboundChannelAdapter.this + ']', ex); } }); } @@ -267,7 +265,6 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport @Override protected void doStart() { - this.active = true; if (this.webSocketContainer instanceof Lifecycle) { ((Lifecycle) this.webSocketContainer).start(); } @@ -275,17 +272,17 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport @Override protected void doStop() { - this.active = false; if (this.webSocketContainer instanceof Lifecycle) { ((Lifecycle) this.webSocketContainer).stop(); } } - private boolean isActive() { - if (!this.active) { + public boolean isActive() { + boolean active = super.isActive(); + if (!active) { logger.warn("MessageProducer '" + this + "' isn't started to accept WebSocket events."); } - return this.active; + return active; } @SuppressWarnings("unchecked") diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java new file mode 100644 index 0000000000..0263496ce7 --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java @@ -0,0 +1,317 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.integration.zeromq.inbound; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; + +import org.zeromq.SocketType; +import org.zeromq.ZContext; +import org.zeromq.ZMQ; +import org.zeromq.ZMsg; + +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.mapping.ConvertingBytesMessageMapper; +import org.springframework.integration.mapping.InboundMessageMapper; +import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter; +import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.integration.zeromq.ZeroMqHeaders; +import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.util.Assert; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; +import reactor.core.scheduler.Schedulers; + +/** + * A {@link MessageProducerSupport} implementation for consuming messages from ZeroMq socket. + * Only {@link SocketType#PAIR}, {@link SocketType#SUB} and {@link SocketType#PULL} are supported. + * This component can bind or connect the socket. + *

+ * When the {@link SocketType#SUB} is used, the received topic is stored in the {@link ZeroMqHeaders#TOPIC}. + * + * @author Artem Bilan + * + * @since 5.4 + */ +@ManagedResource +@IntegrationManagedResource +public class ZeroMqMessageProducer extends MessageProducerSupport { + + public static final Duration DEFAULT_CONSUME_DELAY = Duration.ofSeconds(1); + + private static final List VALID_SOCKET_TYPES = + Arrays.asList(SocketType.PAIR, SocketType.PULL, SocketType.SUB); + + private final Scheduler consumerScheduler = Schedulers.newSingle("zeroMqMessageProducerScheduler"); + + private final AtomicInteger bindPort = new AtomicInteger(); + + private final ZContext context; + + private final SocketType socketType; + + private InboundMessageMapper messageMapper; + + private Consumer socketConfigurer = (socket) -> { }; + + private Duration consumeDelay = DEFAULT_CONSUME_DELAY; + + private String[] topics = { "" }; // Equivalent to ZMQ#SUBSCRIPTION_ALL + + private boolean receiveRaw; + + @Nullable + private String connectUrl; + + private volatile Mono socketMono; + + public ZeroMqMessageProducer(ZContext context) { + this(context, SocketType.PAIR); + } + + public ZeroMqMessageProducer(ZContext context, SocketType socketType) { + Assert.notNull(context, "'context' must not be null"); + Assert.state(VALID_SOCKET_TYPES.contains(socketType), + () -> "'socketType' can only be one of the: " + VALID_SOCKET_TYPES); + this.context = context; + this.socketType = socketType; + } + + /** + * Specify a {@link Duration} to delay consumption when no data received. + * @param consumeDelay the {@link Duration} to delay consumption when empty; + * defaults to {@link #DEFAULT_CONSUME_DELAY}. + */ + public void setConsumeDelay(Duration consumeDelay) { + Assert.notNull(consumeDelay, "'consumeDelay' must not be null"); + this.consumeDelay = consumeDelay; + } + + /** + * Provide an {@link InboundMessageMapper} to convert a consumed data into a message to produce. + * Ignored when {@link #setReceiveRaw(boolean)} is {@code true}. + * @param messageMapper the {@link InboundMessageMapper} to use. + */ + public void setMessageMapper(InboundMessageMapper messageMapper) { + Assert.notNull(messageMapper, "'messageMapper' must not be null"); + this.messageMapper = messageMapper; + } + + /** + * Provide a {@link MessageConverter} (as an alternative to {@link #messageMapper}) + * for converting a consumed data into a message to produce. + * Ignored when {@link #setReceiveRaw(boolean)} is {@code true}. + * @param messageConverter the {@link MessageConverter} to use. + */ + public void setMessageConverter(MessageConverter messageConverter) { + setMessageMapper(new ConvertingBytesMessageMapper(messageConverter)); + } + + /** + * Whether raw {@link ZMsg} is present as a payload of message to produce or + * it is fully converted to a {@link Message} including {@link ZeroMqHeaders#TOPIC} header (if any). + * @param receiveRaw to convert from {@link ZMsg} or not; defaults to convert. + */ + public void setReceiveRaw(boolean receiveRaw) { + this.receiveRaw = receiveRaw; + } + + /** + * Provide a {@link Consumer} to configure a socket with arbitrary options, like security. + * @param socketConfigurer the configurer for socket options. + */ + public void setSocketConfigurer(Consumer socketConfigurer) { + Assert.notNull(socketConfigurer, "'socketConfigurer' must not be null"); + this.socketConfigurer = socketConfigurer; + } + + /** + * Specify topics the {@link SocketType#SUB} socket is going to use for subscription. + * It is ignored for all other {@link SocketType}s supported. + * @param topics the topics to use. + */ + public void setTopics(String... topics) { + Assert.notNull(topics, "'topics' cannot be null"); + Assert.noNullElements(topics, "'topics' must not contain null elements"); + this.topics = Arrays.copyOf(topics, topics.length); + } + + /** + * Configure an URL for {@link org.zeromq.ZMQ.Socket#connect(String)}. + * Mutually exclusive with the {@link #setBindPort(int)}. + * @param connectUrl the URL to connect ZeroMq socket to. + */ + public void setConnectUrl(@Nullable String connectUrl) { + this.connectUrl = connectUrl; + } + + /** + * Configure a port for TCP protocol binding via {@link org.zeromq.ZMQ.Socket#bind(String)}. + * Mutually exclusive with the {@link #setConnectUrl(String)}. + * @param port the port to bind ZeroMq socket to over TCP. + */ + public void setBindPort(int port) { + Assert.isTrue(port > 0, "'port' must not be zero or negative"); + this.bindPort.set(port); + } + + /** + * Return the port a socket is bound or 0 if this message producer has not been started yet + * or the socket is connected - not bound. + * @return the port for a socket or 0. + */ + public int getBoundPort() { + return this.bindPort.get(); + } + + @Override + public String getComponentType() { + return "zeromq:inbound-channel-adapter"; + } + + @Override + protected void onInit() { + super.onInit(); + Assert.state(this.connectUrl == null || this.bindPort.get() == 0, + "Only one of the 'connectUrl' or `bindPort` must be provided on none"); + if (this.messageMapper == null && !this.receiveRaw) { + ConfigurableCompositeMessageConverter messageConverter = new ConfigurableCompositeMessageConverter(); + messageConverter.setBeanFactory(getBeanFactory()); + messageConverter.afterPropertiesSet(); + this.messageMapper = new ConvertingBytesMessageMapper(messageConverter); + } + } + + @ManagedOperation + public void subscribeToTopics(String... topics) { + Assert.state(SocketType.SUB.equals(this.socketType), "Only SUB socket can accept a subscription option."); + Assert.state(isActive(), "This message producer is not active to accept a new subscription."); + + Flux.fromArray(topics) + .flatMap((topic) -> + this.socketMono.doOnNext((socket) -> socket.subscribe(topic))) + .subscribe(); + } + + @ManagedOperation + public void unsubscribeFromTopics(String... topics) { + Assert.state(SocketType.SUB.equals(this.socketType), "Only SUB socket can accept a unsubscription option."); + Assert.state(isActive(), "This message producer is not active to cancel a subscription."); + + Flux.fromArray(topics) + .flatMap((topic) -> + this.socketMono.doOnNext((socket) -> socket.unsubscribe(topic))) + .subscribe(); + } + + @Override + protected void doStart() { + this.socketMono = + Mono.just(this.context.createSocket(this.socketType)) + .publishOn(this.consumerScheduler) + .doOnNext(this.socketConfigurer) + .doOnNext((socket) -> { + if (SocketType.SUB.equals(this.socketType)) { + for (String topic : this.topics) { + socket.subscribe(topic); + } + } + }) + .doOnNext((socket) -> { + if (this.connectUrl != null) { + socket.connect(this.connectUrl); + } + else { + this.bindPort.set(bindSocket(socket, this.bindPort.get())); + } + }) + .cache() + .publishOn(this.consumerScheduler); + + Flux> dataFlux = + this.socketMono + .flatMap((socket) -> { + if (isRunning()) { + ZMsg msg = ZMsg.recvMsg(socket, false); + if (msg != null) { + return Mono.just(msg); + } + } + return Mono.empty(); + }) + .publishOn(Schedulers.boundedElastic()) + .transform((msgMono) -> this.receiveRaw ? mapRaw(msgMono) : convertMessage(msgMono)) + .doOnError((error) -> + logger.error(error, () -> "Error processing ZeroMQ message in the " + this)) + .repeatWhenEmpty((repeat) -> + isActive() ? repeat.delayElements(this.consumeDelay) : repeat) + .repeat(this::isActive) + .doOnComplete(this.consumerScheduler::dispose); + + subscribeToPublisher(dataFlux); + } + + private Mono> mapRaw(Mono msgMono) { + return msgMono.map((msg) -> getMessageBuilderFactory().withPayload(msg).build()); + } + + private Mono> convertMessage(Mono msgMono) { + return msgMono.map((msg) -> { + Map headers = null; + if (msg.size() > 1) { + headers = Collections.singletonMap(ZeroMqHeaders.TOPIC, msg.unwrap().getString(ZMQ.CHARSET)); + } + return this.messageMapper.toMessage(msg.getLast().getData(), headers); // NOSONAR + }); + } + + @Override + protected void doStop() { + this.socketMono.doOnNext(ZMQ.Socket::close).subscribe(); + } + + @Override + public void destroy() { + super.destroy(); + this.socketMono.doOnNext(ZMQ.Socket::close).block(); + } + + private static int bindSocket(ZMQ.Socket socket, int port) { + if (port == 0) { + return socket.bindToRandomPort("tcp://*"); + } + else { + boolean bound = socket.bind("tcp://*:" + port); + if (!bound) { + throw new IllegalArgumentException("Cannot bind ZeroMQ socket to port: " + port); + } + return port; + } + } + +}