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

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}.

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);
 	}
 

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);

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;

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;
 		}

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<ResolvableType> eventSet = new HashSet<ResolvableType>(eventTypes.length);
+		Set<ResolvableType> 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();
 	}
 
 }

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());
 	}
 
 }

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);
 		}

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;
+		}
 	}
 
 }

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();
 		}
 	}

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();
 	}
+
 }

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

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;
 	}
 

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 <K> the key type.
+ * @param <V> the request value type.
+ * @param <R> the reply value type.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ * @author Urs Keller
+ *
+ * @since 5.4
+ *
+ */
+public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implements Pausable, OrderlyShutdownCapable {
+
+	private static final ThreadLocal<AttributeAccessor> ATTRIBUTES_HOLDER = new ThreadLocal<>();
+
+	private final IntegrationRecordMessageListener listener = new IntegrationRecordMessageListener();
+
+	private final AbstractMessageListenerContainer<K, V> messageListenerContainer;
+
+	private final KafkaTemplate<K, R> kafkaTemplate;
+
+	private RetryTemplate retryTemplate;
+
+	private RecoveryCallback<?> recoveryCallback;
+
+	private BiConsumer<Map<TopicPartition, Long>, 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<K, V> messageListenerContainer,
+			KafkaTemplate<K, R> 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<Map<TopicPartition, Long>, 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<K, V> 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<K, V>
+			implements RetryListener {
+
+		IntegrationRecordMessageListener() {
+			super(null, null);
+		}
+
+		@Override
+		public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
+			if (KafkaInboundGateway.this.onPartitionsAssignedSeekCallback != null) {
+				KafkaInboundGateway.this.onPartitionsAssignedSeekCallback.accept(assignments, callback);
+			}
+		}
+
+		@Override
+		public void onMessage(ConsumerRecord<K, V> 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<K, V> record) {
+			Message<?> messageToReturn = message;
+			if (message.getHeaders() instanceof KafkaMessageHeaders) {
+				Map<String, Object> 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 <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
+			if (KafkaInboundGateway.this.retryTemplate != null) {
+				ATTRIBUTES_HOLDER.set(context);
+			}
+			return true;
+		}
+
+		@Override
+		public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
+				Throwable throwable) {
+
+			ATTRIBUTES_HOLDER.remove();
+		}
+
+		@Override
+		public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
+				Throwable throwable) {
+			// Empty
+		}
+
+	}
+
+}

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)

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<String, ?, ?> reactiveStreamOperations;
+
+	private StreamReceiver.StreamReceiverOptions<String, ?> streamReceiverOptions =
+			StreamReceiver.StreamReceiverOptions.builder()
+					.pollTimeout(Duration.ZERO)
+					.build();
+
+	private StreamReceiver<String, ?> 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<String, ?> 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<String, ?> reactiveRedisTemplate =
+				new ReactiveRedisTemplate<>(this.reactiveConnectionFactory, RedisSerializationContext.string());
+		this.reactiveStreamOperations = reactiveRedisTemplate.opsForStream();
+	}
+
+	@Override
+	protected void doStart() {
+		StreamOffset<String> offset = StreamOffset.create(this.streamKey, this.readOffset);
+
+		Flux<? extends Record<String, ?>> 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<? extends Message<?>> messageFlux =
+				events.map((record) -> buildMessageFromRecord(record, this.extractPayload))
+						.onErrorContinue((ex, record) -> {
+							@SuppressWarnings("unchecked")
+							Message<?> failedMessage = buildMessageFromRecord((Record<String, ?>) 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<String, ?> 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();
+	}
+
+}

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();
 	}
 

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<Object> 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) {

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();

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;
 	}
 

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();
 	}
 

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")

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.
+ * <p>
+ * 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<SocketType> 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<byte[]> messageMapper;
+
+	private Consumer<ZMQ.Socket> 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<ZMQ.Socket> 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<byte[]> 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<ZMQ.Socket> 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<? extends Message<?>> 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<Message<?>> mapRaw(Mono<ZMsg> msgMono) {
+		return msgMono.map((msg) -> getMessageBuilderFactory().withPayload(msg).build());
+	}
+
+	private Mono<Message<?>> convertMessage(Mono<ZMsg> msgMono) {
+		return msgMono.map((msg) -> {
+			Map<String, Object> 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;
+		}
+	}
+
+}