From 5c2af845d51b98bb2ad89551573490cc00add295 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 12 Aug 2020 11:35:16 -0400 Subject: [PATCH] Simplify complexity in ZeroMqChannel constructor * Upgrade to Spring AMQP `2.3.0-M2` --- build.gradle | 2 +- .../zeromq/channel/ZeroMqChannel.java | 127 +++++++++--------- 2 files changed, 67 insertions(+), 62 deletions(-) diff --git a/build.gradle b/build.gradle index c383c897d6..45a336d6a9 100644 --- a/build.gradle +++ b/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' 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 4fd6eb1717..decb717eff 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,50 +123,74 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl Supplier localPairConnection = () -> "inproc://" + getComponentName() + ".pair"; - Mono proxyMono = proxyMono(); + Mono proxyMono = prepareProxyMono(); + this.sendSocket = prepareSendSocketMono(localPairConnection, proxyMono); + this.subscribeSocket = prepareSubscribeSocketMono(localPairConnection, proxyMono); + this.subscriberData = prepareSubscriberDataFlux(); + } - 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); + private Mono 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(); + } + } - 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); + private Mono prepareSendSocketMono(Supplier 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 prepareSubscribeSocketMono(Supplier 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> prepareSubscriberDataFlux() { Flux> receiveData = this.subscribeSocket .flatMap((socket) -> { @@ -192,26 +216,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl 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(); - } + return receiveData; } /**