GH-9228: ZeroMQ - provide binding to ZeroMqMessageHandler

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2024 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.
@@ -17,7 +17,7 @@
 package org.springframework.integration.zeromq;
 
 /**
- * The message headers constants to repsent ZeroMq message attributes.
+ * The message headers constants to represent ZeroMq message attributes.
  *
  * @author Artem Bilan
  *

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2020-2024 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 org.zeromq.ZMQ;
+
+/**
+ * Module that wraps common methods of ZeroMq integration classes
+ *
+ * @author Alessio Matricardi
+ *
+ * @since 6.4
+ *
+ */
+public final class ZeroMqUtils {
+
+	/**
+	 * Bind the ZeroMq socket to the given port over the TCP transport protocol.
+	 * @param socket the ZeroMq socket
+	 * @param port the port to bind ZeroMq socket to over TCP. If equal to 0, the socket will bind to a random port.
+	 * @return the effectively bound port
+	 */
+	public 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;
+		}
+	}
+
+	private ZeroMqUtils() {
+	}
+
+}

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMq.java

@@ -25,6 +25,7 @@
  * Factory class for ZeroMq components DSL.
  *
  * @author Artem Bilan
+ * @author Alessio Matricardi
  *
  * @since 5.4
  */
@@ -58,6 +59,17 @@ public static ZeroMqMessageHandlerSpec outboundChannelAdapter(ZContext context,
 		return outboundChannelAdapter(context, () -> connectUrl);
 	}
 
+	/**
+	 * Create an instance of {@link ZeroMqMessageHandlerSpec} for the provided {@link ZContext} and binding port.
+	 * @param context the {@link ZContext} to use.
+	 * @param port the port to bind ZeroMq socket to over TCP.
+	 * @return the spec.
+	 * @since 6.4
+	 */
+	public static ZeroMqMessageHandlerSpec outboundChannelAdapter(ZContext context, int port) {
+		return new ZeroMqMessageHandlerSpec(context, port);
+	}
+
 	/**
 	 * Create an instance of {@link ZeroMqMessageHandlerSpec} for the provided {@link ZContext}
 	 * and connection URL supplier.
@@ -84,6 +96,43 @@ public static ZeroMqMessageHandlerSpec outboundChannelAdapter(ZContext context,
 		return new ZeroMqMessageHandlerSpec(context, connectUrl, socketType);
 	}
 
+	/**
+	 * Create an instance of {@link ZeroMqMessageHandlerSpec} for the provided {@link ZContext}.
+	 * The created socket will be bound to a random port.
+	 * @param context the {@link ZContext} to use.
+	 * @return the spec.
+	 * @since 6.4
+	 */
+	public static ZeroMqMessageHandlerSpec outboundChannelAdapter(ZContext context) {
+		return new ZeroMqMessageHandlerSpec(context);
+	}
+
+	/**
+	 * Create an instance of {@link ZeroMqMessageHandlerSpec} for the provided {@link ZContext} and {@link SocketType}.
+	 * The created socket will be bound to a random port.
+	 * @param context the {@link ZContext} to use.
+	 * @param socketType the {@link SocketType} for ZeroMq socket.
+	 * @return the spec.
+	 * @since 6.4
+	 */
+	public static ZeroMqMessageHandlerSpec outboundChannelAdapter(ZContext context, SocketType socketType) {
+		return new ZeroMqMessageHandlerSpec(context, socketType);
+	}
+
+	/**
+	 * Create an instance of {@link ZeroMqMessageHandlerSpec} for the provided {@link ZContext}, binding port
+	 * and {@link SocketType}.
+	 * @param context the {@link ZContext} to use.
+	 * @param port the port to bind ZeroMq socket to over TCP.
+	 * @param socketType the {@link SocketType} for ZeroMq socket.
+	 * @return the spec.
+	 * @since 6.4
+	 */
+	public static ZeroMqMessageHandlerSpec outboundChannelAdapter(ZContext context, int port,
+			SocketType socketType) {
+		return new ZeroMqMessageHandlerSpec(context, port, socketType);
+	}
+
 	/**
 	 * Create an instance of {@link ZeroMqMessageHandlerSpec} for the provided {@link ZContext},
 	 * connection URL supplier and {@link SocketType}.

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java

@@ -52,6 +52,26 @@ protected ZeroMqMessageHandlerSpec(ZContext context, String connectUrl) {
 		this(context, () -> connectUrl);
 	}
 
+	/**
+	 * Create an instance based on the provided {@link ZContext}.
+	 * The created socket will be bound to a random port.
+	 * @param context the {@link ZContext} to use for creating sockets.
+	 * @since 6.4
+	 */
+	protected ZeroMqMessageHandlerSpec(ZContext context) {
+		this(context, SocketType.PAIR);
+	}
+
+	/**
+	 * Create an instance based on the provided {@link ZContext} and binding port.
+	 * @param context the {@link ZContext} to use for creating sockets.
+	 * @param port the port to bind ZeroMq socket to over TCP.
+	 * @since 6.4
+	 */
+	protected ZeroMqMessageHandlerSpec(ZContext context, int port) {
+		this(context, port, SocketType.PAIR);
+	}
+
 	/**
 	 * Create an instance based on the provided {@link ZContext} and connection string supplier.
 	 * @param context the {@link ZContext} to use for creating sockets.
@@ -73,6 +93,30 @@ protected ZeroMqMessageHandlerSpec(ZContext context, String connectUrl, SocketTy
 		this(context, () -> connectUrl, socketType);
 	}
 
+	/**
+	 * Create an instance based on the provided {@link ZContext} and {@link SocketType}.
+	 * The created socket will be bound to a random port.
+	 * @param context the {@link ZContext} to use for creating sockets.
+	 * @param socketType the {@link SocketType} to use;
+	 *    only {@link SocketType#PAIR}, {@link SocketType#PUB} and {@link SocketType#PUSH} are supported.
+	 * @since 6.4
+	 */
+	protected ZeroMqMessageHandlerSpec(ZContext context, SocketType socketType) {
+		super(new ZeroMqMessageHandler(context, socketType));
+	}
+
+	/**
+	 * Create an instance based on the provided {@link ZContext}, binding port and {@link SocketType}.
+	 * @param context the {@link ZContext} to use for creating sockets.
+	 * @param port the port to bind ZeroMq socket to over TCP.
+	 * @param socketType the {@link SocketType} to use;
+	 *    only {@link SocketType#PAIR}, {@link SocketType#PUB} and {@link SocketType#PUSH} are supported.
+	 * @since 6.4
+	 */
+	protected ZeroMqMessageHandlerSpec(ZContext context, int port, SocketType socketType) {
+		super(new ZeroMqMessageHandler(context, port, socketType));
+	}
+
 	/**
 	 * Create an instance based on the provided {@link ZContext}, connection string supplier and {@link SocketType}.
 	 * @param context the {@link ZContext} to use for creating sockets.

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java

@@ -40,6 +40,7 @@
 import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
 import org.springframework.integration.support.management.IntegrationManagedResource;
 import org.springframework.integration.zeromq.ZeroMqHeaders;
+import org.springframework.integration.zeromq.ZeroMqUtils;
 import org.springframework.jmx.export.annotation.ManagedOperation;
 import org.springframework.jmx.export.annotation.ManagedResource;
 import org.springframework.lang.Nullable;
@@ -263,7 +264,7 @@ protected void doStart() {
 								socket.connect(this.connectUrl);
 							}
 							else {
-								this.bindPort.set(bindSocket(socket, this.bindPort.get()));
+								this.bindPort.set(ZeroMqUtils.bindSocket(socket, this.bindPort.get()));
 							}
 						})
 						.cache()
@@ -319,17 +320,4 @@ public void 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;
-		}
-	}
-
 }