Add zeromq.adoc to index files

* Fix Sonar smells for new ZeroMQ components
* Upgrade to Spring Data `2020.0.0-M2`

build.gradle

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

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java

@@ -290,7 +290,7 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw
 							this.running.set(true);
 							ZMQ.proxy(frontendSocket, backendSocket, captureSocket, controlSocket);
 						}
-						catch (Exception ex) {
+						catch (Exception ex) { // NOSONAR
 							LOG.error("Cannot start ZeroMQ proxy from bean: " + this.beanName, ex);
 						}
 						finally {

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

@@ -123,29 +123,10 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
 
 		Supplier<String> localPairConnection = () -> "inproc://" + getComponentName() + ".pair";
 
-		Mono<?> proxyMono =
-				Mono.defer(() -> {
-					if (this.zeroMqProxy != null) {
-						return Mono.fromCallable(() -> this.zeroMqProxy.getBackendPort())
-								.filter((port) -> port > 0)
-								.repeatWhenEmpty(100, // NOSONAR
-										(repeat) -> repeat.delayElements(Duration.ofMillis(100))) // NOSONAR
-								.doOnNext((port) ->
-										setConnectUrl("tcp://localhost:" + this.zeroMqProxy.getFrontendPort() +
-												':' + this.zeroMqProxy.getBackendPort()))
-								.doOnError((error) ->
-										logger.error("The provided '"
-												+ this.zeroMqProxy + "' has not been started", error));
-					}
-					else {
-						return Mono.empty();
-					}
-				})
-						.cache();
+		Mono<?> proxyMono = proxyMono();
 
 		this.sendSocket =
-				proxyMono
-						.publishOn(this.publisherScheduler)
+				proxyMono.publishOn(this.publisherScheduler)
 						.then(Mono.fromCallable(() ->
 								this.context.createSocket(
 										this.connectSendUrl == null
@@ -165,8 +146,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
 						.publishOn(this.publisherScheduler);
 
 		this.subscribeSocket =
-				proxyMono
-						.publishOn(this.subscriberScheduler)
+				proxyMono.publishOn(this.subscriberScheduler)
 						.then(Mono.fromCallable(() ->
 								this.context.createSocket(
 										this.connectSubscribeUrl == null
@@ -208,14 +188,32 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
 						.repeat(() -> this.initialized);
 
 		if (this.pubSub) {
-			receiveData = receiveData.publish()
-					.autoConnect(1, (disposable) -> this.subscriberDataDisposable = disposable);
+			receiveData =
+					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();
+		}
+	}
+
 	/**
 	 * Configure a connection to the ZeroMQ proxy with the pair of ports over colon
 	 * for proxy frontend and backend sockets. Mutually exclusive with the {@link #setZeroMqProxy(ZeroMqProxy)}.
@@ -226,7 +224,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
 		if (connectUrl != null) {
 			this.connectSendUrl = connectUrl.substring(0, connectUrl.lastIndexOf(':'));
 			this.connectSubscribeUrl =
-					this.connectSendUrl.substring(0, this.connectSendUrl.lastIndexOf(':'))
+					this.connectSendUrl.substring(0, this.connectSendUrl.lastIndexOf(':')) // NOSONAR
 							+ connectUrl.substring(connectUrl.lastIndexOf(':'));
 		}
 	}
@@ -314,7 +312,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
 		this.subscribeSocket.doOnNext(ZMQ.Socket::close).block();
 		this.subscriberScheduler.dispose();
 		if (this.subscriberDataDisposable != null) {
-			this.subscriberDataDisposable.dispose();
+			this.subscriberDataDisposable.dispose(); // NOSONAR
 		}
 	}
 

src/reference/asciidoc/index-single.adoc

@@ -87,6 +87,8 @@ include::./xml.adoc[]
 
 include::./xmpp.adoc[]
 
+include::./zeromq.adoc[]
+
 include::./zookeeper.adoc[]
 
 [[spring-integration-appendices]]

src/reference/asciidoc/index.adoc

@@ -53,6 +53,7 @@ This documentation is also available as single searchable link:index-single.html
 <<./ws.adoc#ws,Web Services Support>> ::
 <<./xml.adoc#xml,XML Support - Dealing with XML Payloads>> ::
 <<./xmpp.adoc#xmpp,XMPP Support>> ::
+<<./zeromq.adoc#zeromq,ZeroMQ Support>> ::
 <<./zookeeper.adoc#zookeeper,Zookeeper Support>> ::
 
 [horizontal]