INT-3045: Add ZeroMqChannel support (#3355)

* INT-3045: Add `ZeroMqChannel` support

JIRA: https://jira.spring.io/browse/INT-3045

Provide a `SubscribableChannel` implementation for ZeroMQ

The general idea is to let to have a distributed channel implementation
where every client can connect to a single server backed by the channel.

The logic in the channel is fully transparent for end-user and there is just
enough to send message to it and subscribe for receiving on the other side.
If PUB/SUB model is used, all the subscribes (even over the network) going to
receive the same published message.
In case of PUSH/PULL only one subscriber in the whole cluster is going to get
the published message

* Use Reactor for better threading control
* JeroMQ is not interruptible-friendly: use control sockets to stop proxy loop
* Name Reactor's schedulers to avoid daemon threads

* * Use try-catch-with-resource to close sockets automatically
* Fix Checkstyle violations
* Use `Mono.handle()` to receive data from the socket

* * Optimize local for just a couple of PAIR sockets
* Implement TCP binding
* Add PUB/SUB tests

* * Fix subscriber scheduler name
* Optimize socket create logic
* Add PUSH/PULL over TCP test

* * Fix subscriber scheduler name
* Optimize socket create logic
* Add PUSH/PULL over TCP test
* Implement PUB/SUB over TCP

* * Introduce `ZeroMqProxy` - Spring-friendly component to configure and manage ZeroMq proxy
* Use this `ZeroMqProxy` logic as an external component for `ZeroMqChannel` testing

* * Fix Checkstyle
* Apply docs polishing
* Expose a capture socket on the proxy
* Implement `DisposableBean` in the `ZeroMqProxy` to destroy an internal executor service
* Add JavaDocs to `ZeroMqChannel`
* Add one more `ZeroMqChannel` to TCP test to be sure that proxy distribution works well

* * Add `hamcrest-core` dependency for Awatility

* * Add more JavaDocs to `ZeroMqProxy` and `ZeroMqChannel`
* Expose `ZeroMqChannel.setZeroMqProxy()` option for easier
configuration within the same application context
* Make `ZeroMqChannel` sockets configuration and connection
dependant on provided `ZeroMqProxy` (if any)
* Add `Consumer<ZMQ.Socket>` configuration callbacks to the `ZeroMqChannel`
* Expose `ZeroMqChannel.consumeDelay` option

* * Add docs for ZeroMQ
* Some additions into a `reactive-streams.adoc`
* Fix typo in the `xmpp.adoc`

* * Add `optional` `jackson-databind` since `ZeroMqChannel` uses it by default
* More words into docs

* * Fix language in docs according review

* Fix language in docs according review

Co-authored-by: Gary Russell <grussell@vmware.com>

* Apply suggestions from code review

Co-authored-by: Oliver <oli-ver@users.noreply.github.com>

* * Fix threading using a `publishOn()` for specific scheduler after `cache()`

* * Remove unused import

* * Change proxy port check from static `Mono.just()` to `Mono.fromCallable()`
to really evaluate the current port state on every repeat
* Add finite `100` repeat number to avoid infinite blocking when proxy is not started at all
* Add `doOnError()` for proxy `Mono` to log `ERROR` when repeat is exhausted

* * Fix Checkstyle violation

Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Oliver <oli-ver@users.noreply.github.com>

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java

@@ -0,0 +1,367 @@
+/*
+ * 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;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.zeromq.SocketType;
+import org.zeromq.ZContext;
+import org.zeromq.ZMQ;
+
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.lang.Nullable;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+import org.springframework.util.Assert;
+
+/**
+ * This class encapsulates the logic to configure and manage a ZeroMQ proxy.
+ * It binds frontend and backend sockets over TCP on all the available network interfaces
+ * with either provided or randomly selected ports.
+ * <p>
+ * The {@link ZeroMqProxy.Type} dictates which pair of ZeroMQ sockets to bind with this proxy
+ * to implement any possible patterns for ZeroMQ intermediary. Defaults to @link {@link ZeroMqProxy.Type#PULL_PUSH}.
+ * <p>
+ * The control socket is exposed as a {@link SocketType#PAIR} with an inter-thread transport
+ * on the {@code "inproc://" + beanName + ".control"} address; it can be obtained via {@link #getControlAddress()}.
+ * Should be used with the same application from {@link SocketType#PAIR} socket to send
+ * {@link zmq.ZMQ#PROXY_TERMINATE}, {@link zmq.ZMQ#PROXY_PAUSE} and/or {@link zmq.ZMQ#PROXY_RESUME} commands.
+ * <p>
+ * If the proxy cannot be started for some reason, an error message is logged and this component is
+ * left in the non-started state.
+ * <p>
+ * With an {@link #exposeCaptureSocket} option, an additional capture data socket is bound to inter-thread transport
+ * as a {@link SocketType#PUB}. There is no specific topic selection, so all the subscribers to this socket
+ * must subscribe with plain {@link ZMQ#SUBSCRIPTION_ALL}.
+ * The address for this socket is {@code "inproc://" + beanName + ".capture"}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.4
+ *
+ * @see ZMQ#proxy(ZMQ.Socket, ZMQ.Socket, ZMQ.Socket)
+ */
+public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAware, DisposableBean {
+
+	private static final Log LOG = LogFactory.getLog(ZeroMqProxy.class);
+
+	private final ZContext context;
+
+	private final Type type;
+
+	private final AtomicBoolean running = new AtomicBoolean();
+
+	private final AtomicInteger frontendPort = new AtomicInteger();
+
+	private final AtomicInteger backendPort = new AtomicInteger();
+
+	private String controlAddress;
+
+	private Executor proxyExecutor;
+
+	private boolean proxyExecutorExplicitlySet;
+
+	@Nullable
+	private Consumer<ZMQ.Socket> frontendSocketConfigurer;
+
+	@Nullable
+	private Consumer<ZMQ.Socket> backendSocketConfigurer;
+
+	private boolean exposeCaptureSocket;
+
+	@Nullable
+	private String captureAddress;
+
+	private String beanName;
+
+	private boolean autoStartup;
+
+	private int phase;
+
+	/**
+	 * Create a {@link ZeroMqProxy} instance based on the provided {@link ZContext}
+	 * and {@link Type#PULL_PUSH} as default mode.
+	 * @param context the {@link ZContext} to use
+	 */
+	public ZeroMqProxy(ZContext context) {
+		this(context, Type.PULL_PUSH);
+	}
+
+	/**
+	 * Create a {@link ZeroMqProxy} instance based on the provided {@link ZContext}
+	 * and {@link Type}.
+	 * @param context the {@link ZContext} to use
+	 * @param type the {@link Type} to use.
+	 */
+	public ZeroMqProxy(ZContext context, Type type) {
+		Assert.notNull(context, "'context' must not be null");
+		Assert.notNull(type, "'type' must not be null");
+		this.context = context;
+		this.type = type;
+	}
+
+	/**
+	 * Configure an executor to perform a ZeroMQ proxy loop.
+	 * The thread is held until ZeroMQ proxy loop is terminated.
+	 * By default an internal {@link Executors#newSingleThreadExecutor} instance is used.
+	 * @param proxyExecutor the {@link Executor} to use for ZeroMQ proxy loop
+	 */
+	public void setProxyExecutor(Executor proxyExecutor) {
+		Assert.notNull(proxyExecutor, "'proxyExecutor' must not be null");
+		this.proxyExecutor = proxyExecutor;
+		this.proxyExecutorExplicitlySet = true;
+	}
+
+	/**
+	 * Specify a fixed port for frontend socket of the proxy.
+	 * @param frontendPort the port to use; must be more than 0
+	 */
+	public void setFrontendPort(int frontendPort) {
+		Assert.isTrue(frontendPort > 0, "'frontendPort' must not be zero or negative");
+		this.frontendPort.set(frontendPort);
+	}
+
+	/**
+	 * Specify a fixed port for backend socket of the proxy.
+	 * @param backendPort the port to use; must be more than 0
+	 */
+	public void setBackendPort(int backendPort) {
+		Assert.isTrue(backendPort > 0, "'backendPort' must not be zero or negative");
+		this.backendPort.set(backendPort);
+	}
+
+	/**
+	 * Provide a {@link Consumer} to configure a proxy frontend socket with arbitrary options, like security.
+	 * @param frontendSocketConfigurer the configurer for frontend socket
+	 */
+	public void setFrontendSocketConfigurer(@Nullable Consumer<ZMQ.Socket> frontendSocketConfigurer) {
+		this.frontendSocketConfigurer = frontendSocketConfigurer;
+	}
+
+	/**
+	 * Provide a {@link Consumer} to configure a proxy backend socket with arbitrary options, like security.
+	 * @param backendSocketConfigurer the configurer for backend socket
+	 */
+	public void setBackendSocketConfigurer(@Nullable Consumer<ZMQ.Socket> backendSocketConfigurer) {
+		this.backendSocketConfigurer = backendSocketConfigurer;
+	}
+
+	/**
+	 * Whether to bind and expose a capture socket for the proxy data.
+	 * @param exposeCaptureSocket true to bind capture socket for proxy
+	 */
+	public void setExposeCaptureSocket(boolean exposeCaptureSocket) {
+		this.exposeCaptureSocket = exposeCaptureSocket;
+	}
+
+	@Override
+	public void setBeanName(String beanName) {
+		this.beanName = beanName;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	public Type getType() {
+		return this.type;
+	}
+
+	/**
+	 * Return the port a frontend socket is bound or 0 if this proxy has not been started yet.
+	 * @return the port for a frontend socket or 0
+	 */
+	public int getFrontendPort() {
+		return this.frontendPort.get();
+	}
+
+	/**
+	 * Return the port a backend socket is bound or null if this proxy has not been started yet.
+	 * @return the port for a backend socket or 0
+	 */
+	public int getBackendPort() {
+		return this.backendPort.get();
+	}
+
+	/**
+	 * Return the address an {@code inproc} control socket is bound or null if this proxy has not been started yet.
+	 * @return the the address for control socket or null
+	 */
+	@Nullable
+	public String getControlAddress() {
+		return this.controlAddress;
+	}
+
+	/**
+	 * Return the address an {@code inproc} capture socket is bound or null if this proxy has not been started yet
+	 * or {@link #captureAddress} is false.
+	 * @return the the address for capture socket or null
+	 */
+	@Nullable
+	public String getCaptureAddress() {
+		return this.captureAddress;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	@Override
+	public int getPhase() {
+		return this.phase;
+	}
+
+	@Override
+	public void afterPropertiesSet() {
+		if (this.proxyExecutor == null) {
+			this.proxyExecutor = Executors.newSingleThreadExecutor(new CustomizableThreadFactory(this.beanName));
+		}
+		this.controlAddress = "inproc://" + this.beanName + ".control";
+		if (this.exposeCaptureSocket) {
+			this.captureAddress = "inproc://" + this.beanName + ".capture";
+		}
+	}
+
+	@Override
+	public synchronized void start() {
+		if (!this.running.get()) {
+			this.proxyExecutor
+					.execute(() -> {
+						ZMQ.Socket captureSocket = null;
+						if (this.exposeCaptureSocket) {
+							captureSocket = this.context.createSocket(SocketType.PUB);
+						}
+						try (
+								ZMQ.Socket frontendSocket = this.context.createSocket(this.type.getFrontendSocketType());
+								ZMQ.Socket backendSocket = this.context.createSocket(this.type.getBackendSocketType());
+								ZMQ.Socket controlSocket = this.context.createSocket(SocketType.PAIR)
+						) {
+
+							if (this.frontendSocketConfigurer != null) {
+								this.frontendSocketConfigurer.accept(frontendSocket);
+							}
+
+							if (this.backendSocketConfigurer != null) {
+								this.backendSocketConfigurer.accept(backendSocket);
+							}
+
+							this.frontendPort.set(bindSocket(frontendSocket, this.frontendPort.get()));
+							this.backendPort.set(bindSocket(backendSocket, this.backendPort.get()));
+							boolean bound = controlSocket.bind(this.controlAddress);
+							if (!bound) {
+								throw new IllegalArgumentException("Cannot bind ZeroMQ socket to address: "
+										+ this.controlAddress);
+							}
+							if (captureSocket != null) {
+								bound = captureSocket.bind(this.captureAddress);
+								if (!bound) {
+									throw new IllegalArgumentException("Cannot bind ZeroMQ socket to address: "
+											+ this.captureAddress);
+								}
+							}
+							this.running.set(true);
+							ZMQ.proxy(frontendSocket, backendSocket, captureSocket, controlSocket);
+						}
+						catch (Exception ex) {
+							LOG.error("Cannot start ZeroMQ proxy from bean: " + this.beanName, ex);
+						}
+						finally {
+							if (captureSocket != null) {
+								captureSocket.close();
+							}
+						}
+					});
+		}
+	}
+
+	@Override
+	public synchronized void stop() {
+		if (this.running.getAndSet(false)) {
+			try (ZMQ.Socket commandSocket = this.context.createSocket(SocketType.PAIR)) {
+				commandSocket.connect(this.controlAddress);
+				commandSocket.send(zmq.ZMQ.PROXY_TERMINATE);
+			}
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running.get();
+	}
+
+	@Override
+	public void destroy() {
+		if (!this.proxyExecutorExplicitlySet) {
+			((ExecutorService) this.proxyExecutor).shutdown();
+		}
+	}
+
+	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;
+		}
+	}
+
+	public enum Type {
+
+		SUB_PUB(SocketType.XSUB, SocketType.XPUB),
+
+		PULL_PUSH(SocketType.PULL, SocketType.PUSH),
+
+		ROUTER_DEALER(SocketType.ROUTER, SocketType.DEALER);
+
+		private final SocketType frontendSocketType;
+
+		private final SocketType backendSocketType;
+
+		Type(SocketType frontendSocketType, SocketType backendSocketType) {
+			this.frontendSocketType = frontendSocketType;
+			this.backendSocketType = backendSocketType;
+		}
+
+		public SocketType getFrontendSocketType() {
+			return this.frontendSocketType;
+		}
+
+		public SocketType getBackendSocketType() {
+			return this.backendSocketType;
+		}
+
+	}
+
+}

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java

@@ -0,0 +1,321 @@
+/*
+ * 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.channel;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import org.zeromq.SocketType;
+import org.zeromq.ZContext;
+import org.zeromq.ZMQ;
+
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.mapping.BytesMessageMapper;
+import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper;
+import org.springframework.integration.zeromq.ZeroMqProxy;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.util.Assert;
+
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
+
+/**
+ * The {@link SubscribableChannel} implementation over ZeroMQ sockets.
+ * It can work in two messaging models:
+ * - {@code push-pull}, where sent messages are distributed to subscribers in a round-robin manner
+ * according a respective ZeroMQ {@link SocketType#PUSH} and {@link SocketType#PULL} socket types logic;
+ * - {@code pub-sub}, where sent messages are distributed to all subscribers;
+ * <p>
+ * This message channel can work in local mode, when a pair of ZeroMQ sockets of {@link SocketType#PAIR} type
+ * are connected between publisher (send operation) and subscriber using inter-thread transport binding.
+ * <p>
+ * In distributed mode this channel has to be connected to an externally managed ZeroMQ proxy.
+ * The {@link #setConnectUrl(String)} has to be as a standard ZeroMQ connect string, but with an extra port
+ * over the colon - representing a frontend and backend sockets pair on ZeroMQ proxy.
+ * For example: {@code tcp://localhost:6001:6002}.
+ * Another option is to provide a reference to the {@link ZeroMqProxy} instance managed in the same application:
+ * frontend and backend ports are evaluated from this proxy and the respective connection string is built from them.
+ * <p>
+ * This way sending and receiving operations on this channel are similar to interaction over a messaging broker.
+ * <p>
+ * An internal logic of this message channel implementation is based on the project Reactor using its
+ * {@link Mono}, {@link Flux} and {@link Scheduler} API for better thead model and flow control to avoid
+ * concurrency primitives for multi-publisher(subscriber) communication within the same application.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.4
+ */
+public class ZeroMqChannel extends AbstractMessageChannel implements SubscribableChannel {
+
+	public static final Duration DEFAULT_CONSUME_DELAY = Duration.ofSeconds(1);
+
+	private final Map<MessageHandler, Disposable> subscribers = new HashMap<>();
+
+	private final Scheduler publisherScheduler = Schedulers.newSingle("publisherScheduler");
+
+	private final Scheduler subscriberScheduler = Schedulers.newSingle("subscriberScheduler");
+
+	private final ZContext context;
+
+	private final boolean pubSub;
+
+	private final Mono<ZMQ.Socket> sendSocket;
+
+	private final Mono<ZMQ.Socket> subscribeSocket;
+
+	private final Flux<? extends Message<?>> subscriberData;
+
+	private Duration consumeDelay = DEFAULT_CONSUME_DELAY;
+
+	private BytesMessageMapper messageMapper = new EmbeddedJsonHeadersMessageMapper();
+
+	private Consumer<ZMQ.Socket> sendSocketConfigurer = (socket) -> { };
+
+	private Consumer<ZMQ.Socket> subscribeSocketConfigurer = (socket) -> { };
+
+	@Nullable
+	private ZeroMqProxy zeroMqProxy;
+
+	@Nullable
+	private volatile String connectSendUrl;
+
+	@Nullable
+	private volatile String connectSubscribeUrl;
+
+	@Nullable
+	private volatile Disposable subscriberDataDisposable;
+
+	private volatile boolean initialized;
+
+	public ZeroMqChannel(ZContext context) {
+		this(context, false);
+	}
+
+	public ZeroMqChannel(ZContext context, boolean pubSub) {
+		Assert.notNull(context, "'context' must not be null");
+		this.context = context;
+		this.pubSub = pubSub;
+
+		Supplier<String> localPairConnection = () -> "inproc://" + getComponentName() + ".pair";
+
+		Mono<?> proxyMono =
+				Mono.defer(() -> {
+					if (this.zeroMqProxy != null) {
+						return Mono.fromCallable(() -> this.zeroMqProxy.getBackendPort())
+								.filter((port) -> port > 0)
+								.repeatWhenEmpty(100, // NOSONAR
+										(repeat) -> repeat.delayElements(Duration.ofMillis(100))) // NOSONAR
+								.doOnNext((port) ->
+										setConnectUrl("tcp://localhost:" + this.zeroMqProxy.getFrontendPort() +
+												':' + this.zeroMqProxy.getBackendPort()))
+								.doOnError((error) ->
+										logger.error("The provided '"
+												+ this.zeroMqProxy + "' has not been started", error));
+					}
+					else {
+						return Mono.empty();
+					}
+				})
+						.cache();
+
+		this.sendSocket =
+				proxyMono
+						.publishOn(this.publisherScheduler)
+						.then(Mono.fromCallable(() ->
+								this.context.createSocket(
+										this.connectSendUrl == null
+												? SocketType.PAIR
+												: (this.pubSub ? SocketType.XPUB : SocketType.PUSH))
+						))
+						.doOnNext(this.sendSocketConfigurer)
+						.doOnNext((socket) ->
+								socket.connect(this.connectSendUrl != null
+										? this.connectSendUrl
+										: localPairConnection.get()))
+						.delayUntil((socket) ->
+								(this.pubSub && this.connectSendUrl != null)
+										? Mono.just(socket).map(ZMQ.Socket::recv)
+										: Mono.empty())
+						.cache()
+						.publishOn(this.publisherScheduler);
+
+		this.subscribeSocket =
+				proxyMono
+						.publishOn(this.subscriberScheduler)
+						.then(Mono.fromCallable(() ->
+								this.context.createSocket(
+										this.connectSubscribeUrl == null
+												? SocketType.PAIR
+												: (this.pubSub ? SocketType.SUB : SocketType.PULL))))
+						.doOnNext(this.subscribeSocketConfigurer)
+						.doOnNext((socket) -> {
+							if (this.connectSubscribeUrl != null) {
+								socket.connect(this.connectSubscribeUrl);
+								if (this.pubSub) {
+									socket.subscribe(ZMQ.SUBSCRIPTION_ALL);
+								}
+							}
+							else {
+								socket.bind(localPairConnection.get());
+							}
+						})
+						.cache()
+						.publishOn(this.subscriberScheduler);
+
+		Flux<? extends Message<?>> receiveData =
+				this.subscribeSocket
+						.flatMap((socket) -> {
+							if (this.initialized) {
+								byte[] data = socket.recv(ZMQ.NOBLOCK);
+								if (data != null) {
+									return Mono.just(data);
+								}
+							}
+							return Mono.empty();
+						})
+						.publishOn(Schedulers.parallel())
+						.map(this.messageMapper::toMessage)
+						.doOnError((error) -> logger.error("Error processing ZeroMQ message", error))
+						.repeatWhenEmpty((repeat) ->
+								this.initialized
+										? repeat.delayElements(this.consumeDelay)
+										: repeat)
+						.repeat(() -> this.initialized);
+
+		if (this.pubSub) {
+			receiveData = receiveData.publish()
+					.autoConnect(1, (disposable) -> this.subscriberDataDisposable = disposable);
+		}
+
+		this.subscriberData = receiveData;
+
+	}
+
+	/**
+	 * Configure a connection to the ZeroMQ proxy with the pair of ports over colon
+	 * for proxy frontend and backend sockets. Mutually exclusive with the {@link #setZeroMqProxy(ZeroMqProxy)}.
+	 * @param connectUrl the connection string in format {@code PROTOCOL://HOST:FRONTEND_PORT:BACKEND_PORT},
+	 *                    e.g. {@code tcp://localhost:6001:6002}
+	 */
+	public void setConnectUrl(@Nullable String connectUrl) {
+		if (connectUrl != null) {
+			this.connectSendUrl = connectUrl.substring(0, connectUrl.lastIndexOf(':'));
+			this.connectSubscribeUrl =
+					this.connectSendUrl.substring(0, this.connectSendUrl.lastIndexOf(':'))
+							+ connectUrl.substring(connectUrl.lastIndexOf(':'));
+		}
+	}
+
+	/**
+	 * Specify a reference to a {@link ZeroMqProxy} instance in the same application
+	 * to rely on its ports configuration and make a natural lifecycle dependency without guessing
+	 * when the proxy is started. Mutually exclusive with the {@link #setConnectUrl(String)}.
+	 * @param zeroMqProxy the {@link ZeroMqProxy} instance to use
+	 */
+	public void setZeroMqProxy(@Nullable ZeroMqProxy zeroMqProxy) {
+		this.zeroMqProxy = zeroMqProxy;
+	}
+
+	public void setConsumeDelay(Duration consumeDelay) {
+		Assert.notNull(consumeDelay, "'consumeDelay' must not be null");
+		this.consumeDelay = consumeDelay;
+	}
+
+	public void setMessageMapper(BytesMessageMapper messageMapper) {
+		Assert.notNull(messageMapper, "'messageMapper' must not be null");
+		this.messageMapper = messageMapper;
+	}
+
+	public void setSendSocketConfigurer(Consumer<ZMQ.Socket> sendSocketConfigurer) {
+		Assert.notNull(sendSocketConfigurer, "'sendSocketConfigurer' must not be null");
+		this.sendSocketConfigurer = sendSocketConfigurer;
+	}
+
+	public void setSubscribeSocketConfigurer(Consumer<ZMQ.Socket> subscribeSocketConfigurer) {
+		Assert.notNull(subscribeSocketConfigurer, "'subscribeSocketConfigurer' must not be null");
+		this.subscribeSocketConfigurer = subscribeSocketConfigurer;
+	}
+
+	@Override
+	protected void onInit() {
+		Assert.state(this.zeroMqProxy == null || this.connectSendUrl == null,
+				"A 'zeroMqProxy' or 'connectUrl' can be provided (or none), but not both.");
+		super.onInit();
+		this.sendSocket.subscribe();
+		this.initialized = true;
+	}
+
+	@Override
+	protected boolean doSend(Message<?> message, long timeout) {
+		Assert.state(this.initialized, "the channel is not initialized yet or already destroyed");
+
+		byte[] data = this.messageMapper.fromMessage(message);
+		Assert.state(data != null, () -> "The '" + this.messageMapper + "' returned null for '" + message + '\'');
+
+		Mono<Boolean> sendMono = this.sendSocket.map((socket) -> socket.send(data));
+		Boolean sent =
+				timeout > 0
+						? sendMono.block(Duration.ofMillis(timeout))
+						: sendMono.block();
+
+		return Boolean.TRUE.equals(sent);
+	}
+
+	@Override
+	public boolean subscribe(MessageHandler handler) {
+		Assert.state(this.initialized, "the channel is not initialized yet or already destroyed");
+		this.subscribers.computeIfAbsent(handler, (key) -> this.subscriberData.subscribe(handler::handleMessage));
+		return true;
+	}
+
+	@Override
+	public boolean unsubscribe(MessageHandler handler) {
+		Disposable disposable = this.subscribers.remove(handler);
+		if (disposable != null) {
+			disposable.dispose();
+			return true;
+		}
+		return false;
+	}
+
+	@Override
+	public void destroy() {
+		this.initialized = false;
+		super.destroy();
+		this.sendSocket.doOnNext(ZMQ.Socket::close).block();
+		this.publisherScheduler.dispose();
+		HashSet<MessageHandler> handlersCopy = new HashSet<>(this.subscribers.keySet());
+		handlersCopy.forEach(this::unsubscribe);
+		this.subscribeSocket.doOnNext(ZMQ.Socket::close).block();
+		this.subscriberScheduler.dispose();
+		if (this.subscriberDataDisposable != null) {
+			this.subscriberDataDisposable.dispose();
+		}
+	}
+
+}

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes for message channels support over ZeroMQ.
+ */
+package org.springframework.integration.zeromq.channel;

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides common classes for supporting ZeroMQ components.
+ */
+package org.springframework.integration.zeromq;

spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java

@@ -0,0 +1,193 @@
+/*
+ * 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.channel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import java.time.Duration;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Test;
+import org.zeromq.SocketType;
+import org.zeromq.ZContext;
+import org.zeromq.ZMQ;
+
+import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper;
+import org.springframework.integration.zeromq.ZeroMqProxy;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.4
+ */
+public class ZeroMqChannelTests {
+
+	private static final ZContext CONTEXT = new ZContext();
+
+	@AfterAll
+	static void tearDown() {
+		CONTEXT.close();
+	}
+
+	@Test
+	void testSimpleSendAndReceive() throws InterruptedException {
+		ZeroMqChannel channel = new ZeroMqChannel(CONTEXT);
+		channel.setBeanName("testChannel1");
+		channel.setConsumeDelay(Duration.ofMillis(10));
+		channel.afterPropertiesSet();
+
+		BlockingQueue<Message<?>> received = new LinkedBlockingQueue<>();
+
+		channel.subscribe(received::offer);
+
+		assertThat(channel.send(new GenericMessage<>("test1"), 1000)).isTrue();
+		assertThat(channel.send(new GenericMessage<>("test2"), 500)).isTrue();
+
+		Message<?> message = received.poll(10, TimeUnit.SECONDS);
+		assertThat(message).isNotNull().extracting(Message::getPayload).isEqualTo("test1");
+		message = received.poll(10, TimeUnit.SECONDS);
+		assertThat(message).isNotNull().extracting(Message::getPayload).isEqualTo("test2");
+
+		// Ensure that second subscriber doesn't make it as pub-sub
+		channel.subscribe(received::offer);
+
+		assertThat(channel.send(new GenericMessage<>("test3"))).isTrue();
+
+		message = received.poll(10, TimeUnit.SECONDS);
+		assertThat(message).isNotNull().extracting(Message::getPayload).isEqualTo("test3");
+		assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull();
+
+		channel.destroy();
+	}
+
+	@Test
+	void testPubSubLocal() throws InterruptedException {
+		ZeroMqChannel channel = new ZeroMqChannel(CONTEXT, true);
+		channel.setBeanName("testChannel2");
+		channel.setConsumeDelay(Duration.ofMillis(10));
+		channel.afterPropertiesSet();
+
+		BlockingQueue<Message<?>> received = new LinkedBlockingQueue<>();
+
+		channel.subscribe(received::offer);
+		channel.subscribe(received::offer);
+
+		GenericMessage<String> testMessage = new GenericMessage<>("test1");
+		assertThat(channel.send(testMessage)).isTrue();
+
+		Message<?> message = received.poll(10, TimeUnit.SECONDS);
+		assertThat(message).isNotNull().isEqualTo(testMessage);
+		message = received.poll(10, TimeUnit.SECONDS);
+		assertThat(message).isNotNull().isEqualTo(testMessage);
+
+		channel.destroy();
+	}
+
+	@Test
+	void testPushPullBind() throws InterruptedException {
+		ZeroMqProxy proxy = new ZeroMqProxy(CONTEXT);
+		proxy.setBeanName("pullPushProxy");
+		proxy.setExposeCaptureSocket(true);
+		proxy.afterPropertiesSet();
+		proxy.start();
+
+		await().until(() -> proxy.getBackendPort() > 0);
+
+		ZMQ.Socket captureSocket = CONTEXT.createSocket(SocketType.SUB);
+		captureSocket.connect(proxy.getCaptureAddress());
+		captureSocket.subscribe(ZMQ.SUBSCRIPTION_ALL);
+
+		ZeroMqChannel channel = new ZeroMqChannel(CONTEXT);
+		channel.setConnectUrl("tcp://localhost:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort());
+		channel.setBeanName("testChannel3");
+		channel.setConsumeDelay(Duration.ofMillis(10));
+		channel.afterPropertiesSet();
+
+		BlockingQueue<Message<?>> received = new LinkedBlockingQueue<>();
+
+		channel.subscribe(received::offer);
+		channel.subscribe(received::offer);
+
+		GenericMessage<String> testMessage = new GenericMessage<>("test1");
+		assertThat(channel.send(testMessage)).isTrue();
+
+		Message<?> message = received.poll(10, TimeUnit.SECONDS);
+		assertThat(message).isNotNull().isEqualTo(testMessage);
+		assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull();
+
+		channel.destroy();
+
+		byte[] recv = captureSocket.recv();
+		assertThat(recv).isNotNull();
+		Message<?> capturedMessage = new EmbeddedJsonHeadersMessageMapper().toMessage(recv);
+		assertThat(capturedMessage).isEqualTo(testMessage);
+		captureSocket.close();
+
+		proxy.stop();
+	}
+
+
+	@Test
+	void testPubSubBind() throws InterruptedException {
+		ZeroMqProxy proxy = new ZeroMqProxy(CONTEXT, ZeroMqProxy.Type.SUB_PUB);
+		proxy.setBeanName("subPubProxy");
+		proxy.afterPropertiesSet();
+		proxy.start();
+
+		ZeroMqChannel channel = new ZeroMqChannel(CONTEXT, true);
+		channel.setZeroMqProxy(proxy);
+		channel.setBeanName("testChannel4");
+		channel.setConsumeDelay(Duration.ofMillis(10));
+		channel.afterPropertiesSet();
+
+		BlockingQueue<Message<?>> received = new LinkedBlockingQueue<>();
+
+		channel.subscribe(received::offer);
+		channel.subscribe(received::offer);
+
+		await().until(() -> proxy.getBackendPort() > 0);
+
+		ZeroMqChannel channel2 = new ZeroMqChannel(CONTEXT, true);
+		channel2.setConnectUrl("tcp://localhost:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort());
+		channel2.setBeanName("testChannel5");
+		channel.setConsumeDelay(Duration.ofMillis(10));
+		channel2.afterPropertiesSet();
+
+		channel.subscribe(received::offer);
+
+		GenericMessage<String> testMessage = new GenericMessage<>("test1");
+		assertThat(channel.send(testMessage)).isTrue();
+
+		Message<?> message = received.poll(10, TimeUnit.SECONDS);
+		assertThat(message).isNotNull().isEqualTo(testMessage);
+		message = received.poll(10, TimeUnit.SECONDS);
+		assertThat(message).isNotNull().isEqualTo(testMessage);
+		message = received.poll(10, TimeUnit.SECONDS);
+		assertThat(message).isNotNull().isEqualTo(testMessage);
+		assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull();
+
+		channel.destroy();
+		proxy.stop();
+	}
+
+}

spring-integration-zeromq/src/test/resources/log4j2-test.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN">
+	<Appenders>
+		<Console name="STDOUT" target="SYSTEM_OUT">
+			<PatternLayout pattern="%d %p [%t] [%c] - %m%n" />
+		</Console>
+	</Appenders>
+	<Loggers>
+		<Logger name="org.springframework.integration" level="warn"/>
+		<Logger name="org.springframework.integration.zeromq" level="warn"/>
+		<Root level="warn">
+			<AppenderRef ref="STDOUT" />
+		</Root>
+	</Loggers>
+</Configuration>