diff --git a/build.gradle b/build.gradle index 3438b8de1b..857a88e423 100644 --- a/build.gradle +++ b/build.gradle @@ -70,6 +70,7 @@ ext { javaxAnnotationVersion= '1.3.2' javaxMailVersion = '1.6.2' jaxbVersion = '2.3.3' + jeroMqVersion = '0.5.2' jmsApiVersion = '2.0.1' jpa21ApiVersion = '1.0.2.Final' jpaApiVersion = '2.2.1' @@ -864,6 +865,18 @@ project('spring-integration-xmpp') { } } +project('spring-integration-zeromq') { + description = 'Spring Integration ZeroMQ Support' + dependencies { + api project(':spring-integration-core') + api "org.zeromq:jeromq:$jeroMqVersion" + + optionalApi 'com.fasterxml.jackson.core:jackson-databind' + + testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion" + } +} + project('spring-integration-zookeeper') { description = 'Spring Integration Zookeeper Support' dependencies { 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 new file mode 100644 index 0000000000..b1309193b1 --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java @@ -0,0 +1,367 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.zeromq; + +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.zeromq.SocketType; +import org.zeromq.ZContext; +import org.zeromq.ZMQ; + +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.SmartLifecycle; +import org.springframework.lang.Nullable; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; +import org.springframework.util.Assert; + +/** + * This class encapsulates the logic to configure and manage a ZeroMQ proxy. + * It binds frontend and backend sockets over TCP on all the available network interfaces + * with either provided or randomly selected ports. + *

+ * The {@link ZeroMqProxy.Type} dictates which pair of ZeroMQ sockets to bind with this proxy + * to implement any possible patterns for ZeroMQ intermediary. Defaults to @link {@link ZeroMqProxy.Type#PULL_PUSH}. + *

+ * The control socket is exposed as a {@link SocketType#PAIR} with an inter-thread transport + * on the {@code "inproc://" + beanName + ".control"} address; it can be obtained via {@link #getControlAddress()}. + * Should be used with the same application from {@link SocketType#PAIR} socket to send + * {@link zmq.ZMQ#PROXY_TERMINATE}, {@link zmq.ZMQ#PROXY_PAUSE} and/or {@link zmq.ZMQ#PROXY_RESUME} commands. + *

+ * If the proxy cannot be started for some reason, an error message is logged and this component is + * left in the non-started state. + *

+ * With an {@link #exposeCaptureSocket} option, an additional capture data socket is bound to inter-thread transport + * as a {@link SocketType#PUB}. There is no specific topic selection, so all the subscribers to this socket + * must subscribe with plain {@link ZMQ#SUBSCRIPTION_ALL}. + * The address for this socket is {@code "inproc://" + beanName + ".capture"}. + * + * @author Artem Bilan + * + * @since 5.4 + * + * @see ZMQ#proxy(ZMQ.Socket, ZMQ.Socket, ZMQ.Socket) + */ +public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAware, DisposableBean { + + private static final Log LOG = LogFactory.getLog(ZeroMqProxy.class); + + private final ZContext context; + + private final Type type; + + private final AtomicBoolean running = new AtomicBoolean(); + + private final AtomicInteger frontendPort = new AtomicInteger(); + + private final AtomicInteger backendPort = new AtomicInteger(); + + private String controlAddress; + + private Executor proxyExecutor; + + private boolean proxyExecutorExplicitlySet; + + @Nullable + private Consumer frontendSocketConfigurer; + + @Nullable + private Consumer backendSocketConfigurer; + + private boolean exposeCaptureSocket; + + @Nullable + private String captureAddress; + + private String beanName; + + private boolean autoStartup; + + private int phase; + + /** + * Create a {@link ZeroMqProxy} instance based on the provided {@link ZContext} + * and {@link Type#PULL_PUSH} as default mode. + * @param context the {@link ZContext} to use + */ + public ZeroMqProxy(ZContext context) { + this(context, Type.PULL_PUSH); + } + + /** + * Create a {@link ZeroMqProxy} instance based on the provided {@link ZContext} + * and {@link Type}. + * @param context the {@link ZContext} to use + * @param type the {@link Type} to use. + */ + public ZeroMqProxy(ZContext context, Type type) { + Assert.notNull(context, "'context' must not be null"); + Assert.notNull(type, "'type' must not be null"); + this.context = context; + this.type = type; + } + + /** + * Configure an executor to perform a ZeroMQ proxy loop. + * The thread is held until ZeroMQ proxy loop is terminated. + * By default an internal {@link Executors#newSingleThreadExecutor} instance is used. + * @param proxyExecutor the {@link Executor} to use for ZeroMQ proxy loop + */ + public void setProxyExecutor(Executor proxyExecutor) { + Assert.notNull(proxyExecutor, "'proxyExecutor' must not be null"); + this.proxyExecutor = proxyExecutor; + this.proxyExecutorExplicitlySet = true; + } + + /** + * Specify a fixed port for frontend socket of the proxy. + * @param frontendPort the port to use; must be more than 0 + */ + public void setFrontendPort(int frontendPort) { + Assert.isTrue(frontendPort > 0, "'frontendPort' must not be zero or negative"); + this.frontendPort.set(frontendPort); + } + + /** + * Specify a fixed port for backend socket of the proxy. + * @param backendPort the port to use; must be more than 0 + */ + public void setBackendPort(int backendPort) { + Assert.isTrue(backendPort > 0, "'backendPort' must not be zero or negative"); + this.backendPort.set(backendPort); + } + + /** + * Provide a {@link Consumer} to configure a proxy frontend socket with arbitrary options, like security. + * @param frontendSocketConfigurer the configurer for frontend socket + */ + public void setFrontendSocketConfigurer(@Nullable Consumer frontendSocketConfigurer) { + this.frontendSocketConfigurer = frontendSocketConfigurer; + } + + /** + * Provide a {@link Consumer} to configure a proxy backend socket with arbitrary options, like security. + * @param backendSocketConfigurer the configurer for backend socket + */ + public void setBackendSocketConfigurer(@Nullable Consumer backendSocketConfigurer) { + this.backendSocketConfigurer = backendSocketConfigurer; + } + + /** + * Whether to bind and expose a capture socket for the proxy data. + * @param exposeCaptureSocket true to bind capture socket for proxy + */ + public void setExposeCaptureSocket(boolean exposeCaptureSocket) { + this.exposeCaptureSocket = exposeCaptureSocket; + } + + @Override + public void setBeanName(String beanName) { + this.beanName = beanName; + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + public Type getType() { + return this.type; + } + + /** + * Return the port a frontend socket is bound or 0 if this proxy has not been started yet. + * @return the port for a frontend socket or 0 + */ + public int getFrontendPort() { + return this.frontendPort.get(); + } + + /** + * Return the port a backend socket is bound or null if this proxy has not been started yet. + * @return the port for a backend socket or 0 + */ + public int getBackendPort() { + return this.backendPort.get(); + } + + /** + * Return the address an {@code inproc} control socket is bound or null if this proxy has not been started yet. + * @return the the address for control socket or null + */ + @Nullable + public String getControlAddress() { + return this.controlAddress; + } + + /** + * Return the address an {@code inproc} capture socket is bound or null if this proxy has not been started yet + * or {@link #captureAddress} is false. + * @return the the address for capture socket or null + */ + @Nullable + public String getCaptureAddress() { + return this.captureAddress; + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + + @Override + public int getPhase() { + return this.phase; + } + + @Override + public void afterPropertiesSet() { + if (this.proxyExecutor == null) { + this.proxyExecutor = Executors.newSingleThreadExecutor(new CustomizableThreadFactory(this.beanName)); + } + this.controlAddress = "inproc://" + this.beanName + ".control"; + if (this.exposeCaptureSocket) { + this.captureAddress = "inproc://" + this.beanName + ".capture"; + } + } + + @Override + public synchronized void start() { + if (!this.running.get()) { + this.proxyExecutor + .execute(() -> { + ZMQ.Socket captureSocket = null; + if (this.exposeCaptureSocket) { + captureSocket = this.context.createSocket(SocketType.PUB); + } + try ( + ZMQ.Socket frontendSocket = this.context.createSocket(this.type.getFrontendSocketType()); + ZMQ.Socket backendSocket = this.context.createSocket(this.type.getBackendSocketType()); + ZMQ.Socket controlSocket = this.context.createSocket(SocketType.PAIR) + ) { + + if (this.frontendSocketConfigurer != null) { + this.frontendSocketConfigurer.accept(frontendSocket); + } + + if (this.backendSocketConfigurer != null) { + this.backendSocketConfigurer.accept(backendSocket); + } + + this.frontendPort.set(bindSocket(frontendSocket, this.frontendPort.get())); + this.backendPort.set(bindSocket(backendSocket, this.backendPort.get())); + boolean bound = controlSocket.bind(this.controlAddress); + if (!bound) { + throw new IllegalArgumentException("Cannot bind ZeroMQ socket to address: " + + this.controlAddress); + } + if (captureSocket != null) { + bound = captureSocket.bind(this.captureAddress); + if (!bound) { + throw new IllegalArgumentException("Cannot bind ZeroMQ socket to address: " + + this.captureAddress); + } + } + this.running.set(true); + ZMQ.proxy(frontendSocket, backendSocket, captureSocket, controlSocket); + } + catch (Exception ex) { + LOG.error("Cannot start ZeroMQ proxy from bean: " + this.beanName, ex); + } + finally { + if (captureSocket != null) { + captureSocket.close(); + } + } + }); + } + } + + @Override + public synchronized void stop() { + if (this.running.getAndSet(false)) { + try (ZMQ.Socket commandSocket = this.context.createSocket(SocketType.PAIR)) { + commandSocket.connect(this.controlAddress); + commandSocket.send(zmq.ZMQ.PROXY_TERMINATE); + } + } + } + + @Override + public boolean isRunning() { + return this.running.get(); + } + + @Override + public void destroy() { + if (!this.proxyExecutorExplicitlySet) { + ((ExecutorService) this.proxyExecutor).shutdown(); + } + } + + private static int bindSocket(ZMQ.Socket socket, int port) { + if (port == 0) { + return socket.bindToRandomPort("tcp://*"); + } + else { + boolean bound = socket.bind("tcp://*:" + port); + if (!bound) { + throw new IllegalArgumentException("Cannot bind ZeroMQ socket to port: " + port); + } + return port; + } + } + + public enum Type { + + SUB_PUB(SocketType.XSUB, SocketType.XPUB), + + PULL_PUSH(SocketType.PULL, SocketType.PUSH), + + ROUTER_DEALER(SocketType.ROUTER, SocketType.DEALER); + + private final SocketType frontendSocketType; + + private final SocketType backendSocketType; + + Type(SocketType frontendSocketType, SocketType backendSocketType) { + this.frontendSocketType = frontendSocketType; + this.backendSocketType = backendSocketType; + } + + public SocketType getFrontendSocketType() { + return this.frontendSocketType; + } + + public SocketType getBackendSocketType() { + return this.backendSocketType; + } + + } + +} 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 new file mode 100644 index 0000000000..567a962573 --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java @@ -0,0 +1,321 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.zeromq.channel; + +import java.time.Duration; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Supplier; + +import org.zeromq.SocketType; +import org.zeromq.ZContext; +import org.zeromq.ZMQ; + +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.mapping.BytesMessageMapper; +import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; +import org.springframework.integration.zeromq.ZeroMqProxy; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.util.Assert; + +import reactor.core.Disposable; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; +import reactor.core.scheduler.Schedulers; + +/** + * The {@link SubscribableChannel} implementation over ZeroMQ sockets. + * It can work in two messaging models: + * - {@code push-pull}, where sent messages are distributed to subscribers in a round-robin manner + * according a respective ZeroMQ {@link SocketType#PUSH} and {@link SocketType#PULL} socket types logic; + * - {@code pub-sub}, where sent messages are distributed to all subscribers; + *

+ * This message channel can work in local mode, when a pair of ZeroMQ sockets of {@link SocketType#PAIR} type + * are connected between publisher (send operation) and subscriber using inter-thread transport binding. + *

+ * In distributed mode this channel has to be connected to an externally managed ZeroMQ proxy. + * The {@link #setConnectUrl(String)} has to be as a standard ZeroMQ connect string, but with an extra port + * over the colon - representing a frontend and backend sockets pair on ZeroMQ proxy. + * For example: {@code tcp://localhost:6001:6002}. + * Another option is to provide a reference to the {@link ZeroMqProxy} instance managed in the same application: + * frontend and backend ports are evaluated from this proxy and the respective connection string is built from them. + *

+ * This way sending and receiving operations on this channel are similar to interaction over a messaging broker. + *

+ * An internal logic of this message channel implementation is based on the project Reactor using its + * {@link Mono}, {@link Flux} and {@link Scheduler} API for better thead model and flow control to avoid + * concurrency primitives for multi-publisher(subscriber) communication within the same application. + * + * @author Artem Bilan + * + * @since 5.4 + */ +public class ZeroMqChannel extends AbstractMessageChannel implements SubscribableChannel { + + public static final Duration DEFAULT_CONSUME_DELAY = Duration.ofSeconds(1); + + private final Map subscribers = new HashMap<>(); + + private final Scheduler publisherScheduler = Schedulers.newSingle("publisherScheduler"); + + private final Scheduler subscriberScheduler = Schedulers.newSingle("subscriberScheduler"); + + private final ZContext context; + + private final boolean pubSub; + + private final Mono sendSocket; + + private final Mono subscribeSocket; + + private final Flux> subscriberData; + + private Duration consumeDelay = DEFAULT_CONSUME_DELAY; + + private BytesMessageMapper messageMapper = new EmbeddedJsonHeadersMessageMapper(); + + private Consumer sendSocketConfigurer = (socket) -> { }; + + private Consumer subscribeSocketConfigurer = (socket) -> { }; + + @Nullable + private ZeroMqProxy zeroMqProxy; + + @Nullable + private volatile String connectSendUrl; + + @Nullable + private volatile String connectSubscribeUrl; + + @Nullable + private volatile Disposable subscriberDataDisposable; + + private volatile boolean initialized; + + public ZeroMqChannel(ZContext context) { + this(context, false); + } + + public ZeroMqChannel(ZContext context, boolean pubSub) { + Assert.notNull(context, "'context' must not be null"); + this.context = context; + this.pubSub = pubSub; + + 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(); + + 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); + + Flux> receiveData = + this.subscribeSocket + .flatMap((socket) -> { + if (this.initialized) { + byte[] data = socket.recv(ZMQ.NOBLOCK); + if (data != null) { + return Mono.just(data); + } + } + return Mono.empty(); + }) + .publishOn(Schedulers.parallel()) + .map(this.messageMapper::toMessage) + .doOnError((error) -> logger.error("Error processing ZeroMQ message", error)) + .repeatWhenEmpty((repeat) -> + this.initialized + ? repeat.delayElements(this.consumeDelay) + : repeat) + .repeat(() -> this.initialized); + + if (this.pubSub) { + receiveData = receiveData.publish() + .autoConnect(1, (disposable) -> this.subscriberDataDisposable = disposable); + } + + this.subscriberData = receiveData; + + } + + /** + * 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)}. + * @param connectUrl the connection string in format {@code PROTOCOL://HOST:FRONTEND_PORT:BACKEND_PORT}, + * e.g. {@code tcp://localhost:6001:6002} + */ + public void setConnectUrl(@Nullable String connectUrl) { + if (connectUrl != null) { + this.connectSendUrl = connectUrl.substring(0, connectUrl.lastIndexOf(':')); + this.connectSubscribeUrl = + this.connectSendUrl.substring(0, this.connectSendUrl.lastIndexOf(':')) + + connectUrl.substring(connectUrl.lastIndexOf(':')); + } + } + + /** + * Specify a reference to a {@link ZeroMqProxy} instance in the same application + * to rely on its ports configuration and make a natural lifecycle dependency without guessing + * when the proxy is started. Mutually exclusive with the {@link #setConnectUrl(String)}. + * @param zeroMqProxy the {@link ZeroMqProxy} instance to use + */ + public void setZeroMqProxy(@Nullable ZeroMqProxy zeroMqProxy) { + this.zeroMqProxy = zeroMqProxy; + } + + public void setConsumeDelay(Duration consumeDelay) { + Assert.notNull(consumeDelay, "'consumeDelay' must not be null"); + this.consumeDelay = consumeDelay; + } + + public void setMessageMapper(BytesMessageMapper messageMapper) { + Assert.notNull(messageMapper, "'messageMapper' must not be null"); + this.messageMapper = messageMapper; + } + + public void setSendSocketConfigurer(Consumer sendSocketConfigurer) { + Assert.notNull(sendSocketConfigurer, "'sendSocketConfigurer' must not be null"); + this.sendSocketConfigurer = sendSocketConfigurer; + } + + public void setSubscribeSocketConfigurer(Consumer subscribeSocketConfigurer) { + Assert.notNull(subscribeSocketConfigurer, "'subscribeSocketConfigurer' must not be null"); + this.subscribeSocketConfigurer = subscribeSocketConfigurer; + } + + @Override + protected void onInit() { + Assert.state(this.zeroMqProxy == null || this.connectSendUrl == null, + "A 'zeroMqProxy' or 'connectUrl' can be provided (or none), but not both."); + super.onInit(); + this.sendSocket.subscribe(); + this.initialized = true; + } + + @Override + protected boolean doSend(Message message, long timeout) { + Assert.state(this.initialized, "the channel is not initialized yet or already destroyed"); + + byte[] data = this.messageMapper.fromMessage(message); + Assert.state(data != null, () -> "The '" + this.messageMapper + "' returned null for '" + message + '\''); + + Mono sendMono = this.sendSocket.map((socket) -> socket.send(data)); + Boolean sent = + timeout > 0 + ? sendMono.block(Duration.ofMillis(timeout)) + : sendMono.block(); + + return Boolean.TRUE.equals(sent); + } + + @Override + public boolean subscribe(MessageHandler handler) { + Assert.state(this.initialized, "the channel is not initialized yet or already destroyed"); + this.subscribers.computeIfAbsent(handler, (key) -> this.subscriberData.subscribe(handler::handleMessage)); + return true; + } + + @Override + public boolean unsubscribe(MessageHandler handler) { + Disposable disposable = this.subscribers.remove(handler); + if (disposable != null) { + disposable.dispose(); + return true; + } + return false; + } + + @Override + public void destroy() { + this.initialized = false; + super.destroy(); + this.sendSocket.doOnNext(ZMQ.Socket::close).block(); + this.publisherScheduler.dispose(); + HashSet handlersCopy = new HashSet<>(this.subscribers.keySet()); + handlersCopy.forEach(this::unsubscribe); + this.subscribeSocket.doOnNext(ZMQ.Socket::close).block(); + this.subscriberScheduler.dispose(); + if (this.subscriberDataDisposable != null) { + this.subscriberDataDisposable.dispose(); + } + } + +} diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/package-info.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/package-info.java new file mode 100644 index 0000000000..d0cb1239f8 --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes for message channels support over ZeroMQ. + */ +package org.springframework.integration.zeromq.channel; diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/package-info.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/package-info.java new file mode 100644 index 0000000000..eef1a1a592 --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides common classes for supporting ZeroMQ components. + */ +package org.springframework.integration.zeromq; diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java new file mode 100644 index 0000000000..b73dba86ff --- /dev/null +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java @@ -0,0 +1,193 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.zeromq.channel; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import java.time.Duration; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Test; +import org.zeromq.SocketType; +import org.zeromq.ZContext; +import org.zeromq.ZMQ; + +import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; +import org.springframework.integration.zeromq.ZeroMqProxy; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Artem Bilan + * + * @since 5.4 + */ +public class ZeroMqChannelTests { + + private static final ZContext CONTEXT = new ZContext(); + + @AfterAll + static void tearDown() { + CONTEXT.close(); + } + + @Test + void testSimpleSendAndReceive() throws InterruptedException { + ZeroMqChannel channel = new ZeroMqChannel(CONTEXT); + channel.setBeanName("testChannel1"); + channel.setConsumeDelay(Duration.ofMillis(10)); + channel.afterPropertiesSet(); + + BlockingQueue> received = new LinkedBlockingQueue<>(); + + channel.subscribe(received::offer); + + assertThat(channel.send(new GenericMessage<>("test1"), 1000)).isTrue(); + assertThat(channel.send(new GenericMessage<>("test2"), 500)).isTrue(); + + Message message = received.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull().extracting(Message::getPayload).isEqualTo("test1"); + message = received.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull().extracting(Message::getPayload).isEqualTo("test2"); + + // Ensure that second subscriber doesn't make it as pub-sub + channel.subscribe(received::offer); + + assertThat(channel.send(new GenericMessage<>("test3"))).isTrue(); + + message = received.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull().extracting(Message::getPayload).isEqualTo("test3"); + assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull(); + + channel.destroy(); + } + + @Test + void testPubSubLocal() throws InterruptedException { + ZeroMqChannel channel = new ZeroMqChannel(CONTEXT, true); + channel.setBeanName("testChannel2"); + channel.setConsumeDelay(Duration.ofMillis(10)); + channel.afterPropertiesSet(); + + BlockingQueue> received = new LinkedBlockingQueue<>(); + + channel.subscribe(received::offer); + channel.subscribe(received::offer); + + GenericMessage testMessage = new GenericMessage<>("test1"); + assertThat(channel.send(testMessage)).isTrue(); + + Message message = received.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull().isEqualTo(testMessage); + message = received.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull().isEqualTo(testMessage); + + channel.destroy(); + } + + @Test + void testPushPullBind() throws InterruptedException { + ZeroMqProxy proxy = new ZeroMqProxy(CONTEXT); + proxy.setBeanName("pullPushProxy"); + proxy.setExposeCaptureSocket(true); + proxy.afterPropertiesSet(); + proxy.start(); + + await().until(() -> proxy.getBackendPort() > 0); + + ZMQ.Socket captureSocket = CONTEXT.createSocket(SocketType.SUB); + captureSocket.connect(proxy.getCaptureAddress()); + captureSocket.subscribe(ZMQ.SUBSCRIPTION_ALL); + + ZeroMqChannel channel = new ZeroMqChannel(CONTEXT); + channel.setConnectUrl("tcp://localhost:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort()); + channel.setBeanName("testChannel3"); + channel.setConsumeDelay(Duration.ofMillis(10)); + channel.afterPropertiesSet(); + + BlockingQueue> received = new LinkedBlockingQueue<>(); + + channel.subscribe(received::offer); + channel.subscribe(received::offer); + + GenericMessage testMessage = new GenericMessage<>("test1"); + assertThat(channel.send(testMessage)).isTrue(); + + Message message = received.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull().isEqualTo(testMessage); + assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull(); + + channel.destroy(); + + byte[] recv = captureSocket.recv(); + assertThat(recv).isNotNull(); + Message capturedMessage = new EmbeddedJsonHeadersMessageMapper().toMessage(recv); + assertThat(capturedMessage).isEqualTo(testMessage); + captureSocket.close(); + + proxy.stop(); + } + + + @Test + void testPubSubBind() throws InterruptedException { + ZeroMqProxy proxy = new ZeroMqProxy(CONTEXT, ZeroMqProxy.Type.SUB_PUB); + proxy.setBeanName("subPubProxy"); + proxy.afterPropertiesSet(); + proxy.start(); + + ZeroMqChannel channel = new ZeroMqChannel(CONTEXT, true); + channel.setZeroMqProxy(proxy); + channel.setBeanName("testChannel4"); + channel.setConsumeDelay(Duration.ofMillis(10)); + channel.afterPropertiesSet(); + + BlockingQueue> received = new LinkedBlockingQueue<>(); + + channel.subscribe(received::offer); + channel.subscribe(received::offer); + + await().until(() -> proxy.getBackendPort() > 0); + + ZeroMqChannel channel2 = new ZeroMqChannel(CONTEXT, true); + channel2.setConnectUrl("tcp://localhost:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort()); + channel2.setBeanName("testChannel5"); + channel.setConsumeDelay(Duration.ofMillis(10)); + channel2.afterPropertiesSet(); + + channel.subscribe(received::offer); + + GenericMessage testMessage = new GenericMessage<>("test1"); + assertThat(channel.send(testMessage)).isTrue(); + + Message message = received.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull().isEqualTo(testMessage); + message = received.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull().isEqualTo(testMessage); + message = received.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull().isEqualTo(testMessage); + assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull(); + + channel.destroy(); + proxy.stop(); + } + +} diff --git a/spring-integration-zeromq/src/test/resources/log4j2-test.xml b/spring-integration-zeromq/src/test/resources/log4j2-test.xml new file mode 100644 index 0000000000..09dc3b2cc5 --- /dev/null +++ b/spring-integration-zeromq/src/test/resources/log4j2-test.xml @@ -0,0 +1,15 @@ + + + + + + + + + + + + + + + diff --git a/src/reference/asciidoc/reactive-streams.adoc b/src/reference/asciidoc/reactive-streams.adoc index 1d8b248bcd..0055a86b82 100644 --- a/src/reference/asciidoc/reactive-streams.adoc +++ b/src/reference/asciidoc/reactive-streams.adoc @@ -155,7 +155,8 @@ A reactive outbound channel adapter implementation is about initiation (or conti An inbound payload could be a reactive type per se or as an event of the whole integration flow which is a part of reactive stream on top. A returned reactive type can be subscribed immediately if we are in one-way, fire-and-forget scenario, or it is propagated downstream (request-reply scenarios) for further integration flow or an explicit subscription in the target business logic, but still downstream preserving reactive streams semantics. -Currently Spring Integration provides channel adapter (or gateway) implementations for <<./webflux.adoc#webflux,WebFlux>>, <<./rsocket.adoc#rsocket,RSocket>> and <<./mongodb.adoc#mongodb,MongoDb>>. +Currently Spring Integration provides channel adapter (or gateway) implementations for <<./webflux.adoc#webflux,WebFlux>>, <<./rsocket.adoc#rsocket,RSocket>>, <<./mongodb.adoc#mongodb,MongoDb>> and <<./r2dbc.adoc#r2dbc,R2DBC>>. +The <<./redis.adoc#redis-stream-outbound,Redis Stream Channel Adapters>> are also reactive and uses `ReactiveStreamOperations` from Spring Data. Also an https://github.com/spring-projects/spring-integration-extensions/tree/master/spring-integration-cassandra[Apache Cassandra Extension] provides a `MessageHandler` implementation for the Cassandra reactive driver. -More reactive channel adapters are coming, for example for https://r2dbc.io/[R2DBC], for Apache Kafka in https://github.com/spring-projects/spring-integration-kafka[Spring Integration Kafka] based on the `ReactiveKafkaProducerTemplate` and `ReactiveKafkaConsumerTemplate` from https://spring.io/projects/spring-kafka[Spring for Apache Kafka] etc. +More reactive channel adapters are coming, for example for Apache Kafka in <<./kafka.adoc#kafka,Kafka>> based on the `ReactiveKafkaProducerTemplate` and `ReactiveKafkaConsumerTemplate` from https://spring.io/projects/spring-kafka[Spring for Apache Kafka] etc. For many other non-reactive channel adapters thread pools are recommended to avoid blocking during reactive stream processing. diff --git a/src/reference/asciidoc/redis.adoc b/src/reference/asciidoc/redis.adoc index 023922fd31..bca511cfee 100644 --- a/src/reference/asciidoc/redis.adoc +++ b/src/reference/asciidoc/redis.adoc @@ -793,3 +793,8 @@ Starting with version 5.0, the `RedisLockRegistry` implements `ExpirableLockRegi === Redis Stream Outbound Channel Adapter TBD + +[[redis-stream-inbound]] +=== Redis Stream Inbound Channel Adapter + +TBD diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index b653d84c0f..6315fc5ee2 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -23,21 +23,30 @@ See <<./kafka.adoc#kafka,Spring for Apache Kafka Support>> for more information. The `KafkaProducerMessageHandler` `sendTimeoutExpression` default has changed. See <<./kafka.adoc#kafka-outbound,Kafka Outbound Channel Adapter>> for more information. +[[x5.4-r2dbc]] ==== R2DBC Channel Adapters The Channel Adapters for R2DBC database interaction have been introduced. See <<./r2dbc.adoc#r2dbc,R2DBC Support>> for more information. +[[x5.4-redis-stream]] ==== Redis Stream Support The Channel Adapters for Redis Stream support have been introduced. See <<./redis.adoc#redis-stream-outbound,Redis Stream Outbound Channel Adapter>> for more information. +[[x5.4-renewable-lock]] ==== Renewable Lock Registry A Renewable lock registry has been introduced to allow renew lease of a distributed lock. See <<./jdbc.adoc#jdbc-lock-registry,JDBC implementation>> for more information. +[[x5.4-zeromq]] +==== ZeroMQ Support + +A `ZeroMqChannel` has been introduced. +See <<./zeromq.adoc#zeromq,ZeroMQ Support>> for more information. + [[x5.4-general]] === General Changes @@ -63,6 +72,7 @@ See <<./ip.adoc#ip-collaborating-adapters,Collaborating Channel Adapters>> and < The `spring-integration-rmi` module is deprecated with no replacement and is going to be removed in the next major version. See <<./rmi.adoc#rmi, RMI Support>> for more information. +[[x5.4-amqp]] === AMQP Changes The outbound endpoints now have a new mechanism for handling publisher confirms and returns. diff --git a/src/reference/asciidoc/xmpp.adoc b/src/reference/asciidoc/xmpp.adoc index 8236dd511c..a109a06b1e 100644 --- a/src/reference/asciidoc/xmpp.adoc +++ b/src/reference/asciidoc/xmpp.adoc @@ -314,7 +314,7 @@ public class CustomConnectionConfiguration { SASLAuthentication.supportSASLMechanism("EXTERNAL", 0); // static initializer ConnectionConfiguration config = new ConnectionConfiguration("localhost", 5223); - config.setTrustorePath("path_to_truststore.jks"); + config.setKeystorePath("path_to_truststore.jks"); config.setSecurityEnabled(true); config.setSocketFactory(SSLSocketFactory.getDefault()); return new XMPPConnection(config); diff --git a/src/reference/asciidoc/zeromq.adoc b/src/reference/asciidoc/zeromq.adoc new file mode 100644 index 0000000000..40ddbace35 --- /dev/null +++ b/src/reference/asciidoc/zeromq.adoc @@ -0,0 +1,104 @@ +[[zeromq]] +== ZeroMQ Support + +Spring Integration provides components to support https://zeromq.org/[ZeroMQ] communication in the application. +The implementation is based on the well-supported Java API of the https://github.com/zeromq/jeromq[JeroMQ] library. +All components encapsulate ZeroMQ socket lifecycles and manage threads for them internally making interactions with these components lock-free and thread-safe. + +You need to include this dependency into your project: + +==== +.Maven +[source, xml, subs="normal"] +---- + + org.springframework.integration + spring-integration-zeromq + {project-version} + +---- + +.Gradle +[source, groovy, subs="normal"] +---- +compile "org.springframework.integration:spring-integration-zeromq:{project-version}" +---- +==== + +[[zeromq-proxy]] +=== ZeroMQ Proxy + +The `ZeroMqProxy` is a Spring-friendly wrapper for the built-in `ZMQ.proxy()` https://zguide.zeromq.org/page:chapter2#toc15[function]. +It encapsulates socket lifecycles and thread management. +The clients of this proxy still can use a standard ZeroMQ socket connection and interaction API. +Alongside with the standard `ZContext` it requires one of the well-known ZeroMQ proxy modes: SUB/PUB, PULL/PUSH or ROUTER/DEALER. +This way an appropriate pair of ZeroMQ socket types are used for the frontend and backend of the proxy. +See `ZeroMqProxy.Type` for details. + +The `ZeroMqProxy` implements `SmartLifecycle` to create, bind and configure the sockets and to start `ZMQ.proxy()` in a dedicated thread from an `Executor` (if any). +The binding for frontend and backend sockets is done over the `tcp://` protocol onto all of the available network interfaces with the provided ports. +Otherwise they are bound to random ports which can be obtained later via the respective `getFrontendPort()` and `getBackendPort()` API methods. + +The control socket is exposed as a `SocketType.PAIR` with an inter-thread transport on the `"inproc://" + beanName + ".control"` address; it can be obtained via `getControlAddress()`. +It should be used with the same application from another `SocketType.PAIR` socket to send `ZMQ.PROXY_TERMINATE`, `ZMQ.PROXY_PAUSE` and/or `ZMQ.PROXY_RESUME` commands. +The `ZeroMqProxy` performs a `ZMQ.PROXY_TERMINATE` command when `stop()` is called for its lifecycle to terminate the `ZMQ.proxy()` loop and close all the bound sockets gracefully. + +The `setExposeCaptureSocket(boolean)` option causes this component to bind an additional inter-thread socket with `SocketType.PUB` to capture and publish all the communication between the frontend and backend sockets as it states with `ZMQ.proxy()` implementation. +This socket is bound to the `"inproc://" + beanName + ".capture"` address and doesn't expect any specific subscription for filtering. + +The frontend and backend sockets can be customized with additional properties, such as read/write timeout or security. +This customization is available through `setFrontendSocketConfigurer(Consumer)` and `setBackendSocketConfigurer(Consumer)` callbacks, respectively. + +The `ZeroMqProxy` could be provided as simple bean like this: + +==== +[source,java] +---- +@Bean +ZeroMqProxy zeroMqProxy() { + ZeroMqProxy proxy = new ZeroMqProxy(CONTEXT, ZeroMqProxy.Type.SUB_PUB); + proxy.setExposeCaptureSocket(true); + proxy.setFrontendPort(6001); + proxy.setBackendPort(6002); + return proxy; +} +---- +==== + +All the client nodes should connect to the host of this proxy via `tcp://` and use the respective port of their interest. + +[[zeromq-message-channel]] +=== ZeroMQ Message Channel + +The `ZeroMqChannel` is a `SubscribableChannel` which uses a pair of ZeroMQ sockets to connect publishers and subscribers for messaging interaction. +It can work in a PUB/SUB mode (defaults to PUSH/PULL); it can also be used as a local inter-thread channel (uses `PAIR` sockets) - the `connectUrl` is not provided in this case. +In distributed mode it has to be connected to an externally managed ZeroMQ proxy, where it can exchange messages with other similar channels connected to the same proxy. +The connect url option is a standard ZeroMQ connection string with the protocol and host and a pair of ports over colon for frontend and backend sockets of the ZeroMQ proxy. +For convenience, the channel could be supplied with the `ZeroMqProxy` instance instead of connection string, if it is configured in the same application as the proxy. + +Both sending and receiving sockets are managed in their own dedicated threads making this channel concurrency-friendly. +This way we can publish and consume to/from a `ZeroMqChannel` from different threads without synchronization. + +By default the `ZeroMqChannel` uses an `EmbeddedJsonHeadersMessageMapper` to (de)serialize the `Message` (including headers) from/to `byte[]` using a Jackson JSON processor. +This logic can be configured via `setMessageMapper(BytesMessageMapper)`. + +Sending and receiving sockets can be customized for any options (read/write timeout, security etc.) via respective `setSendSocketConfigurer(Consumer)` and `setSubscribeSocketConfigurer(Consumer)` callbacks. + +The internal logic of the `ZeroMqChannel` is based on the reactive streams via Project Reactor `Flux` and `Mono` operators. +This provides easier threading control and allows lock-free concurrent publication and consumption to/from the channel. +Local PUB/SUB logic is implemented as a `Flux.publish()` operator to allow all of the local subscribers to this channel to receive the same published message, as distributed subscribers to the `PUB` socket. + +The following is a simple example of a `ZeroMqChannel` configuration: + +==== +[source,java] +---- +@Bean +ZeroMqChannel zeroMqPubSubChannel(ZContext context) { + ZeroMqChannel channel = new ZeroMqChannel(context, true); + channel.setConnectUrl("tcp://localhost:6001:6002"); + channel.setConsumeDelay(Duration.ofMillis(100)); + return channel; +} +---- +====