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

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) ->

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);

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<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 @@ public class ZeroMqChannelTests {
 		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.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<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.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<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();