Simplify complexity in ZeroMqChannel constructor

* Upgrade to Spring AMQP `2.3.0-M2`

Author: artembilan

build.gradle

@@ -98,7 +98,7 @@ ext {
 	servletApiVersion = '4.0.1'
 	smackVersion = '4.3.4'
 	soapVersion = '1.4.0'
-	springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.0-SNAPSHOT'
+	springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.0-M2'
 	springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2020.0.0-M2'
 	springKafkaVersion = '2.6.0-SNAPSHOT'
 	springRetryVersion = '1.3.0'

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java

@@ -123,50 +123,74 @@ public ZeroMqChannel(ZContext context, boolean pubSub) {
 
 		Supplier<String> localPairConnection = () -> "inproc://" + getComponentName() + ".pair";
 
-		Mono<?> proxyMono = proxyMono();
-
-		this.sendSocket =
-				proxyMono.publishOn(this.publisherScheduler)
-						.then(Mono.fromCallable(() ->
-								this.context.createSocket(
-										this.connectSendUrl == null
-												? SocketType.PAIR
-												: (this.pubSub ? SocketType.XPUB : SocketType.PUSH))
-						))
-						.doOnNext(this.sendSocketConfigurer)
-						.doOnNext((socket) ->
-								socket.connect(this.connectSendUrl != null
-										? this.connectSendUrl
-										: localPairConnection.get()))
-						.delayUntil((socket) ->
-								(this.pubSub && this.connectSendUrl != null)
-										? Mono.just(socket).map(ZMQ.Socket::recv)
-										: Mono.empty())
-						.cache()
-						.publishOn(this.publisherScheduler);
-
-		this.subscribeSocket =
-				proxyMono.publishOn(this.subscriberScheduler)
-						.then(Mono.fromCallable(() ->
-								this.context.createSocket(
-										this.connectSubscribeUrl == null
-												? SocketType.PAIR
-												: (this.pubSub ? SocketType.SUB : SocketType.PULL))))
-						.doOnNext(this.subscribeSocketConfigurer)
-						.doOnNext((socket) -> {
-							if (this.connectSubscribeUrl != null) {
-								socket.connect(this.connectSubscribeUrl);
-								if (this.pubSub) {
-									socket.subscribe(ZMQ.SUBSCRIPTION_ALL);
-								}
-							}
-							else {
-								socket.bind(localPairConnection.get());
-							}
-						})
-						.cache()
-						.publishOn(this.subscriberScheduler);
+		Mono<?> proxyMono = prepareProxyMono();
+		this.sendSocket = prepareSendSocketMono(localPairConnection, proxyMono);
+		this.subscribeSocket = prepareSubscribeSocketMono(localPairConnection, proxyMono);
+		this.subscriberData = prepareSubscriberDataFlux();
+	}
+
+	private Mono<Integer> prepareProxyMono() {
+		if (this.zeroMqProxy != null) {
+			return Mono.fromCallable(() -> this.zeroMqProxy.getBackendPort())
+					.filter((proxyPort) -> proxyPort > 0)
+					.repeatWhenEmpty(100, (repeat) -> repeat.delayElements(Duration.ofMillis(100))) // NOSONAR
+					.doOnNext((proxyPort) ->
+							setConnectUrl("tcp://localhost:" + this.zeroMqProxy.getFrontendPort() +
+									':' + this.zeroMqProxy.getBackendPort()))
+					.doOnError((error) ->
+							logger.error("The provided '" + this.zeroMqProxy + "' has not been started", error))
+					.cache();
+		}
+		else {
+			return Mono.empty();
+		}
+	}
 
+	private Mono<ZMQ.Socket> prepareSendSocketMono(Supplier<String> localPairConnection, Mono<?> proxyMono) {
+		return proxyMono.publishOn(this.publisherScheduler)
+				.then(Mono.fromCallable(() ->
+						this.context.createSocket(
+								this.connectSendUrl == null
+										? SocketType.PAIR
+										: (this.pubSub ? SocketType.XPUB : SocketType.PUSH))
+				))
+				.doOnNext(this.sendSocketConfigurer)
+				.doOnNext((socket) ->
+						socket.connect(this.connectSendUrl != null
+								? this.connectSendUrl
+								: localPairConnection.get()))
+				.delayUntil((socket) ->
+						(this.pubSub && this.connectSendUrl != null)
+								? Mono.just(socket).map(ZMQ.Socket::recv)
+								: Mono.empty())
+				.cache()
+				.publishOn(this.publisherScheduler);
+	}
+
+	private Mono<ZMQ.Socket> prepareSubscribeSocketMono(Supplier<String> localPairConnection, Mono<?> proxyMono) {
+		return proxyMono.publishOn(this.subscriberScheduler)
+				.then(Mono.fromCallable(() ->
+						this.context.createSocket(
+								this.connectSubscribeUrl == null
+										? SocketType.PAIR
+										: (this.pubSub ? SocketType.SUB : SocketType.PULL))))
+				.doOnNext(this.subscribeSocketConfigurer)
+				.doOnNext((socket) -> {
+					if (this.connectSubscribeUrl != null) {
+						socket.connect(this.connectSubscribeUrl);
+						if (this.pubSub) {
+							socket.subscribe(ZMQ.SUBSCRIPTION_ALL);
+						}
+					}
+					else {
+						socket.bind(localPairConnection.get());
+					}
+				})
+				.cache()
+				.publishOn(this.subscriberScheduler);
+	}
+
+	private Flux<? extends Message<?>> prepareSubscriberDataFlux() {
 		Flux<? extends Message<?>> receiveData =
 				this.subscribeSocket
 						.flatMap((socket) -> {
@@ -192,26 +216,7 @@ public ZeroMqChannel(ZContext context, boolean pubSub) {
 					receiveData.publish()
 							.autoConnect(1, (disposable) -> this.subscriberDataDisposable = disposable);
 		}
-
-		this.subscriberData = receiveData;
-
-	}
-
-	private Mono<Integer> proxyMono() {
-		if (this.zeroMqProxy != null) {
-			return Mono.fromCallable(() -> this.zeroMqProxy.getBackendPort())
-					.filter((proxyPort) -> proxyPort > 0)
-					.repeatWhenEmpty(100, (repeat) -> repeat.delayElements(Duration.ofMillis(100))) // NOSONAR
-					.doOnNext((proxyPort) ->
-							setConnectUrl("tcp://localhost:" + this.zeroMqProxy.getFrontendPort() +
-									':' + this.zeroMqProxy.getBackendPort()))
-					.doOnError((error) ->
-							logger.error("The provided '" + this.zeroMqProxy + "' has not been started", error))
-					.cache();
-		}
-		else {
-			return Mono.empty();
-		}
+		return receiveData;
 	}
 
 	/**