From e70e15c310e075ceb3cade1d1da541b9de6435cc Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 29 Apr 2021 16:09:49 -0400 Subject: [PATCH] Add ZeroMqChannel test with Curve Auth Related to https://stackoverflow.com/questions/67214907/zeromq-with-spring-spring-integration-zeromq * Fix typos in `ZeroMqChannel` JavaDocs --- .../zeromq/channel/ZeroMqChannel.java | 6 +- .../zeromq/channel/ZeroMqChannelTests.java | 67 +++++++++++++++++++ 2 files changed, 71 insertions(+), 2 deletions(-) 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(); + } + }