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

INT-3045: Add in & out ZeroMq channel adapters (#3388)

Browse Source 
* 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 <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2020-09-25 13:46:01 -04:00
committed by GitHub
parent 9d34cfd4dd
commit a62a7d1ddd
24 changed files with 1864 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
spring-integration-core/src/main/java/org/springframework/integration/dsl/ReactiveMessageHandlerSpec.java Normal file
View File

@@ -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<S extends ReactiveMessageHandlerSpec<S, H>, H extends ReactiveMessageHandler>
extends MessageHandlerSpec<S, ReactiveMessageHandlerAdapter>
implements ComponentsRegistration {
protected final H reactiveMessageHandler; // NOSONAR - final
protected ReactiveMessageHandlerSpec(H reactiveMessageHandler) {
this.reactiveMessageHandler = reactiveMessageHandler;
this.target = new ReactiveMessageHandlerAdapter(this.reactiveMessageHandler);
}
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.reactiveMessageHandler, null);
}
}

77
spring-integration-core/src/main/java/org/springframework/integration/mapping/ConvertingBytesMessageMapper.java Normal file
View File

@@ -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}.
* <p>
* The {@link MessageConverter} must not return {@code null} from its
* {@link MessageConverter#fromMessage(Message, Class)} and {@link MessageConverter#toMessage(Object, MessageHeaders)}
* methods.
* <p>
* 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<String, Object> 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;
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/support/json/EmbeddedJsonHeadersMessageMapper.java
View File

@@ -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) {

7
spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java
View File

@@ -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()

57
spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java Normal file
View File

@@ -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<MimeType> {
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());
}
}

26
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java
View File

@@ -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<ReactiveMongoDbMessageHandlerSpec, ReactiveMessageHandlerAdapter>
extends ReactiveMessageHandlerSpec<ReactiveMongoDbMessageHandlerSpec, ReactiveMongoDbStoringMessageHandler>
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<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.messageHandler, null);
}
}

38
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqHeaders.java Normal file
View File

@@ -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() {
}
}

2
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java
View File

@@ -97,7 +97,7 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw
private String beanName;
private boolean autoStartup;
private boolean autoStartup = true;
private int phase;

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

@@ -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<Integer> prepareProxyMono() {
if (this.zeroMqProxy != null) {
return Mono.fromCallable(() -> this.zeroMqProxy.getBackendPort())
.filter((proxyPort) -> proxyPort > 0)
.repeatWhenEmpty(100, (repeat) -> repeat.delayElements(Duration.ofMillis(100))) // NOSONAR
.doOnNext((proxyPort) ->
setConnectUrl("tcp://localhost:" + this.zeroMqProxy.getFrontendPort() +
':' + this.zeroMqProxy.getBackendPort()))
.doOnError((error) ->
logger.error("The provided '" + this.zeroMqProxy + "' has not been started", error))
.cache();
}
else {
return Mono.empty();
}
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<ZMQ.Socket> prepareSendSocketMono(Supplier<String> 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<ZMQ.Socket> 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<ZMQ.Socket> subscribeSocketConfigurer) {
Assert.notNull(subscribeSocketConfigurer, "'subscribeSocketConfigurer' must not be null");
this.subscribeSocketConfigurer = subscribeSocketConfigurer;

97
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMq.java Normal file
View File

@@ -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() {
}
}

109
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqChannelSpec.java Normal file
View File

@@ -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<ZeroMqChannelSpec, ZeroMqChannel> {
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<ZMQ.Socket> 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<ZMQ.Socket> subscribeSocketConfigurer) {
this.channel.setSubscribeSocketConfigurer(subscribeSocketConfigurer);
return this;
}
}

142
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java Normal file
View File

@@ -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<ZeroMqMessageHandlerSpec, ZeroMqMessageHandler> {
/**
* 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<byte[]> 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<ZMQ.Socket> 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<Message<?>, 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;
}
}

133
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java Normal file
View File

@@ -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<ZeroMqMessageProducerSpec, ZeroMqMessageProducer> {
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<byte[]> 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<ZMQ.Socket> 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;
}
}

4
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes for supporting ZeroMQ component via Java DSL.
*/
package org.springframework.integration.zeromq.dsl;

322
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java Normal file
View File

@@ -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.
* <p>
* 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<SocketType> 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<byte[]> messageMapper;
private Consumer<ZMQ.Socket> 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<ZMQ.Socket> 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<byte[]> 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<ZMQ.Socket> 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<? extends Message<?>> 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<Message<?>> mapRaw(Mono<ZMsg> msgMono) {
return msgMono.map((msg) -> getMessageBuilderFactory().withPayload(msg).build());
}
private Mono<Message<?>> convertMessage(Mono<ZMsg> msgMono) {
return msgMono.map((msg) -> {
Map<String, Object> 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;
}
}
}

4
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes for inbound channel adapters over ZeroMQ.
*/
package org.springframework.integration.zeromq.inbound;

214
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java Normal file
View File

@@ -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.
* <p>
* 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.
* <p>
* 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<SocketType> VALID_SOCKET_TYPES =
Arrays.asList(SocketType.PAIR, SocketType.PUSH, SocketType.PUB);
private final Scheduler publisherScheduler = Schedulers.newSingle("zeroMqMessageHandlerScheduler");
private final Mono<ZMQ.Socket> socketMono;
private OutboundMessageMapper<byte[]> messageMapper;
private Consumer<ZMQ.Socket> 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<byte[]> 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<ZMQ.Socket> 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<Void> 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();
}
}

4
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes for outbound channel adapters over ZeroMQ.
*/
package org.springframework.integration.zeromq.outbound;

10
spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java
View File

@@ -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<String> 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();
}

153
spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/dsl/ZeroMqDslTests.java Normal file
View File

@@ -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<Message<?>> 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"));
}
}
}

143
spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java Normal file
View File

@@ -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();
}
}

138
spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java Normal file
View File

@@ -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) -> 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();
}
}

2
src/reference/asciidoc/whats-new.adoc
View File

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

111
src/reference/asciidoc/zeromq.adoc
View File

@@ -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<ZMQ.Socket> 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<byte[]>` 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<ZMQ.Socket> 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) -> 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")))
}
----
====