diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java
index 0d35979a3f..33000180c2 100644
--- a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java
+++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java
@@ -48,7 +48,7 @@ import reactor.core.scheduler.Schedulers;
* 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;
+ * - {@code pub-sub}, where sent messages are distributed to all subscribers.
*
* 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.
@@ -63,7 +63,7 @@ import reactor.core.scheduler.Schedulers;
* This way sending and receiving operations on this channel are similar to interaction over a messaging broker.
*
* 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
+ * {@link Mono}, {@link Flux} and {@link Scheduler} API for better thread model and flow control to avoid
* concurrency primitives for multi-publisher(subscriber) communication within the same application.
*
* @author Artem Bilan
@@ -278,6 +278,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
/**
* The {@link Consumer} callback to configure a publishing socket.
+ * The send socket is connected to the frontend socket of ZeroMQ proxy (if any).
* @param sendSocketConfigurer the {@link Consumer} to use.
*/
public void setSendSocketConfigurer(Consumer sendSocketConfigurer) {
@@ -287,6 +288,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
/**
* The {@link Consumer} callback to configure a consuming socket.
+ * The subscribe socket is connected to the backend socket of ZeroMQ proxy (if any).
* @param subscribeSocketConfigurer the {@link Consumer} to use.
*/
public void setSubscribeSocketConfigurer(Consumer subscribeSocketConfigurer) {
diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java
index 612885195f..e7b97d3e3a 100644
--- a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java
+++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java
@@ -29,6 +29,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Test;
import org.zeromq.SocketType;
+import org.zeromq.ZAuth;
+import org.zeromq.ZCert;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
@@ -222,4 +224,69 @@ public class ZeroMqChannelTests {
proxy.stop();
}
+ @Test
+ void testPubSubWithCurve() throws InterruptedException {
+ new ZAuth(CONTEXT).configureCurve(ZAuth.CURVE_ALLOW_ANY);
+
+ ZMQ.Curve.KeyPair frontendKeyPair = ZMQ.Curve.generateKeyPair();
+ ZMQ.Curve.KeyPair backendKeyPair = ZMQ.Curve.generateKeyPair();
+
+ ZeroMqProxy proxy = new ZeroMqProxy(CONTEXT, ZeroMqProxy.Type.SUB_PUB);
+ proxy.setBeanName("subPubCurveProxy");
+ proxy.setFrontendSocketConfigurer(socket -> {
+ socket.setZAPDomain("global".getBytes());
+ socket.setCurveServer(true);
+ socket.setCurvePublicKey(frontendKeyPair.publicKey.getBytes());
+ socket.setCurveSecretKey(frontendKeyPair.secretKey.getBytes());
+ });
+ proxy.setBackendSocketConfigurer(socket -> {
+ socket.setZAPDomain("global".getBytes());
+ socket.setCurveServer(true);
+ socket.setCurvePublicKey(backendKeyPair.publicKey.getBytes());
+ socket.setCurveSecretKey(backendKeyPair.secretKey.getBytes());
+ });
+ proxy.afterPropertiesSet();
+ proxy.start();
+
+ ZeroMqChannel channel = new ZeroMqChannel(CONTEXT, true);
+ channel.setZeroMqProxy(proxy);
+ channel.setBeanName("testChannelWithCurve");
+ channel.setSendSocketConfigurer(socket -> {
+ ZCert clientCert = new ZCert();
+ socket.setCurvePublicKey(clientCert.getPublicKey());
+ socket.setCurveSecretKey(clientCert.getSecretKey());
+ socket.setCurveServerKey(frontendKeyPair.publicKey.getBytes());
+ });
+ channel.setSubscribeSocketConfigurer(socket -> {
+ ZCert clientCert = new ZCert();
+ socket.setCurvePublicKey(clientCert.getPublicKey());
+ socket.setCurveSecretKey(clientCert.getSecretKey());
+ socket.setCurveServerKey(backendKeyPair.publicKey.getBytes());
+ }
+ );
+ channel.setConsumeDelay(Duration.ofMillis(10));
+ channel.afterPropertiesSet();
+
+ BlockingQueue> received = new LinkedBlockingQueue<>();
+
+ channel.subscribe(received::offer);
+ channel.subscribe(received::offer);
+
+ await().until(() -> proxy.getBackendPort() > 0);
+
+ // Give it some time to connect and subscribe
+ Thread.sleep(1000);
+
+ GenericMessage 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();
+ proxy.stop();
+ }
+
}