From 0ea9d700deb5496c4ea6a3b97753d41762fea74f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 22 Apr 2021 11:59:14 -0400 Subject: [PATCH] Fix ZeroMQ components initialization SO: https://stackoverflow.com/questions/67214907/zeromq-with-spring-spring-integration-zeromq The `Mono` is created in several places in ZeroMQ components from their constructors. That leads to the reactive stream to be configured just after ctor, which will ignore any changes to the options which are used from that `Mono` definition. For example this code `.doOnNext(this.sendSocketConfigurer)` is done once during reactive stream definition. * Fix all the ZeroMQ components to defer usage of the options which could be changed after ctor initialization * Cover affected option changes in the tests **Cherry-pick to `5.4.x`** --- .../zeromq/channel/ZeroMqChannel.java | 8 ++--- .../zeromq/outbound/ZeroMqMessageHandler.java | 4 +-- .../zeromq/channel/ZeroMqChannelTests.java | 30 ++++++++++++++++++- .../inbound/ZeroMqMessageProducerTests.java | 10 ++++++- .../outbound/ZeroMqMessageHandlerTests.java | 11 +++++++ 5 files changed, 55 insertions(+), 8 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 f5ae7370d7..0d35979a3f 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 @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -168,7 +168,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl ? SocketType.PAIR : (this.pubSub ? SocketType.PUB : SocketType.PUSH)) )) - .doOnNext(this.sendSocketConfigurer) + .doOnNext((socket) -> this.sendSocketConfigurer.accept(socket)) .doOnNext((socket) -> socket.connect(this.connectSendUrl != null ? this.connectSendUrl @@ -184,7 +184,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl this.connectSubscribeUrl == null ? SocketType.PAIR : (this.pubSub ? SocketType.SUB : SocketType.PULL)))) - .doOnNext(this.subscribeSocketConfigurer) + .doOnNext((socket) -> this.subscribeSocketConfigurer.accept(socket)) .doOnNext((socket) -> { if (this.connectSubscribeUrl != null) { if (this.pubSub) { @@ -213,7 +213,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl return Mono.empty(); }) .publishOn(Schedulers.parallel()) - .map(this.messageMapper::toMessage) + .map((data) -> this.messageMapper.toMessage(data)) .doOnError((error) -> logger.error(error, () -> "Error processing ZeroMQ message in the " + this)) .repeatWhenEmpty((repeat) -> diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java index 7e55ac4a00..5a9e38389c 100644 --- a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -104,7 +104,7 @@ public class ZeroMqMessageHandler extends AbstractReactiveMessageHandler { this.socketMono = Mono.just(context.createSocket(socketType)) .publishOn(this.publisherScheduler) - .doOnNext(this.socketConfigurer) + .doOnNext((socket) -> this.socketConfigurer.accept(socket)) .doOnNext((socket) -> socket.connect(connectUrl)) .cache() .publishOn(this.publisherScheduler); 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 66a22ce0d3..612885195f 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 @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -20,9 +20,11 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; import java.time.Duration; +import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; @@ -31,10 +33,13 @@ import org.zeromq.ZContext; import org.zeromq.ZMQ; import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; +import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.zeromq.ZeroMqProxy; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; +import reactor.core.publisher.Mono; + /** * @author Artem Bilan * @@ -54,8 +59,29 @@ public class ZeroMqChannelTests { ZeroMqChannel channel = new ZeroMqChannel(CONTEXT); channel.setBeanName("testChannel1"); channel.setConsumeDelay(Duration.ofMillis(10)); + channel.setSendSocketConfigurer(socket -> socket.setZapDomain("global")); + channel.setSubscribeSocketConfigurer(socket -> socket.setZapDomain("local")); + AtomicBoolean customMessageMapperCalled = new AtomicBoolean(); + channel.setMessageMapper(new EmbeddedJsonHeadersMessageMapper() { + + @Override public Message toMessage(byte[] bytes, Map headers) { + customMessageMapperCalled.set(true); + return super.toMessage(bytes, headers); + } + + }); channel.afterPropertiesSet(); + @SuppressWarnings("unchecked") + Mono sendSocketMono = TestUtils.getPropertyValue(channel, "sendSocket", Mono.class); + ZMQ.Socket sendSocket = sendSocketMono.block(Duration.ofSeconds(10)); + assertThat(sendSocket.getZapDomain()).isEqualTo("global"); + + @SuppressWarnings("unchecked") + Mono subscribeSocketMono = TestUtils.getPropertyValue(channel, "subscribeSocket", Mono.class); + ZMQ.Socket subscribeSocket = subscribeSocketMono.block(Duration.ofSeconds(10)); + assertThat(subscribeSocket.getZapDomain()).isEqualTo("local"); + BlockingQueue> received = new LinkedBlockingQueue<>(); channel.subscribe(received::offer); @@ -78,6 +104,8 @@ public class ZeroMqChannelTests { assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull(); channel.destroy(); + + assertThat(customMessageMapperCalled.get()).isTrue(); } @Test diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java index d367d85926..58f7385eee 100644 --- a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -33,8 +33,10 @@ import org.zeromq.ZMsg; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.support.GenericMessage; +import reactor.core.publisher.Mono; import reactor.test.StepVerifier; /** @@ -67,9 +69,15 @@ public class ZeroMqMessageProducerTests { messageProducer.setMessageMapper((object, headers) -> new GenericMessage<>(new String(object))); messageProducer.setConsumeDelay(Duration.ofMillis(10)); messageProducer.setBeanFactory(mock(BeanFactory.class)); + messageProducer.setSocketConfigurer(s -> s.setZapDomain("global")); messageProducer.afterPropertiesSet(); messageProducer.start(); + @SuppressWarnings("unchecked") + Mono socketMono = TestUtils.getPropertyValue(messageProducer, "socketMono", Mono.class); + ZMQ.Socket socketInUse = socketMono.block(Duration.ofSeconds(10)); + assertThat(socketInUse.getZapDomain()).isEqualTo("global"); + ZMQ.Socket socket = CONTEXT.createSocket(SocketType.PAIR); await().until(() -> messageProducer.getBoundPort() > 0); diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java index a77a98d509..8793c5b26c 100644 --- a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java @@ -21,6 +21,8 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; import static org.mockito.Mockito.mock; +import java.time.Duration; + import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; import org.zeromq.SocketType; @@ -32,11 +34,14 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; +import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.zeromq.ZeroMqProxy; import org.springframework.messaging.Message; import org.springframework.messaging.converter.ByteArrayMessageConverter; import org.springframework.messaging.support.GenericMessage; +import reactor.core.publisher.Mono; + /** * @author Artem Bilan * @@ -59,8 +64,14 @@ public class ZeroMqMessageHandlerTests { ZeroMqMessageHandler messageHandler = new ZeroMqMessageHandler(CONTEXT, socketAddress); messageHandler.setBeanFactory(mock(BeanFactory.class)); + messageHandler.setSocketConfigurer(s -> s.setZapDomain("global")); messageHandler.afterPropertiesSet(); + @SuppressWarnings("unchecked") + Mono socketMono = TestUtils.getPropertyValue(messageHandler, "socketMono", Mono.class); + ZMQ.Socket socketInUse = socketMono.block(Duration.ofSeconds(10)); + assertThat(socketInUse.getZapDomain()).isEqualTo("global"); + Message testMessage = new GenericMessage<>("test"); messageHandler.handleMessage(testMessage).subscribe();