* 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

Author: artembilan

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

@@ -46,7 +46,7 @@
  * 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 tne {@code "inproc://" + beanName + ".control"} address; it can be obtained via {@link #getControlAddress()}.
+ * 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>
@@ -56,6 +56,7 @@
  * 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
  *
@@ -100,28 +101,53 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw
 
 	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 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);

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

@@ -20,6 +20,7 @@
 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;
@@ -29,6 +30,7 @@
 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;
@@ -55,7 +57,10 @@
  * 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}.
- * This way a sending and receiving operations on this channel are similar to interaction over a messaging broker.
+ * 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 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
@@ -67,6 +72,8 @@
  */
 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");
@@ -83,8 +90,17 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
 
 	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;
 
@@ -107,30 +123,51 @@ public ZeroMqChannel(ZContext context, boolean pubSub) {
 
 		Supplier<String> localPairConnection = () -> "inproc://" + getComponentName() + ".pair";
 
+		Mono<?> proxyMono =
+				Mono.defer(() -> {
+					if (this.zeroMqProxy != null) {
+						return Mono.just(this.zeroMqProxy.getBackendPort())
+								.filter((port) -> port > 0)
+								.repeatWhenEmpty((repeat) -> repeat.delayElements(Duration.ofMillis(100))) // NOSONAR
+								.doOnNext((port) ->
+										setConnectUrl("tcp://localhost:" + this.zeroMqProxy.getFrontendPort() +
+												':' + this.zeroMqProxy.getBackendPort()));
+					}
+					else {
+						return Mono.empty();
+					}
+				})
+						.cache();
+
 		this.sendSocket =
-				Mono.fromCallable(() ->
-						this.context.createSocket(
-								this.connectSendUrl == null
-										? SocketType.PAIR
-										: (this.pubSub ? SocketType.XPUB : SocketType.PUSH))
-				)
+				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())
+						.delayUntil((socket) ->
+								(this.pubSub && this.connectSendUrl != null)
+										? Mono.just(socket).map(ZMQ.Socket::recv)
+										: Mono.empty())
 						.cache();
 
 		this.subscribeSocket =
-				Mono.fromCallable(() ->
-						this.context.createSocket(
-								this.connectSubscribeUrl == null
-										? SocketType.PAIR
-										: (this.pubSub ? SocketType.SUB : SocketType.PULL)))
+				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);
@@ -160,7 +197,7 @@ public ZeroMqChannel(ZContext context, boolean pubSub) {
 						.doOnError((error) -> logger.error("Error processing ZeroMQ message", error))
 						.repeatWhenEmpty((repeat) ->
 								this.initialized
-										? repeat.delayElements(Duration.ofMillis(100))
+										? repeat.delayElements(this.consumeDelay)
 										: repeat)
 						.repeat(() -> this.initialized);
 
@@ -173,6 +210,12 @@ public ZeroMqChannel(ZContext context, boolean pubSub) {
 
 	}
 
+	/**
+	 * 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(':'));
@@ -182,13 +225,40 @@ public void setConnectUrl(@Nullable String connectUrl) {
 		}
 	}
 
+	/**
+	 * 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,
+				"Or 'zeroMqProxy' or 'connectUrl' can be provided (or none), but not both.");
 		super.onInit();
 		this.sendSocket.subscribe();
 		this.initialized = true;

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

@@ -19,6 +19,7 @@
 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;
@@ -52,6 +53,7 @@ static void tearDown() {
 	void testSimpleSendAndReceive() throws InterruptedException {
 		ZeroMqChannel channel = new ZeroMqChannel(CONTEXT);
 		channel.setBeanName("testChannel1");
+		channel.setConsumeDelay(Duration.ofMillis(10));
 		channel.afterPropertiesSet();
 
 		BlockingQueue<Message<?>> received = new LinkedBlockingQueue<>();
@@ -82,6 +84,7 @@ void testSimpleSendAndReceive() throws InterruptedException {
 	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<>();
@@ -115,8 +118,9 @@ void testPushPullBind() throws InterruptedException {
 		captureSocket.subscribe(ZMQ.SUBSCRIPTION_ALL);
 
 		ZeroMqChannel channel = new ZeroMqChannel(CONTEXT);
-		channel.setConnectUrl("tcp://*:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort());
+		channel.setConnectUrl("tcp://localhost:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort());
 		channel.setBeanName("testChannel3");
+		channel.setConsumeDelay(Duration.ofMillis(10));
 		channel.afterPropertiesSet();
 
 		BlockingQueue<Message<?>> received = new LinkedBlockingQueue<>();
@@ -150,21 +154,23 @@ void testPubSubBind() throws InterruptedException {
 		proxy.afterPropertiesSet();
 		proxy.start();
 
-		await().until(() -> proxy.getBackendPort() > 0);
-
 		ZeroMqChannel channel = new ZeroMqChannel(CONTEXT, true);
-		channel.setConnectUrl("tcp://*:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort());
+		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://*:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort());
+		channel2.setConnectUrl("tcp://localhost:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort());
 		channel2.setBeanName("testChannel5");
+		channel.setConsumeDelay(Duration.ofMillis(10));
 		channel2.afterPropertiesSet();
 
 		channel.subscribe(received::offer);