Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Simplify complexity in ZeroMqChannel constructor

Browse Source 
* Upgrade to Spring AMQP `2.3.0-M2`
This commit is contained in:
Artem Bilan
2020-08-12 11:35:16 -04:00
parent 77936b9251
commit 5c2af845d5
2 changed files with 67 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

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

@@ -123,50 +123,74 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
Supplier<String> 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<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();
}
}
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<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 class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
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;
}
/**