Add ZeroMqChannel test with Curve Auth
Related to https://stackoverflow.com/questions/67214907/zeromq-with-spring-spring-integration-zeromq * Fix typos in `ZeroMqChannel` JavaDocs
This commit is contained in:
@@ -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.
|
||||
* <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.
|
||||
@@ -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.
|
||||
* <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
|
||||
* {@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<ZMQ.Socket> 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<ZMQ.Socket> subscribeSocketConfigurer) {
|
||||
|
||||
@@ -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<Message<?>> 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<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();
|
||||
proxy.stop();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user