From 77936b9251b94fa3cf4e2e308310e4c52a7607bc Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 12 Aug 2020 11:09:32 -0400 Subject: [PATCH] Add `zeromq.adoc` to index files * Fix Sonar smells for new ZeroMQ components * Upgrade to Spring Data `2020.0.0-M2` --- build.gradle | 2 +- .../integration/zeromq/ZeroMqProxy.java | 2 +- .../zeromq/channel/ZeroMqChannel.java | 52 +++++++++---------- src/reference/asciidoc/index-single.adoc | 2 + src/reference/asciidoc/index.adoc | 1 + 5 files changed, 30 insertions(+), 29 deletions(-) diff --git a/build.gradle b/build.gradle index 0afda50ed9..c383c897d6 100644 --- a/build.gradle +++ b/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' diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java index b1309193b1..a607161df1 100644 --- a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java +++ b/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 { 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 567a962573..4fd6eb1717 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 @@ -123,29 +123,10 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl Supplier 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 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 } } diff --git a/src/reference/asciidoc/index-single.adoc b/src/reference/asciidoc/index-single.adoc index 34d8634a9a..f81acc6630 100644 --- a/src/reference/asciidoc/index-single.adoc +++ b/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]] diff --git a/src/reference/asciidoc/index.adoc b/src/reference/asciidoc/index.adoc index cb2f4643da..11aac7e4d3 100644 --- a/src/reference/asciidoc/index.adoc +++ b/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]