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`**

Author: artembilan

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 @@ private Mono<ZMQ.Socket> prepareSendSocketMono(Supplier<String> localPairConnect
 										? 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 @@ private Mono<ZMQ.Socket> prepareSubscribeSocketMono(Supplier<String> localPairCo
 								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 @@ private Flux<? extends Message<?>> prepareSubscriberDataFlux() {
 							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) ->

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 ZeroMqMessageHandler(ZContext context, String connectUrl, SocketType sock
 		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);

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.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.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 @@ void testSimpleSendAndReceive() throws InterruptedException {
 		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<String, Object> headers) {
+				customMessageMapperCalled.set(true);
+				return super.toMessage(bytes, headers);
+			}
+
+		});
 		channel.afterPropertiesSet();
 
+		@SuppressWarnings("unchecked")
+		Mono<ZMQ.Socket> sendSocketMono = TestUtils.getPropertyValue(channel, "sendSocket", Mono.class);
+		ZMQ.Socket sendSocket = sendSocketMono.block(Duration.ofSeconds(10));
+		assertThat(sendSocket.getZapDomain()).isEqualTo("global");
+
+		@SuppressWarnings("unchecked")
+		Mono<ZMQ.Socket> subscribeSocketMono = TestUtils.getPropertyValue(channel, "subscribeSocket", Mono.class);
+		ZMQ.Socket subscribeSocket = subscribeSocketMono.block(Duration.ofSeconds(10));
+		assertThat(subscribeSocket.getZapDomain()).isEqualTo("local");
+
 		BlockingQueue<Message<?>> received = new LinkedBlockingQueue<>();
 
 		channel.subscribe(received::offer);
@@ -78,6 +104,8 @@ void testSimpleSendAndReceive() throws InterruptedException {
 		assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull();
 
 		channel.destroy();
+
+		assertThat(customMessageMapperCalled.get()).isTrue();
 	}
 
 	@Test

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.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 @@ void testMessageProducerForPair() {
 		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<ZMQ.Socket> 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);

spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java

@@ -21,6 +21,8 @@
 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.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 @@ void testMessageHandlerForPair() {
 
 		ZeroMqMessageHandler messageHandler = new ZeroMqMessageHandler(CONTEXT, socketAddress);
 		messageHandler.setBeanFactory(mock(BeanFactory.class));
+		messageHandler.setSocketConfigurer(s -> s.setZapDomain("global"));
 		messageHandler.afterPropertiesSet();
 
+		@SuppressWarnings("unchecked")
+		Mono<ZMQ.Socket> 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();