From a62a7d1ddd6ddb3a23e7ec4daf3ce5d4367d0f9f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 25 Sep 2020 13:46:01 -0400 Subject: [PATCH] INT-3045: Add in & out ZeroMq channel adapters (#3388) * INT-3045: Add in & out ZeroMq channel adapters JIRA: https://jira.spring.io/browse/INT-3045 * Add `ZeroMqMessageHandler` to produce messages into one-way ZeroMq sockets * Add `ZeroMqMessageProducer` to consumer messages from one-way ZeroMq sockets * Add `ConvertingBytesMessageMapper` impl for the `BytesMessageMapper` to delegate an actual conversion into the provided `MessageConverter` * Add `ZeroMqHeaders` for message headers constants representing ZeroMq message attributes * Fix `ZeroMqChannel` for the proper deferred `zeroMqProxy` evaluation * Add more JavaDocs * Fix `ZeroMqChannelTests.testPubSubBind()` to be sure that really all the subscribed channels get the same message from the `PUB` socket * * Fix typo in the `ConvertingBytesMessageMapper` * Add `this` for `doOnError()` in the `ZeroMqChannel` & `ZeroMqMessageProducer` * Change the bind logic in the `ZeroMqMessageProducer` to `port` and let it to bind to random port. The actual port is available later via `getBoundPort()` * Introduce a `ZeroMqMessageProducer.receiveRaw()` to let received `ZMsg` to be produce as a `payload` * Add a logic into `ZeroMqMessageHandler` to treat `ZMsg` in the payload of request message as is without any conversion * Fix race condition in the `ZeroMqMessageProducer` to destroy `consumerScheduler` when the main `Flux` is complete * * Add Java DSL for ZeroMq components * Extract `ReactiveMessageHandlerSpec` for `ReactiveMessageHandler` impls * Add debug message into `EmbeddedJsonHeadersMessageMapper` when cannot `decodeNativeFormat()` * Make `ReactiveMongoDbMessageHandlerSpec` extending `ReactiveMessageHandlerSpec` * Make `ZeroMqProxy` `autoStartup` by default * Add `ZeroMqDslTests` to cover all the Java DSL for ZeroMq * Introduce a `MimeTypeSerializer` to serialize a `MimeType` into JSON as a plain string; use it as extra serializer in the `JacksonJsonUtils.messagingAwareMapper()` * Fix typo for the `AllowListTypeResolverBuilder` inner class * * Add some docs * Fix Checkstyle violations * * More docs * Fix language in Docs Co-authored-by: Gary Russell Co-authored-by: Gary Russell --- .../dsl/ReactiveMessageHandlerSpec.java | 48 +++ .../mapping/ConvertingBytesMessageMapper.java | 77 +++++ .../EmbeddedJsonHeadersMessageMapper.java | 6 +- .../support/json/JacksonJsonUtils.java | 7 +- .../support/json/MimeTypeSerializer.java | 57 ++++ .../ReactiveMongoDbMessageHandlerSpec.java | 26 +- .../integration/zeromq/ZeroMqHeaders.java | 38 +++ .../integration/zeromq/ZeroMqProxy.java | 2 +- .../zeromq/channel/ZeroMqChannel.java | 70 ++-- .../integration/zeromq/dsl/ZeroMq.java | 97 ++++++ .../zeromq/dsl/ZeroMqChannelSpec.java | 109 ++++++ .../zeromq/dsl/ZeroMqMessageHandlerSpec.java | 142 ++++++++ .../zeromq/dsl/ZeroMqMessageProducerSpec.java | 133 ++++++++ .../integration/zeromq/dsl/package-info.java | 4 + .../zeromq/inbound/ZeroMqMessageProducer.java | 322 ++++++++++++++++++ .../zeromq/inbound/package-info.java | 4 + .../zeromq/outbound/ZeroMqMessageHandler.java | 214 ++++++++++++ .../zeromq/outbound/package-info.java | 4 + .../zeromq/channel/ZeroMqChannelTests.java | 10 +- .../zeromq/dsl/ZeroMqDslTests.java | 153 +++++++++ .../inbound/ZeroMqMessageProducerTests.java | 143 ++++++++ .../outbound/ZeroMqMessageHandlerTests.java | 138 ++++++++ src/reference/asciidoc/whats-new.adoc | 2 +- src/reference/asciidoc/zeromq.adoc | 111 ++++++ 24 files changed, 1864 insertions(+), 53 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/dsl/ReactiveMessageHandlerSpec.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/mapping/ConvertingBytesMessageMapper.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqHeaders.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMq.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqChannelSpec.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/package-info.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/package-info.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java create mode 100644 spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/package-info.java create mode 100644 spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/dsl/ZeroMqDslTests.java create mode 100644 spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java create mode 100644 spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ReactiveMessageHandlerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ReactiveMessageHandlerSpec.java new file mode 100644 index 0000000000..6074e2f38d --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ReactiveMessageHandlerSpec.java @@ -0,0 +1,48 @@ +/* + * 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.dsl; + +import java.util.Collections; +import java.util.Map; + +import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; +import org.springframework.messaging.ReactiveMessageHandler; + +/** + * The {@link MessageHandlerSpec} extension for {@link ReactiveMessageHandler}. + * + * @author Artem Bilan + * + * @since 5.4 + */ +public abstract class ReactiveMessageHandlerSpec, H extends ReactiveMessageHandler> + extends MessageHandlerSpec + implements ComponentsRegistration { + + protected final H reactiveMessageHandler; // NOSONAR - final + + protected ReactiveMessageHandlerSpec(H reactiveMessageHandler) { + this.reactiveMessageHandler = reactiveMessageHandler; + this.target = new ReactiveMessageHandlerAdapter(this.reactiveMessageHandler); + } + + @Override + public Map getComponentsToRegister() { + return Collections.singletonMap(this.reactiveMessageHandler, null); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/mapping/ConvertingBytesMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/mapping/ConvertingBytesMessageMapper.java new file mode 100644 index 0000000000..099f023402 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/mapping/ConvertingBytesMessageMapper.java @@ -0,0 +1,77 @@ +/* + * 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.mapping; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Map; + +import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.util.Assert; + +/** + * The {@link BytesMessageMapper} implementation to delegate to/from {@link Message} + * conversion into the provided {@link MessageConverter}. + *

+ * The {@link MessageConverter} must not return {@code null} from its + * {@link MessageConverter#fromMessage(Message, Class)} and {@link MessageConverter#toMessage(Object, MessageHeaders)} + * methods. + *

+ * If {@link MessageConverter#fromMessage(Message, Class)} returns {@link String}, it is converted to {@link byte[]} + * using a {@link StandardCharsets#UTF_8} encoding. + * + * @author Artem Bilan + * + * @since 5.4 + */ +public class ConvertingBytesMessageMapper implements BytesMessageMapper { + + private final MessageConverter messageConverter; + + public ConvertingBytesMessageMapper(MessageConverter messageConverter) { + Assert.notNull(messageConverter, "'messageConverter' must not be null"); + this.messageConverter = messageConverter; + } + + @Override + @NonNull + public Message toMessage(byte[] bytes, @Nullable Map headers) { + MessageHeaders messageHeaders = null; + if (headers != null) { + messageHeaders = new MessageHeaders(headers); + } + Message message = this.messageConverter.toMessage(bytes, messageHeaders); + Assert.state(message != null, () -> + "the '" + this.messageConverter + "' produced null for bytes:" + Arrays.toString(bytes)); + return message; + } + + @Override + @NonNull + public byte[] fromMessage(Message message) { + Object result = this.messageConverter.fromMessage(message, byte[].class); + Assert.state(result != null, () -> "the '" + this.messageConverter + "' produced null for message: " + message); + return result instanceof String + ? ((String) result).getBytes(StandardCharsets.UTF_8) + : (byte[]) result; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java index e3e7ec61f8..fd3f1f49ee 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java @@ -224,16 +224,14 @@ public class EmbeddedJsonHeadersMessageMapper implements BytesMessageMapper { message = decodeNativeFormat(bytes, headers); } catch (@SuppressWarnings("unused") Exception e) { - // empty + this.logger.debug("Failed to decode native format", e); } if (message == null) { try { message = (Message) this.objectMapper.readValue(bytes, Object.class); } catch (Exception e) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Failed to decode JSON", e); - } + this.logger.debug("Failed to decode JSON", e); } } if (message != null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java index a1485507c3..61d936327b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java @@ -67,7 +67,7 @@ public final class JacksonJsonUtils { if (JacksonPresent.isJackson2Present()) { ObjectMapper mapper = new Jackson2JsonObjectMapper().getObjectMapper(); - mapper.setDefaultTyping(new AllowlistTypeResolverBuilder(trustedPackages)); + mapper.setDefaultTyping(new AllowListTypeResolverBuilder(trustedPackages)); GenericMessageJacksonDeserializer genericMessageDeserializer = new GenericMessageJacksonDeserializer(); genericMessageDeserializer.setMapper(mapper); @@ -83,6 +83,7 @@ public final class JacksonJsonUtils { SimpleModule simpleModule = new SimpleModule() .addSerializer(new MessageHeadersJacksonSerializer()) + .addSerializer(new MimeTypeSerializer()) .addDeserializer(GenericMessage.class, genericMessageDeserializer) .addDeserializer(ErrorMessage.class, errorMessageDeserializer) .addDeserializer(AdviceMessage.class, adviceMessageDeserializer) @@ -107,13 +108,13 @@ public final class JacksonJsonUtils { * * @since 4.3.11 */ - private static final class AllowlistTypeResolverBuilder extends ObjectMapper.DefaultTypeResolverBuilder { + private static final class AllowListTypeResolverBuilder extends ObjectMapper.DefaultTypeResolverBuilder { private static final long serialVersionUID = 1L; private final String[] trustedPackages; - AllowlistTypeResolverBuilder(String... trustedPackages) { + AllowListTypeResolverBuilder(String... trustedPackages) { super(ObjectMapper.DefaultTyping.NON_FINAL, //we do explicit validation in the TypeIdResolver BasicPolymorphicTypeValidator.builder() diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java new file mode 100644 index 0000000000..0220dc14ab --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java @@ -0,0 +1,57 @@ +/* + * 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.support.json; + +import java.io.IOException; + +import org.springframework.util.MimeType; + +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.databind.JsonSerializer; +import com.fasterxml.jackson.databind.SerializerProvider; +import com.fasterxml.jackson.databind.jsontype.TypeSerializer; +import com.fasterxml.jackson.databind.ser.std.StdSerializer; + +/** + * Simple {@link JsonSerializer} extension to represent a {@link MimeType} object in the + * target JSON as a plain string. + * + * @author Artem Bilan + * + * @since 5.4 + */ +public class MimeTypeSerializer extends StdSerializer { + + private static final long serialVersionUID = 1L; + + public MimeTypeSerializer() { + super(MimeType.class); + } + + @Override + public void serializeWithType(MimeType value, JsonGenerator generator, SerializerProvider serializers, + TypeSerializer typeSer) throws IOException { + + serialize(value, generator, serializers); + } + + @Override + public void serialize(MimeType value, JsonGenerator generator, SerializerProvider provider) throws IOException { + generator.writeString(value.toString()); + } + +} diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java index f4ab5e64db..2a4a42c850 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java @@ -16,8 +16,6 @@ package org.springframework.integration.mongodb.dsl; -import java.util.Collections; -import java.util.Map; import java.util.function.Function; import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; @@ -27,8 +25,8 @@ import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.dsl.ComponentsRegistration; import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.dsl.ReactiveMessageHandlerSpec; import org.springframework.integration.expression.FunctionExpression; -import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; import org.springframework.integration.mongodb.outbound.ReactiveMongoDbStoringMessageHandler; import org.springframework.messaging.Message; @@ -41,22 +39,15 @@ import org.springframework.messaging.Message; * @since 5.3 */ public class ReactiveMongoDbMessageHandlerSpec - extends MessageHandlerSpec + extends ReactiveMessageHandlerSpec implements ComponentsRegistration { - protected final ReactiveMongoDbStoringMessageHandler messageHandler; // NOSONAR - final - protected ReactiveMongoDbMessageHandlerSpec(ReactiveMongoDatabaseFactory mongoDbFactory) { - this(new ReactiveMongoDbStoringMessageHandler(mongoDbFactory)); + super(new ReactiveMongoDbStoringMessageHandler(mongoDbFactory)); } protected ReactiveMongoDbMessageHandlerSpec(ReactiveMongoOperations reactiveMongoOperations) { - this(new ReactiveMongoDbStoringMessageHandler(reactiveMongoOperations)); - } - - private ReactiveMongoDbMessageHandlerSpec(ReactiveMongoDbStoringMessageHandler messageHandler) { - this.messageHandler = messageHandler; - this.target = new ReactiveMessageHandlerAdapter(this.messageHandler); + super(new ReactiveMongoDbStoringMessageHandler(reactiveMongoOperations)); } /** @@ -65,7 +56,7 @@ public class ReactiveMongoDbMessageHandlerSpec * @return the spec */ public ReactiveMongoDbMessageHandlerSpec mongoConverter(MongoConverter mongoConverter) { - this.messageHandler.setMongoConverter(mongoConverter); + this.reactiveMessageHandler.setMongoConverter(mongoConverter); return this; } @@ -96,13 +87,8 @@ public class ReactiveMongoDbMessageHandlerSpec * @return the spec */ public ReactiveMongoDbMessageHandlerSpec collectionNameExpression(Expression collectionNameExpression) { - this.messageHandler.setCollectionNameExpression(collectionNameExpression); + this.reactiveMessageHandler.setCollectionNameExpression(collectionNameExpression); return this; } - @Override - public Map getComponentsToRegister() { - return Collections.singletonMap(this.messageHandler, null); - } - } diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqHeaders.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqHeaders.java new file mode 100644 index 0000000000..838ed6be9d --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqHeaders.java @@ -0,0 +1,38 @@ +/* + * 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; + +/** + * The message headers constants to repsent ZeroMq message attributes. + * + * @author Artem Bilan + * + * @since 5.4 + */ +public final class ZeroMqHeaders { + + public static final String PREFIX = "zeromq_"; + + /** + * A ZeroMq pub/sub message topic header. + */ + public static final String TOPIC = PREFIX + "topic"; + + private ZeroMqHeaders() { + } + +} 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 a607161df1..727c8f9650 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 @@ -97,7 +97,7 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw private String beanName; - private boolean autoStartup; + private boolean autoStartup = true; private int phase; 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 decb717eff..1d7d0abc0a 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 @@ -112,10 +112,21 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl private volatile boolean initialized; + /** + * Create a channel instance based on the provided {@link ZContext} with push/pull + * communication model. + * @param context the {@link ZContext} to use. + */ public ZeroMqChannel(ZContext context) { this(context, false); } + /** + * Create a channel instance based on the provided {@link ZContext} and provided + * communication model. + * @param context the {@link ZContext} to use. + * @param pubSub the communication model: push/pull or pub/sub. + */ public ZeroMqChannel(ZContext context, boolean pubSub) { Assert.notNull(context, "'context' must not be null"); this.context = context; @@ -130,20 +141,22 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl } 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(); - } + return Mono.defer(() -> { + 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)); + } + else { + return Mono.empty(); + } + }) + .cache(); } private Mono prepareSendSocketMono(Supplier localPairConnection, Mono proxyMono) { @@ -152,17 +165,13 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl this.context.createSocket( this.connectSendUrl == null ? SocketType.PAIR - : (this.pubSub ? SocketType.XPUB : SocketType.PUSH)) + : (this.pubSub ? SocketType.PUB : 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); } @@ -177,10 +186,10 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl .doOnNext(this.subscribeSocketConfigurer) .doOnNext((socket) -> { if (this.connectSubscribeUrl != null) { - socket.connect(this.connectSubscribeUrl); if (this.pubSub) { socket.subscribe(ZMQ.SUBSCRIPTION_ALL); } + socket.connect(this.connectSubscribeUrl); } else { socket.bind(localPairConnection.get()); @@ -204,7 +213,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl }) .publishOn(Schedulers.parallel()) .map(this.messageMapper::toMessage) - .doOnError((error) -> logger.error("Error processing ZeroMQ message", error)) + .doOnError((error) -> logger.error("Error processing ZeroMQ message in the " + this, error)) .repeatWhenEmpty((repeat) -> this.initialized ? repeat.delayElements(this.consumeDelay) @@ -244,21 +253,40 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl this.zeroMqProxy = zeroMqProxy; } + /** + * Specify a {@link Duration} to delay consumption when no data received. + * @param consumeDelay the {@link Duration} to delay consumption when empty; + * defaults to {@link #DEFAULT_CONSUME_DELAY}. + */ public void setConsumeDelay(Duration consumeDelay) { Assert.notNull(consumeDelay, "'consumeDelay' must not be null"); this.consumeDelay = consumeDelay; } + /** + * Provide a {@link BytesMessageMapper} to convert to/from messages when send or receive happens + * on the sockets. + * @param messageMapper the {@link BytesMessageMapper} to use; + * defaults to {@link EmbeddedJsonHeadersMessageMapper}. + */ public void setMessageMapper(BytesMessageMapper messageMapper) { Assert.notNull(messageMapper, "'messageMapper' must not be null"); this.messageMapper = messageMapper; } + /** + * The {@link Consumer} callback to configure a publishing socket. + * @param sendSocketConfigurer the {@link Consumer} to use. + */ public void setSendSocketConfigurer(Consumer sendSocketConfigurer) { Assert.notNull(sendSocketConfigurer, "'sendSocketConfigurer' must not be null"); this.sendSocketConfigurer = sendSocketConfigurer; } + /** + * The {@link Consumer} callback to configure a consuming socket. + * @param subscribeSocketConfigurer the {@link Consumer} to use. + */ public void setSubscribeSocketConfigurer(Consumer subscribeSocketConfigurer) { Assert.notNull(subscribeSocketConfigurer, "'subscribeSocketConfigurer' must not be null"); this.subscribeSocketConfigurer = subscribeSocketConfigurer; diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMq.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMq.java new file mode 100644 index 0000000000..7dace41972 --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMq.java @@ -0,0 +1,97 @@ +/* + * 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.dsl; + +import org.zeromq.SocketType; +import org.zeromq.ZContext; + +/** + * Factory class for ZeroMq components DSL. + * + * @author Artem Bilan + * + * @since 5.4 + */ +public final class ZeroMq { + + /** + * Create an instance of {@link ZeroMqChannelSpec} based on the provided {@link ZContext}. + * @param context the {@link ZContext} to use. + * @return the spec. + */ + public static ZeroMqChannelSpec zeroMqChannel(ZContext context) { + return new ZeroMqChannelSpec(context, false); + } + + /** + * Create an instance of {@link ZeroMqChannelSpec} in pub/sub mode based on the provided {@link ZContext}. + * @param context the {@link ZContext} to use. + * @return the spec. + */ + public static ZeroMqChannelSpec pubSubZeroMqChannel(ZContext context) { + return new ZeroMqChannelSpec(context, true); + } + + /** + * Create an instance of {@link ZeroMqMessageHandlerSpec} for the provided {@link ZContext} and connection URL. + * @param context the {@link ZContext} to use. + * @param connectUrl the URL to connect a ZeroMq socket to. + * @return the spec. + */ + public static ZeroMqMessageHandlerSpec outboundChannelAdapter(ZContext context, String connectUrl) { + return new ZeroMqMessageHandlerSpec(context, connectUrl); + } + + /** + * Create an instance of {@link ZeroMqMessageHandlerSpec} for the provided {@link ZContext}, connection URL + * and {@link SocketType}. + * @param context the {@link ZContext} to use. + * @param connectUrl the URL to connect a ZeroMq socket to. + * @param socketType the {@link SocketType} for ZeroMq socket. + * @return the spec. + */ + public static ZeroMqMessageHandlerSpec outboundChannelAdapter(ZContext context, String connectUrl, + SocketType socketType) { + + return new ZeroMqMessageHandlerSpec(context, connectUrl, socketType); + } + + /** + * Create an instance of {@link ZeroMqMessageProducerSpec} for the provided {@link ZContext}. + * @param context the {@link ZContext} to use. + * @return the spec. + */ + public static ZeroMqMessageProducerSpec inboundChannelAdapter(ZContext context) { + return new ZeroMqMessageProducerSpec(context); + } + + /** + * Create an instance of {@link ZeroMqMessageProducerSpec} for the provided {@link ZContext} + * and {@link SocketType}. + * @param context the {@link ZContext} to use. + * @param socketType the {@link SocketType} for ZeroMq socket. + * @return the spec. + */ + public static ZeroMqMessageProducerSpec inboundChannelAdapter(ZContext context, SocketType socketType) { + return new ZeroMqMessageProducerSpec(context, socketType); + } + + + private ZeroMq() { + } + +} diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqChannelSpec.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqChannelSpec.java new file mode 100644 index 0000000000..8a559dbc45 --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqChannelSpec.java @@ -0,0 +1,109 @@ +/* + * 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.dsl; + +import java.time.Duration; +import java.util.function.Consumer; + +import org.zeromq.ZContext; +import org.zeromq.ZMQ; + +import org.springframework.integration.dsl.MessageChannelSpec; +import org.springframework.integration.mapping.BytesMessageMapper; +import org.springframework.integration.zeromq.ZeroMqProxy; +import org.springframework.integration.zeromq.channel.ZeroMqChannel; + +/** + * The {@link MessageChannelSpec} for a {@link ZeroMqChannel}. + * + * @author Artem Bilan + * + * @since 5.4 + */ +public class ZeroMqChannelSpec extends MessageChannelSpec { + + protected ZeroMqChannelSpec(ZContext context, boolean pubSub) { + this.channel = new ZeroMqChannel(context, pubSub); + } + + /** + * 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 #zeroMqProxy(ZeroMqProxy)}. + * @param connectUrl the connection string in format {@code PROTOCOL://HOST:FRONTEND_PORT:BACKEND_PORT}, + * e.g. {@code tcp://localhost:6001:6002} + * @return the spec + */ + public ZeroMqChannelSpec connectUrl(String connectUrl) { + this.channel.setConnectUrl(connectUrl); + return this; + } + + /** + * 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 #connectUrl(String)}. + * @param zeroMqProxy the {@link ZeroMqProxy} instance to use + * @return the spec + */ + public ZeroMqChannelSpec zeroMqProxy(ZeroMqProxy zeroMqProxy) { + this.channel.setZeroMqProxy(zeroMqProxy); + return this; + } + + /** + * Specify a {@link Duration} to delay consumption when no data received. + * @param consumeDelay the {@link Duration} to delay consumption when empty; + * defaults to {@link ZeroMqChannel#DEFAULT_CONSUME_DELAY}. + * @return the spec + */ + public ZeroMqChannelSpec consumeDelay(Duration consumeDelay) { + this.channel.setConsumeDelay(consumeDelay); + return this; + } + + /** + * Provide a {@link BytesMessageMapper} to convert to/from messages when send or receive happens + * on the sockets. + * @param messageMapper the {@link BytesMessageMapper} to use. + * @return the spec + */ + public ZeroMqChannelSpec messageMapper(BytesMessageMapper messageMapper) { + this.channel.setMessageMapper(messageMapper); + return this; + } + + /** + * The {@link Consumer} callback to configure a publishing socket. + * @param sendSocketConfigurer the {@link Consumer} to use. + * @return the spec + */ + public ZeroMqChannelSpec sendSocketConfigurer(Consumer sendSocketConfigurer) { + this.channel.setSendSocketConfigurer(sendSocketConfigurer); + return this; + } + + /** + * The {@link Consumer} callback to configure a consuming socket. + * @param subscribeSocketConfigurer the {@link Consumer} to use. + * @return the spec + */ + public ZeroMqChannelSpec subscribeSocketConfigurer(Consumer subscribeSocketConfigurer) { + this.channel.setSubscribeSocketConfigurer(subscribeSocketConfigurer); + return this; + } + +} diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java new file mode 100644 index 0000000000..dbcce5eb0b --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java @@ -0,0 +1,142 @@ +/* + * 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.dsl; + +import java.util.function.Consumer; +import java.util.function.Function; + +import org.zeromq.SocketType; +import org.zeromq.ZContext; +import org.zeromq.ZMQ; + +import org.springframework.expression.Expression; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.dsl.ReactiveMessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.mapping.OutboundMessageMapper; +import org.springframework.integration.zeromq.outbound.ZeroMqMessageHandler; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; + +/** + * The {@link MessageHandlerSpec} extension for {@link ZeroMqMessageHandler}. + * + * @author Artem Bilan + * + * @since 5.4 + */ +public class ZeroMqMessageHandlerSpec + extends ReactiveMessageHandlerSpec { + + /** + * Create an instance based on the provided {@link ZContext} and connection string. + * @param context the {@link ZContext} to use for creating sockets. + * @param connectUrl the URL to connect the socket to. + */ + protected ZeroMqMessageHandlerSpec(ZContext context, String connectUrl) { + super(new ZeroMqMessageHandler(context, connectUrl)); + } + + /** + * Create an instance based on the provided {@link ZContext}, connection string and {@link SocketType}. + * @param context the {@link ZContext} to use for creating sockets. + * @param connectUrl the URL to connect the socket to. + * @param socketType the {@link SocketType} to use; + * only {@link SocketType#PAIR}, {@link SocketType#PUB} and {@link SocketType#PUSH} are supported. + */ + protected ZeroMqMessageHandlerSpec(ZContext context, String connectUrl, SocketType socketType) { + super(new ZeroMqMessageHandler(context, connectUrl, socketType)); + } + + /** + * Provide an {@link OutboundMessageMapper} to convert a request message into {@code byte[]} + * for sending into ZeroMq socket. + * @param messageMapper the {@link OutboundMessageMapper} to use. + * @return the spec + */ + public ZeroMqMessageHandlerSpec messageMapper(OutboundMessageMapper messageMapper) { + this.reactiveMessageHandler.setMessageMapper(messageMapper); + return this; + } + + /** + * Provide a {@link MessageConverter} (as an alternative to {@link #messageMapper}) + * for converting a request message into {@code byte[]} for sending into ZeroMq socket. + * @param messageConverter the {@link MessageConverter} to use. + * @return the spec + */ + public ZeroMqMessageHandlerSpec messageConverter(MessageConverter messageConverter) { + this.reactiveMessageHandler.setMessageConverter(messageConverter); + return this; + } + + /** + * Provide a {@link Consumer} to configure a socket with arbitrary options, like security. + * @param socketConfigurer the configurer for socket options. + * @return the spec + */ + public ZeroMqMessageHandlerSpec socketConfigurer(Consumer socketConfigurer) { + this.reactiveMessageHandler.setSocketConfigurer(socketConfigurer); + return this; + } + + /** + * Specify a topic the {@link SocketType#PUB} socket is going to use for distributing messages into the + * subscriptions. It is ignored for all other {@link SocketType}s supported. + * @param topic the topic to use. + * @return the spec + */ + public ZeroMqMessageHandlerSpec topic(String topic) { + this.reactiveMessageHandler.setTopic(topic); + return this; + } + + /** + * Specify a {@link Function} to evaluate a topic a {@link SocketType#PUB} + * is going to use for distributing messages into the + * subscriptions.It is ignored for all other {@link SocketType}s supported. + * @param topicFunction the {@link Function} to evaluate topic for publishing. + * @return the spec + */ + public ZeroMqMessageHandlerSpec topicFunction(Function, String> topicFunction) { + return topicExpression(new FunctionExpression<>(topicFunction)); + } + + /** + * Specify a SpEL expression to evaluate a topic a {@link SocketType#PUB} + * is going to use for distributing messages into the + * subscriptions.It is ignored for all other {@link SocketType}s supported. + * @param topicExpression the expression to evaluate topic for publishing. + * @return the spec + */ + public ZeroMqMessageHandlerSpec topicExpression(String topicExpression) { + return topicExpression(PARSER.parseExpression(topicExpression)); + } + + /** + * Specify a SpEL expression to evaluate a topic a {@link SocketType#PUB} + * is going to use for distributing messages into the + * subscriptions.It is ignored for all other {@link SocketType}s supported. + * @param topicExpression the expression to evaluate topic for publishing. + * @return the spec + */ + public ZeroMqMessageHandlerSpec topicExpression(Expression topicExpression) { + this.reactiveMessageHandler.setTopicExpression(topicExpression); + return this; + } + +} diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java new file mode 100644 index 0000000000..5b94e3bb8e --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java @@ -0,0 +1,133 @@ +/* + * 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.dsl; + +import java.time.Duration; +import java.util.function.Consumer; + +import org.zeromq.SocketType; +import org.zeromq.ZContext; +import org.zeromq.ZMQ; +import org.zeromq.ZMsg; + +import org.springframework.integration.dsl.MessageProducerSpec; +import org.springframework.integration.mapping.InboundMessageMapper; +import org.springframework.integration.zeromq.ZeroMqHeaders; +import org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; + +/** + * @author Artem Bilan + * + * @since 5.4 + */ +public class ZeroMqMessageProducerSpec + extends MessageProducerSpec { + + protected ZeroMqMessageProducerSpec(ZContext context) { + super(new ZeroMqMessageProducer(context)); + } + + protected ZeroMqMessageProducerSpec(ZContext context, SocketType socketType) { + super(new ZeroMqMessageProducer(context, socketType)); + } + + /** + * Specify a {@link Duration} to delay consumption when no data received. + * @param consumeDelay the {@link Duration} to delay consumption when empty. + * @return the spec + */ + public ZeroMqMessageProducerSpec consumeDelay(Duration consumeDelay) { + this.target.setConsumeDelay(consumeDelay); + return this; + } + + /** + * Provide an {@link InboundMessageMapper} to convert a consumed data into a message to produce. + * @param messageMapper the {@link InboundMessageMapper} to use. + * @return the spec + */ + public ZeroMqMessageProducerSpec messageMapper(InboundMessageMapper messageMapper) { + this.target.setMessageMapper(messageMapper); + return this; + } + + /** + * Provide a {@link MessageConverter} (as an alternative to {@link #messageMapper}) + * for converting a consumed data into a message to produce. + * @param messageConverter the {@link MessageConverter} to use. + * @return the spec + */ + public ZeroMqMessageProducerSpec messageConverter(MessageConverter messageConverter) { + this.target.setMessageConverter(messageConverter); + return this; + } + + /** + * Whether raw {@link ZMsg} is present as a payload of message to produce or + * it is fully converted to a {@link Message} including {@link ZeroMqHeaders#TOPIC} header (if any). + * @param receiveRaw to convert from {@link ZMsg} or not; defaults to convert. + * @return the spec + */ + public ZeroMqMessageProducerSpec receiveRaw(boolean receiveRaw) { + this.target.setReceiveRaw(receiveRaw); + return this; + } + + /** + * Provide a {@link Consumer} to configure a socket with arbitrary options, like security. + * @param socketConfigurer the configurer for socket options. + * @return the spec + */ + public ZeroMqMessageProducerSpec socketConfigurer(Consumer socketConfigurer) { + this.target.setSocketConfigurer(socketConfigurer); + return this; + } + + /** + * Specify topics the {@link SocketType#SUB} socket is going to use for subscription. + * It is ignored for all other {@link SocketType}s supported. + * @param topics the topics to use. + * @return the spec + */ + public ZeroMqMessageProducerSpec topics(String... topics) { + this.target.setTopics(topics); + return this; + } + + /** + * Configure an URL for {@link org.zeromq.ZMQ.Socket#connect(String)}. + * @param connectUrl the URL to connect ZeroMq socket to. + * @return the spec + */ + public ZeroMqMessageProducerSpec connectUrl(String connectUrl) { + this.target.setConnectUrl(connectUrl); + return this; + } + + /** + * Configure a port for TCP protocol binding via {@link org.zeromq.ZMQ.Socket#bind(String)}. + * @param port the port to bind ZeroMq socket to over TCP. + * @return the spec + */ + public ZeroMqMessageProducerSpec bindPort(int port) { + this.target.setBindPort(port); + return this; + } + +} diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/package-info.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/package-info.java new file mode 100644 index 0000000000..a6e8db60ad --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes for supporting ZeroMQ component via Java DSL. + */ +package org.springframework.integration.zeromq.dsl; diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java new file mode 100644 index 0000000000..774cb0b5cc --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java @@ -0,0 +1,322 @@ +/* + * 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.inbound; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; + +import org.zeromq.SocketType; +import org.zeromq.ZContext; +import org.zeromq.ZMQ; +import org.zeromq.ZMsg; + +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.mapping.ConvertingBytesMessageMapper; +import org.springframework.integration.mapping.InboundMessageMapper; +import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter; +import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.integration.zeromq.ZeroMqHeaders; +import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.util.Assert; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; +import reactor.core.scheduler.Schedulers; + +/** + * A {@link MessageProducerSupport} implementation for consuming messages from ZeroMq socket. + * Only {@link SocketType#PAIR}, {@link SocketType#SUB} and {@link SocketType#PULL} are supported. + * This component can bind or connect the socket. + *

+ * When the {@link SocketType#SUB} is used, the received topic is stored in the {@link ZeroMqHeaders#TOPIC}. + * + * @author Artem Bilan + * + * @since 5.4 + */ +@ManagedResource +@IntegrationManagedResource +public class ZeroMqMessageProducer extends MessageProducerSupport { + + public static final Duration DEFAULT_CONSUME_DELAY = Duration.ofSeconds(1); + + private static final List VALID_SOCKET_TYPES = + Arrays.asList(SocketType.PAIR, SocketType.PULL, SocketType.SUB); + + private final Scheduler consumerScheduler = Schedulers.newSingle("zeroMqMessageProducerScheduler"); + + private final AtomicInteger bindPort = new AtomicInteger(); + + private final ZContext context; + + private final SocketType socketType; + + private InboundMessageMapper messageMapper; + + private Consumer socketConfigurer = (socket) -> { }; + + private Duration consumeDelay = DEFAULT_CONSUME_DELAY; + + private String[] topics = { "" }; // Equivalent to ZMQ#SUBSCRIPTION_ALL + + private boolean receiveRaw; + + @Nullable + private String connectUrl; + + private volatile Mono socketMono; + + private volatile boolean active; + + public ZeroMqMessageProducer(ZContext context) { + this(context, SocketType.PAIR); + } + + public ZeroMqMessageProducer(ZContext context, SocketType socketType) { + Assert.notNull(context, "'context' must not be null"); + Assert.state(VALID_SOCKET_TYPES.contains(socketType), + () -> "'socketType' can only be one of the: " + VALID_SOCKET_TYPES); + this.context = context; + this.socketType = socketType; + } + + /** + * Specify a {@link Duration} to delay consumption when no data received. + * @param consumeDelay the {@link Duration} to delay consumption when empty; + * defaults to {@link #DEFAULT_CONSUME_DELAY}. + */ + public void setConsumeDelay(Duration consumeDelay) { + Assert.notNull(consumeDelay, "'consumeDelay' must not be null"); + this.consumeDelay = consumeDelay; + } + + /** + * Provide an {@link InboundMessageMapper} to convert a consumed data into a message to produce. + * Ignored when {@link #setReceiveRaw(boolean)} is {@code true}. + * @param messageMapper the {@link InboundMessageMapper} to use. + */ + public void setMessageMapper(InboundMessageMapper messageMapper) { + Assert.notNull(messageMapper, "'messageMapper' must not be null"); + this.messageMapper = messageMapper; + } + + /** + * Provide a {@link MessageConverter} (as an alternative to {@link #messageMapper}) + * for converting a consumed data into a message to produce. + * Ignored when {@link #setReceiveRaw(boolean)} is {@code true}. + * @param messageConverter the {@link MessageConverter} to use. + */ + public void setMessageConverter(MessageConverter messageConverter) { + setMessageMapper(new ConvertingBytesMessageMapper(messageConverter)); + } + + /** + * Whether raw {@link ZMsg} is present as a payload of message to produce or + * it is fully converted to a {@link Message} including {@link ZeroMqHeaders#TOPIC} header (if any). + * @param receiveRaw to convert from {@link ZMsg} or not; defaults to convert. + */ + public void setReceiveRaw(boolean receiveRaw) { + this.receiveRaw = receiveRaw; + } + + /** + * Provide a {@link Consumer} to configure a socket with arbitrary options, like security. + * @param socketConfigurer the configurer for socket options. + */ + public void setSocketConfigurer(Consumer socketConfigurer) { + Assert.notNull(socketConfigurer, "'socketConfigurer' must not be null"); + this.socketConfigurer = socketConfigurer; + } + + /** + * Specify topics the {@link SocketType#SUB} socket is going to use for subscription. + * It is ignored for all other {@link SocketType}s supported. + * @param topics the topics to use. + */ + public void setTopics(String... topics) { + Assert.notNull(topics, "'topics' cannot be null"); + Assert.noNullElements(topics, "'topics' must not contain null elements"); + this.topics = Arrays.copyOf(topics, topics.length); + } + + /** + * Configure an URL for {@link org.zeromq.ZMQ.Socket#connect(String)}. + * Mutually exclusive with the {@link #setBindPort(int)}. + * @param connectUrl the URL to connect ZeroMq socket to. + */ + public void setConnectUrl(@Nullable String connectUrl) { + this.connectUrl = connectUrl; + } + + /** + * Configure a port for TCP protocol binding via {@link org.zeromq.ZMQ.Socket#bind(String)}. + * Mutually exclusive with the {@link #setConnectUrl(String)}. + * @param port the port to bind ZeroMq socket to over TCP. + */ + public void setBindPort(int port) { + Assert.isTrue(port > 0, "'port' must not be zero or negative"); + this.bindPort.set(port); + } + + /** + * Return the port a socket is bound or 0 if this message producer has not been started yet + * or the socket is connected - not bound. + * @return the port for a socket or 0. + */ + public int getBoundPort() { + return this.bindPort.get(); + } + + @Override + public String getComponentType() { + return "zeromq:inbound-channel-adapter"; + } + + @Override + protected void onInit() { + super.onInit(); + Assert.state(this.connectUrl == null || this.bindPort.get() == 0, + "Only one of the 'connectUrl' or `bindPort` must be provided on none"); + if (this.messageMapper == null && !this.receiveRaw) { + ConfigurableCompositeMessageConverter messageConverter = new ConfigurableCompositeMessageConverter(); + messageConverter.setBeanFactory(getBeanFactory()); + messageConverter.afterPropertiesSet(); + this.messageMapper = new ConvertingBytesMessageMapper(messageConverter); + } + } + + @ManagedOperation + public void subscribeToTopics(String... topics) { + Assert.state(SocketType.SUB.equals(this.socketType), "Only SUB socket can accept a subscription option."); + Assert.state(this.active, "This message producer is not active to accept a new subscription."); + + Flux.fromArray(topics) + .flatMap((topic) -> + this.socketMono.doOnNext((socket) -> socket.subscribe(topic))) + .subscribe(); + } + + @ManagedOperation + public void unsubscribeFromTopics(String... topics) { + Assert.state(SocketType.SUB.equals(this.socketType), "Only SUB socket can accept a unsubscription option."); + Assert.state(this.active, "This message producer is not active to cancel a subscription."); + + Flux.fromArray(topics) + .flatMap((topic) -> + this.socketMono.doOnNext((socket) -> socket.unsubscribe(topic))) + .subscribe(); + } + + @Override + protected void doStart() { + this.socketMono = + Mono.just(this.context.createSocket(this.socketType)) + .publishOn(this.consumerScheduler) + .doOnNext(this.socketConfigurer) + .doOnNext((socket) -> { + if (SocketType.SUB.equals(this.socketType)) { + for (String topic : this.topics) { + socket.subscribe(topic); + } + } + }) + .doOnNext((socket) -> { + if (this.connectUrl != null) { + socket.connect(this.connectUrl); + } + else { + this.bindPort.set(bindSocket(socket, this.bindPort.get())); + } + }) + .cache() + .publishOn(this.consumerScheduler); + + this.active = true; + + Flux> dataFlux = + this.socketMono + .flatMap((socket) -> { + if (isRunning()) { + ZMsg msg = ZMsg.recvMsg(socket, false); + if (msg != null) { + return Mono.just(msg); + } + } + return Mono.empty(); + }) + .publishOn(Schedulers.boundedElastic()) + .transform((msgMono) -> this.receiveRaw ? mapRaw(msgMono) : convertMessage(msgMono)) + .doOnError((error) -> logger.error("Error processing ZeroMQ message in the " + this, error)) + .repeatWhenEmpty((repeat) -> + this.active ? repeat.delayElements(this.consumeDelay) : repeat) + .repeat(() -> this.active) + .doOnComplete(this.consumerScheduler::dispose); + + subscribeToPublisher(dataFlux); + } + + private Mono> mapRaw(Mono msgMono) { + return msgMono.map((msg) -> getMessageBuilderFactory().withPayload(msg).build()); + } + + private Mono> convertMessage(Mono msgMono) { + return msgMono.map((msg) -> { + Map headers = null; + if (msg.size() > 1) { + headers = Collections.singletonMap(ZeroMqHeaders.TOPIC, msg.unwrap().getString(ZMQ.CHARSET)); + } + return this.messageMapper.toMessage(msg.getLast().getData(), headers); // NOSONAR + }); + } + + @Override + protected void doStop() { + this.active = false; + this.socketMono.doOnNext(ZMQ.Socket::close).subscribe(); + } + + @Override + public void destroy() { + this.active = false; + super.destroy(); + this.socketMono.doOnNext(ZMQ.Socket::close).block(); + } + + 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; + } + } + +} diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/package-info.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/package-info.java new file mode 100644 index 0000000000..db224238af --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes for inbound channel adapters over ZeroMQ. + */ +package org.springframework.integration.zeromq.inbound; diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java new file mode 100644 index 0000000000..4a7ccd546f --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java @@ -0,0 +1,214 @@ +/* + * 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.outbound; + +import java.util.Arrays; +import java.util.List; +import java.util.function.Consumer; + +import org.zeromq.SocketType; +import org.zeromq.ZContext; +import org.zeromq.ZFrame; +import org.zeromq.ZMQ; +import org.zeromq.ZMsg; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.handler.AbstractReactiveMessageHandler; +import org.springframework.integration.mapping.ConvertingBytesMessageMapper; +import org.springframework.integration.mapping.OutboundMessageMapper; +import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.util.Assert; + +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; +import reactor.core.scheduler.Schedulers; +import zmq.socket.pubsub.Pub; + +/** + * The {@link AbstractReactiveMessageHandler} implementation for publishing messages over ZeroMq socket. + * Only {@link SocketType#PAIR}, {@link SocketType#PUB} and {@link SocketType#PUSH} are supported. + * This component is only connecting (no Binding) to another side, e.g. ZeroMq proxy. + *

+ * When the {@link SocketType#PUB} is used, the {@link #topicExpression} is evaluated against a + * request message to inject a topic frame into a ZeroMq message if it is not {@code null}. + * The subscriber side must receive the topic frame first before parsing the actual data. + *

+ * When the payload of the request message is a {@link ZMsg}, no any conversion and topic extraction happen: + * the {@link ZMsg} is sent into a socket as is and it is not destroyed for possible further reusing. + * + * @author Artem Bilan + * + * @since 5.4 + */ +public class ZeroMqMessageHandler extends AbstractReactiveMessageHandler { + + private static final List VALID_SOCKET_TYPES = + Arrays.asList(SocketType.PAIR, SocketType.PUSH, SocketType.PUB); + + private final Scheduler publisherScheduler = Schedulers.newSingle("zeroMqMessageHandlerScheduler"); + + private final Mono socketMono; + + private OutboundMessageMapper messageMapper; + + private Consumer socketConfigurer = (socket) -> { }; + + private Expression topicExpression = new LiteralExpression(null); + + private EvaluationContext evaluationContext; + + private volatile boolean initialized; + + /** + * Create an instance based on the provided {@link ZContext} and connection string. + * @param context the {@link ZContext} to use for creating sockets. + * @param connectUrl the URL to connect the socket to. + */ + public ZeroMqMessageHandler(ZContext context, String connectUrl) { + this(context, connectUrl, SocketType.PAIR); + } + + /** + * Create an instance based on the provided {@link ZContext}, connection string and {@link SocketType}. + * @param context the {@link ZContext} to use for creating sockets. + * @param connectUrl the URL to connect the socket to. + * @param socketType the {@link SocketType} to use; + * only {@link SocketType#PAIR}, {@link SocketType#PUB} and {@link SocketType#PUSH} are supported. + */ + public ZeroMqMessageHandler(ZContext context, String connectUrl, SocketType socketType) { + Assert.notNull(context, "'context' must not be null"); + Assert.hasText(connectUrl, "'connectUrl' must not be empty"); + Assert.state(VALID_SOCKET_TYPES.contains(socketType), + () -> "'socketType' can only be one of the: " + VALID_SOCKET_TYPES); + this.socketMono = + Mono.just(context.createSocket(socketType)) + .publishOn(this.publisherScheduler) + .doOnNext(this.socketConfigurer) + .doOnNext((socket) -> socket.connect(connectUrl)) + .cache() + .publishOn(this.publisherScheduler); + } + + /** + * Provide an {@link OutboundMessageMapper} to convert a request message into {@code byte[]} + * for sending into ZeroMq socket. + * Ignored when {@link Message#getPayload()} is an instance of {@link ZMsg}. + * @param messageMapper the {@link OutboundMessageMapper} to use. + */ + public void setMessageMapper(OutboundMessageMapper messageMapper) { + Assert.notNull(messageMapper, "'messageMapper' must not be null"); + this.messageMapper = messageMapper; + } + + /** + * Provide a {@link MessageConverter} (as an alternative to {@link #messageMapper}) + * for converting a request message into {@code byte[]} for sending into ZeroMq socket. + * Ignored when {@link Message#getPayload()} is an instance of {@link ZMsg}. + * @param messageConverter the {@link MessageConverter} to use. + */ + public void setMessageConverter(MessageConverter messageConverter) { + setMessageMapper(new ConvertingBytesMessageMapper(messageConverter)); + } + + /** + * Provide a {@link Consumer} to configure a socket with arbitrary options, like security. + * @param socketConfigurer the configurer for socket options. + */ + public void setSocketConfigurer(Consumer socketConfigurer) { + Assert.notNull(socketConfigurer, "'socketConfigurer' must not be null"); + this.socketConfigurer = socketConfigurer; + } + + /** + * Specify a topic the {@link SocketType#PUB} socket is going to use for distributing messages into the + * subscriptions. It is ignored for all other {@link SocketType}s supported. + * @param topic the topic to use. + */ + public void setTopic(String topic) { + setTopicExpression(new LiteralExpression(topic)); + } + + /** + * Specify a SpEL expression to evaluate a topic a {@link SocketType#PUB} + * is going to use for distributing messages into the + * subscriptions.It is ignored for all other {@link SocketType}s supported. + * @param topicExpression the expression to evaluate topic for publishing. + */ + public void setTopicExpression(Expression topicExpression) { + Assert.notNull(topicExpression, "'topicExpression' must not be null"); + this.topicExpression = topicExpression; + } + + @Override + public String getComponentType() { + return "zeromq:outbound-channel-adapter"; + } + + @Override + protected void onInit() { + super.onInit(); + BeanFactory beanFactory = getBeanFactory(); + this.evaluationContext = ExpressionUtils.createSimpleEvaluationContext(beanFactory); + if (this.messageMapper == null) { + ConfigurableCompositeMessageConverter messageConverter = new ConfigurableCompositeMessageConverter(); + messageConverter.setBeanFactory(beanFactory); + messageConverter.afterPropertiesSet(); + this.messageMapper = new ConvertingBytesMessageMapper(messageConverter); + } + this.socketMono.subscribe(); + this.initialized = true; + } + + @Override + protected Mono handleMessageInternal(Message message) { + Assert.state(this.initialized, "the message handler is not initialized yet or already destroyed"); + return this.socketMono + .doOnNext((socket) -> { + ZMsg msg; + if (message.getPayload() instanceof ZMsg) { + msg = (ZMsg) message.getPayload(); + } + else { + msg = new ZMsg(); + msg.add(this.messageMapper.fromMessage(message)); + if (socket.base() instanceof Pub) { + String topic = this.topicExpression.getValue(this.evaluationContext, message, String.class); + if (topic != null) { + msg.wrap(new ZFrame(topic)); + } + } + } + msg.send(socket, false); + }) + .then(); + } + + @Override + public void destroy() { + this.initialized = false; + super.destroy(); + this.socketMono.doOnNext(ZMQ.Socket::close).block(); + this.publisherScheduler.dispose(); + } + +} diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/package-info.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/package-info.java new file mode 100644 index 0000000000..6db6048387 --- /dev/null +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes for outbound channel adapters over ZeroMQ. + */ +package org.springframework.integration.zeromq.outbound; 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 index b73dba86ff..66a22ce0d3 100644 --- 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 @@ -114,8 +114,8 @@ public class ZeroMqChannelTests { await().until(() -> proxy.getBackendPort() > 0); ZMQ.Socket captureSocket = CONTEXT.createSocket(SocketType.SUB); - captureSocket.connect(proxy.getCaptureAddress()); captureSocket.subscribe(ZMQ.SUBSCRIPTION_ALL); + captureSocket.connect(proxy.getCaptureAddress()); ZeroMqChannel channel = new ZeroMqChannel(CONTEXT); channel.setConnectUrl("tcp://localhost:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort()); @@ -170,10 +170,13 @@ public class ZeroMqChannelTests { ZeroMqChannel channel2 = new ZeroMqChannel(CONTEXT, true); channel2.setConnectUrl("tcp://localhost:" + proxy.getFrontendPort() + ':' + proxy.getBackendPort()); channel2.setBeanName("testChannel5"); - channel.setConsumeDelay(Duration.ofMillis(10)); + channel2.setConsumeDelay(Duration.ofMillis(10)); channel2.afterPropertiesSet(); - channel.subscribe(received::offer); + channel2.subscribe(received::offer); + + // Give it some time to connect and subscribe + Thread.sleep(1000); GenericMessage testMessage = new GenericMessage<>("test1"); assertThat(channel.send(testMessage)).isTrue(); @@ -187,6 +190,7 @@ public class ZeroMqChannelTests { assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull(); channel.destroy(); + channel2.destroy(); proxy.stop(); } diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/dsl/ZeroMqDslTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/dsl/ZeroMqDslTests.java new file mode 100644 index 0000000000..ff65b8fc39 --- /dev/null +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/dsl/ZeroMqDslTests.java @@ -0,0 +1,153 @@ +/* + * 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.dsl; + +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.Test; +import org.zeromq.SocketType; +import org.zeromq.ZContext; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.Transformers; +import org.springframework.integration.dsl.context.IntegrationFlowContext; +import org.springframework.integration.zeromq.ZeroMqHeaders; +import org.springframework.integration.zeromq.ZeroMqProxy; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.util.SocketUtils; + +/** + * @author Artem Bilan + * + * @since 5.4 + */ +@SpringJUnitConfig +@DirtiesContext +public class ZeroMqDslTests { + + private static final int PROXY_PUB_PORT = SocketUtils.findAvailableTcpPort(); + + @Autowired + ZContext context; + + @Autowired + @Qualifier("publishToZeroMqPubSubFlow.input") + MessageChannel publishToZeroMqPubSubFlowInput; + + @Autowired + ZeroMqProxy subPubZeroMqProxy; + + @Autowired + ZeroMqProxy pullPushZeroMqProxy; + + @Autowired + IntegrationFlowContext integrationFlowContext; + + @Test + void testZeroMqDslIntegration() throws InterruptedException { + BlockingQueue> results = new LinkedBlockingQueue<>(); + + await().until(() -> this.subPubZeroMqProxy.getBackendPort() > 0); + + for (int i = 0; i < 2; i++) { + IntegrationFlow consumerFlow = + IntegrationFlows.from( + ZeroMq.inboundChannelAdapter(this.context, SocketType.SUB) + .connectUrl("tcp://localhost:" + this.subPubZeroMqProxy.getBackendPort()) + .topics("someTopic") + .consumeDelay(Duration.ofMillis(100))) + .channel(ZeroMq.zeroMqChannel(this.context).zeroMqProxy(this.pullPushZeroMqProxy)) + .transform(Transformers.objectToString()) + .handle(results::offer) + .get(); + + this.integrationFlowContext.registration(consumerFlow).register(); + } + + // Give it some time to connect and subscribe + Thread.sleep(2000); + + this.publishToZeroMqPubSubFlowInput.send(new GenericMessage<>("test")); + + Message message = results.poll(10, TimeUnit.SECONDS); + assertThat(message).isNotNull() + .extracting(Message::getPayload) + .isEqualTo("test"); + + assertThat(message.getHeaders()).containsEntry(ZeroMqHeaders.TOPIC, "someTopic"); + + message = results.poll(1, TimeUnit.SECONDS); + assertThat(message).isNotNull() + .extracting(Message::getPayload) + .isEqualTo("test"); + + // With Pub/Sub channel we would have 4 messages. + assertThat(results.poll(1, TimeUnit.SECONDS)).isNull(); + + this.integrationFlowContext.getRegistry() + .values() + .forEach(IntegrationFlowContext.IntegrationFlowRegistration::destroy); + } + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + ZContext context() { + return new ZContext(); + } + + @Bean + ZeroMqProxy subPubZeroMqProxy() { + ZeroMqProxy zeroMqProxy = new ZeroMqProxy(context(), ZeroMqProxy.Type.SUB_PUB); + zeroMqProxy.setFrontendPort(PROXY_PUB_PORT); + return zeroMqProxy; + } + + @Bean + ZeroMqProxy pullPushZeroMqProxy() { + return new ZeroMqProxy(context()); + } + + @Bean + IntegrationFlow publishToZeroMqPubSubFlow() { + return flow -> + flow.handle(ZeroMq.outboundChannelAdapter(context(), "tcp://localhost:" + PROXY_PUB_PORT, + SocketType.PUB) + .topic("someTopic")); + } + + } + +} diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java new file mode 100644 index 0000000000..d367d85926 --- /dev/null +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java @@ -0,0 +1,143 @@ +/* + * 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.inbound; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.mockito.Mockito.mock; + +import java.time.Duration; + +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Test; +import org.zeromq.SocketType; +import org.zeromq.ZContext; +import org.zeromq.ZFrame; +import org.zeromq.ZMQ; +import org.zeromq.ZMsg; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.messaging.support.GenericMessage; + +import reactor.test.StepVerifier; + +/** + * @author Artem Bilan + * + * @since 5.4 + */ +public class ZeroMqMessageProducerTests { + + private static final ZContext CONTEXT = new ZContext(); + + @AfterAll + static void tearDown() { + CONTEXT.close(); + } + + @Test + void testMessageProducerForPair() { + FluxMessageChannel outputChannel = new FluxMessageChannel(); + + StepVerifier stepVerifier = + StepVerifier.create(outputChannel) + .assertNext((message) -> assertThat(message.getPayload()).isEqualTo("test")) + .assertNext((message) -> assertThat(message.getPayload()).isEqualTo("test2")) + .thenCancel() + .verifyLater(); + + ZeroMqMessageProducer messageProducer = new ZeroMqMessageProducer(CONTEXT); + messageProducer.setOutputChannel(outputChannel); + messageProducer.setMessageMapper((object, headers) -> new GenericMessage<>(new String(object))); + messageProducer.setConsumeDelay(Duration.ofMillis(10)); + messageProducer.setBeanFactory(mock(BeanFactory.class)); + messageProducer.afterPropertiesSet(); + messageProducer.start(); + + ZMQ.Socket socket = CONTEXT.createSocket(SocketType.PAIR); + + await().until(() -> messageProducer.getBoundPort() > 0); + + socket.connect("tcp://localhost:" + messageProducer.getBoundPort()); + + socket.send("test"); + socket.send("test2"); + + stepVerifier.verify(); + + messageProducer.destroy(); + socket.close(); + } + + @Test + void testMessageProducerForPubSubReceiveRaw() throws InterruptedException { + String socketAddress = "inproc://messageProducer.test"; + ZMQ.Socket socket = CONTEXT.createSocket(SocketType.PUB); + socket.bind(socketAddress); + + FluxMessageChannel outputChannel = new FluxMessageChannel(); + + StepVerifier stepVerifier = + StepVerifier.create(outputChannel) + .assertNext((message) -> + assertThat(message.getPayload()) + .asInstanceOf(InstanceOfAssertFactories.type(ZMsg.class)) + .extracting(ZMsg::unwrap) + .isEqualTo(new ZFrame("testTopic"))) + .assertNext((message) -> + assertThat(message.getPayload()) + .asInstanceOf(InstanceOfAssertFactories.type(ZMsg.class)) + .extracting(ZMsg::unwrap) + .isEqualTo(new ZFrame("otherTopic"))) + .thenCancel() + .verifyLater(); + + ZeroMqMessageProducer messageProducer = new ZeroMqMessageProducer(CONTEXT, SocketType.SUB); + messageProducer.setOutputChannel(outputChannel); + messageProducer.setTopics("test"); + messageProducer.setReceiveRaw(true); + messageProducer.setConnectUrl(socketAddress); + messageProducer.setConsumeDelay(Duration.ofMillis(10)); + messageProducer.setBeanFactory(mock(BeanFactory.class)); + messageProducer.afterPropertiesSet(); + messageProducer.start(); + + // Give it some time to connect and subscribe + Thread.sleep(2000); + + ZMsg msg = ZMsg.newStringMsg("test"); + msg.wrap(new ZFrame("testTopic")); + msg.send(socket); + + messageProducer.subscribeToTopics("other"); + + // Give it some time to connect and subscribe + Thread.sleep(2000); + + msg = ZMsg.newStringMsg("test"); + msg.wrap(new ZFrame("otherTopic")); + msg.send(socket); + + stepVerifier.verify(Duration.ofSeconds(10)); + + messageProducer.destroy(); + socket.close(); + } + +} diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java new file mode 100644 index 0000000000..8af81d024d --- /dev/null +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java @@ -0,0 +1,138 @@ +/* + * 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.outbound; + + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.mockito.Mockito.mock; + +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.zeromq.ZMsg; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; +import org.springframework.integration.zeromq.ZeroMqProxy; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.ByteArrayMessageConverter; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Artem Bilan + * + * @since 5.4 + */ +public class ZeroMqMessageHandlerTests { + + private static final ZContext CONTEXT = new ZContext(); + + @AfterAll + static void tearDown() { + CONTEXT.close(); + } + + @Test + void testMessageHandlerForPair() { + String socketAddress = "inproc://messageHandler.test"; + ZMQ.Socket socket = CONTEXT.createSocket(SocketType.PAIR); + socket.bind(socketAddress); + + ZeroMqMessageHandler messageHandler = new ZeroMqMessageHandler(CONTEXT, socketAddress); + messageHandler.setBeanFactory(mock(BeanFactory.class)); + messageHandler.afterPropertiesSet(); + + Message testMessage = new GenericMessage<>("test"); + messageHandler.handleMessage(testMessage).subscribe(); + + assertThat(socket.recvStr()).isEqualTo("test"); + + messageHandler.handleMessage(new GenericMessage<>(ZMsg.newStringMsg("test2"))).subscribe(); + + assertThat(socket.recvStr()).isEqualTo("test2"); + + messageHandler.destroy(); + socket.close(); + } + + @Test + void testMessageHandlerForPubSub() throws InterruptedException { + ZMQ.Socket subSocket = CONTEXT.createSocket(SocketType.SUB); + subSocket.setReceiveTimeOut(10_000); + int port = subSocket.bindToRandomPort("tcp://*"); + subSocket.subscribe("test"); + + ZeroMqMessageHandler messageHandler = + new ZeroMqMessageHandler(CONTEXT, "tcp://localhost:" + port, SocketType.PUB); + messageHandler.setBeanFactory(mock(BeanFactory.class)); + messageHandler.setTopicExpression( + new FunctionExpression>((message) -> message.getHeaders().get("topic"))); + messageHandler.setMessageMapper(new EmbeddedJsonHeadersMessageMapper()); + messageHandler.afterPropertiesSet(); + + // Give it some time to bind and subscribe + Thread.sleep(2000); + + Message testMessage = MessageBuilder.withPayload("test").setHeader("topic", "testTopic").build(); + messageHandler.handleMessage(testMessage).subscribe(); + + ZMsg msg = ZMsg.recvMsg(subSocket); + assertThat(msg).isNotNull(); + assertThat(msg.unwrap().getString(ZMQ.CHARSET)).isEqualTo("testTopic"); + Message capturedMessage = new EmbeddedJsonHeadersMessageMapper().toMessage(msg.getFirst().getData()); + assertThat(capturedMessage).isEqualTo(testMessage); + + msg.destroy(); + messageHandler.destroy(); + subSocket.close(); + } + + @Test + void testMessageHandlerForPushPullOverProxy() { + ZeroMqProxy proxy = new ZeroMqProxy(CONTEXT); + proxy.setBeanName("pullPushProxy"); + proxy.afterPropertiesSet(); + proxy.start(); + + await().until(() -> proxy.getBackendPort() > 0); + + ZMQ.Socket pullSocket = CONTEXT.createSocket(SocketType.PULL); + pullSocket.setReceiveTimeOut(10_000); + pullSocket.connect("tcp://localhost:" + proxy.getBackendPort()); + + ZeroMqMessageHandler messageHandler = + new ZeroMqMessageHandler(CONTEXT, "tcp://localhost:" + proxy.getFrontendPort(), SocketType.PUSH); + messageHandler.setBeanFactory(mock(BeanFactory.class)); + messageHandler.setMessageConverter(new ByteArrayMessageConverter()); + messageHandler.afterPropertiesSet(); + + Message testMessage = new GenericMessage<>("test".getBytes()); + messageHandler.handleMessage(testMessage).subscribe(); + + assertThat(pullSocket.recvStr()).isEqualTo("test"); + + messageHandler.destroy(); + pullSocket.close(); + proxy.stop(); + } + +} diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index b4c057dbfe..9848ad6f57 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -44,7 +44,7 @@ See <<./jdbc.adoc#jdbc-lock-registry,JDBC implementation>> for more information. [[x5.4-zeromq]] ==== ZeroMQ Support -A `ZeroMqChannel` has been introduced. +`ZeroMqChannel`, `ZeroMqMessageHandler` and `ZeroMqMessageProducer` have been introduced. See <<./zeromq.adoc#zeromq,ZeroMQ Support>> for more information. [[x5.4-general]] diff --git a/src/reference/asciidoc/zeromq.adoc b/src/reference/asciidoc/zeromq.adoc index 40ddbace35..c76f54c4e6 100644 --- a/src/reference/asciidoc/zeromq.adoc +++ b/src/reference/asciidoc/zeromq.adoc @@ -102,3 +102,114 @@ ZeroMqChannel zeroMqPubSubChannel(ZContext context) { } ---- ==== + +[[zeromq-inbound-channel-adapter]] +=== ZeroMq Inbound Channel Adapter + +The `ZeroMqMessageProducer` is a `MessageProducerSupport` implementation with reactive semantics. +It constantly reads the data from a ZeroMq socket in a non-blocking manner and publishes the messages to an infinite `Flux` which is subscribed to by a `FluxMessageChannel` or explicitly in the `start()` method, if the output channel is not reactive. +When no data are received on the socket, a `consumeDelay` (defaults to 1 second) is applied before the next read attempt. + + +Only `SocketType.PAIR`, `SocketType.PULL` and `SocketType.SUB` are supported by the `ZeroMqMessageProducer`. +This component can connect to the remote socket or bind onto TCP protocol with the provided or random port. +The actual port can be obtained via `getBoundPort()` after this component is started and ZeroMq socket is bound. +The socket options (e.g. security or write timeout) can be configured via `setSocketConfigurer(Consumer socketConfigurer)` callback. + +If the `receiveRaw` option is set to `true`, a `ZMsg`, consumed from the socket, is sent as is in the payload of the produced `Message`: it's up to the downstream flow to parse and convert the `ZMsg`. +Otherwise an `InboundMessageMapper` is used to convert the consumed data into a `Message`. +If the received `ZMsg` is multi-frame, the first frame is treated as the `ZeroMqHeaders.TOPIC` header this ZeroMq message was published to. + +With `SocketType.SUB`, the `ZeroMqMessageProducer` uses the provided `topics` option for subscriptions; defaults to subscribe to all. +Subscriptions can be adjusted at runtime using `subscribeToTopics()` and `unsubscribeFromTopics()` `@ManagedOperation` s. + +Here is a sample of `ZeroMqMessageProducer` configuration: + +==== +[source,java] +---- +@Bean +ZeroMqMessageProducer zeroMqMessageProducer(ZContext context, MessageChannel outputChannel) { + ZeroMqMessageProducer messageProducer = new ZeroMqMessageProducer(context, SocketType.SUB); + messageProducer.setOutputChannel(outputChannel); + messageProducer.setTopics("some"); + messageProducer.setReceiveRaw(true); + messageProducer.setBindPort(7070); + messageProducer.setConsumeDelay(Duration.ofMillis(100)); + return messageProducer; +} +---- +==== + +[[zeromq-outbound-channel-adapter]] +=== ZeroMq Outbound Channel Adapter + +The `ZeroMqMessageHandler` is a `ReactiveMessageHandler` implementation to produce publish messages into a ZeroMq socket. +Only `SocketType.PAIR`, `SocketType.PUSH` and `SocketType.PUB` are supported. +The `ZeroMqMessageHandler` only supports connecting the ZeroMq socket; binding is not supported. +When the `SocketType.PUB` is used, the `topicExpression` is evaluated against a request message to inject a topic frame into a ZeroMq message if it is not null. +The subscriber side (`SocketType.SUB`) must receive the topic frame first before parsing the actual data. +When the payload of the request message is a `ZMsg`, no conversion or topic extraction is performed: the `ZMsg` is sent into a socket as is and it is not destroyed for possible further reuse. +Otherwise an `OutboundMessageMapper` is used to convert a request message (or just its payload) into a ZeroMq frame to publish. +By default a `ConvertingBytesMessageMapper` is used supplied with a `ConfigurableCompositeMessageConverter`. +The socket options (e.g. security or write timeout) can be configured via `setSocketConfigurer(Consumer socketConfigurer)` callback. + +Here is a sample of `ZeroMqMessageHandler` configuration: + +==== +[source,java] +---- +@Bean +@ServiceActivator(inputChannel = "zeroMqPublisherChannel") +ZeroMqMessageHandler zeroMqMessageHandler(ZContext context) { + ZeroMqMessageHandler messageHandler = + new ZeroMqMessageHandler(context, "tcp://localhost:6060", SocketType.PUB); + messageHandler.setTopicExpression( + new FunctionExpression>((message) -> message.getHeaders().get("topic"))); + messageHandler.setMessageMapper(new EmbeddedJsonHeadersMessageMapper()); +} +---- +==== + +[[zeromq-dsl]] +=== ZeroMq Java DSL Support + +The `spring-integration-zeromq` provide a convenient Java DSL fluent API via `ZeroMq` factory and `IntegrationComponentSpec` implementations for the components mentioned above. + +This is a sample of Java DSL for `ZeroMqChannel`: + +==== +[source,java] +---- +.channel(ZeroMq.zeroMqChannel(this.context) + .connectUrl("tcp://localhost:6001:6002") + .consumeDelay(Duration.ofMillis(100))) +} +---- +==== + +The Inbound Channel Adapter for ZeroMq Java DSL is: + +==== +[source,java] +---- +IntegrationFlows.from( + ZeroMq.inboundChannelAdapter(this.context, SocketType.SUB) + .connectUrl("tcp://localhost:9000") + .topics("someTopic") + .receiveRaw(true) + .consumeDelay(Duration.ofMillis(100))) +} +---- +==== + +The Outbound Channel Adapter for ZeroMq Java DSL is: + +==== +[source,java] +---- +.handle(ZeroMq.outboundChannelAdapter(this.context, "tcp://localhost:9001", SocketType.PUB) + .topicFunction(message -> message.getHeaders().get("myTopic"))) +} +---- +====