Migrating zeromq components from the 3.1 WIP branch to main
zeromq-supplier zeromq-consumer zeromq-source zeromq-sink
This commit is contained in:
Soby Chacko
30
consumer/zeromq-consumer/README.adoc
Normal file
30
consumer/zeromq-consumer/README.adoc
Normal file
@@ -0,0 +1,30 @@
|
||||
# ZeroMQ Consumer
|
||||
|
||||
A consumer that allows you to send messages through a ZeroMQ socker.
|
||||
|
||||
## Beans for injection
|
||||
|
||||
You can import the `ZeroMqConsumerConfiguration` in the application and then inject the following bean.
|
||||
|
||||
`Function<Flux<Message<?>>, Mono<Void>> zeromqConsumer`
|
||||
|
||||
You need to inject this as `Function<Flux<Message<?>>, Mono<Void>> zeromqConsumer`.
|
||||
|
||||
You can use `zeromqConsumer` as a qualifier when injecting.
|
||||
|
||||
**NOTE:** This is a functional endpoint. One will need to subscribe to this endpoint in order to start accepting data
|
||||
on it.
|
||||
|
||||
## Configuration Options
|
||||
|
||||
All configuration properties are prefixed with `zeromq.consumer`.
|
||||
|
||||
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerProperties.java[ZeroMqConsumerProperties].
|
||||
|
||||
## Tests
|
||||
|
||||
See this link:src/test/java/org/springframework/cloud/fn/consumer/zeromq/[test suite] for the various ways, this consumer is used.
|
||||
|
||||
## Other usage
|
||||
|
||||
See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/zeromq-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes a ZeroMQ Sink.
|
||||
59
consumer/zeromq-consumer/pom.xml
Normal file
59
consumer/zeromq-consumer/pom.xml
Normal file
@@ -0,0 +1,59 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>spring-functions-parent</artifactId>
|
||||
<version>${revision}</version>
|
||||
<relativePath>../../spring-functions-parent</relativePath>
|
||||
</parent>
|
||||
|
||||
<artifactId>zeromq-consumer</artifactId>
|
||||
<name>zeromq-consumer</name>
|
||||
<description>ZeroMQ consumer</description>
|
||||
|
||||
<dependencies>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>config-common</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- Spring Integration dependencies -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-zeromq</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- Spring Boot dependencies -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-json</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-validation</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
|
||||
<!-- Test dependencies -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.projectreactor</groupId>
|
||||
<artifactId>reactor-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,77 @@
|
||||
/*
|
||||
* Copyright 2016-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.cloud.fn.consumer.zeromq;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
|
||||
import org.zeromq.ZContext;
|
||||
import org.zeromq.ZMQ;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.mapping.OutboundMessageMapper;
|
||||
import org.springframework.integration.zeromq.outbound.ZeroMqMessageHandler;
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Daniel Frey
|
||||
* @since 3.1.0
|
||||
*/
|
||||
@Configuration
|
||||
@EnableConfigurationProperties(ZeroMqConsumerProperties.class)
|
||||
public class ZeroMqConsumerConfiguration {
|
||||
|
||||
@Bean
|
||||
public ZContext zContext() {
|
||||
return new ZContext();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ZeroMqMessageHandler zeromqMessageHandler(ZeroMqConsumerProperties properties, ZContext zContext,
|
||||
@Autowired(required = false) Consumer<ZMQ.Socket> socketConfigurer,
|
||||
@Autowired(required = false) OutboundMessageMapper<byte[]> messageMapper) {
|
||||
ZeroMqMessageHandler zeroMqMessageHandler = new ZeroMqMessageHandler(zContext, properties.getConnectUrl(),
|
||||
properties.getSocketType());
|
||||
|
||||
if (properties.getTopic() != null) {
|
||||
zeroMqMessageHandler.setTopicExpression(properties.getTopic());
|
||||
}
|
||||
|
||||
if (socketConfigurer != null) {
|
||||
zeroMqMessageHandler.setSocketConfigurer(socketConfigurer);
|
||||
}
|
||||
|
||||
if (messageMapper != null) {
|
||||
zeroMqMessageHandler.setMessageMapper(messageMapper);
|
||||
}
|
||||
|
||||
return zeroMqMessageHandler;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Flux<Message<?>>, Mono<Void>> zeromqConsumer(ZeroMqMessageHandler zeromqMessageHandler) {
|
||||
return input -> input.flatMap(zeromqMessageHandler::handleMessage)
|
||||
.ignoreElements();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
/*
|
||||
* Copyright 2016-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.cloud.fn.consumer.zeromq;
|
||||
|
||||
import javax.validation.constraints.NotEmpty;
|
||||
import javax.validation.constraints.NotNull;
|
||||
|
||||
import org.zeromq.SocketType;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.expression.Expression;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Daniel Frey
|
||||
* @since 3.1.0
|
||||
*/
|
||||
@ConfigurationProperties("zeromq.consumer")
|
||||
@Validated
|
||||
public class ZeroMqConsumerProperties {
|
||||
|
||||
/**
|
||||
* The Socket Type the connection should establish.
|
||||
*/
|
||||
private SocketType socketType = SocketType.PUB;
|
||||
|
||||
/**
|
||||
* Connection URL for connecting to the ZeroMQ Socket.
|
||||
*/
|
||||
private String connectUrl;
|
||||
|
||||
/**
|
||||
* A Topic SpEL expression to evaluate a topic before sending messages to subscribers.
|
||||
*/
|
||||
private Expression topic;
|
||||
|
||||
@NotNull(message = "'socketType' is required")
|
||||
public SocketType getSocketType() {
|
||||
return socketType;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param socketType the {@link SocketType} to establish.
|
||||
*/
|
||||
public void setSocketType(SocketType socketType) {
|
||||
this.socketType = socketType;
|
||||
}
|
||||
|
||||
@NotEmpty(message = "connectUrl is required like protocol://server:port")
|
||||
public String getConnectUrl() {
|
||||
return connectUrl;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param connectUrl The ZeroMQ socket to expose
|
||||
*/
|
||||
public void setConnectUrl(String connectUrl) {
|
||||
this.connectUrl = connectUrl;
|
||||
}
|
||||
|
||||
public Expression getTopic() {
|
||||
return topic;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param topic The 'topic' SpEL expression to set
|
||||
*/
|
||||
public void setTopic(Expression topic) {
|
||||
this.topic = topic;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
/*
|
||||
* Copyright 2016-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.cloud.fn.consumer.zeromq;
|
||||
|
||||
import java.util.function.Function;
|
||||
|
||||
import org.junit.jupiter.api.AfterAll;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.zeromq.SocketType;
|
||||
import org.zeromq.ZContext;
|
||||
import org.zeromq.ZMQ;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Daniel Frey since 3.1.0
|
||||
*/
|
||||
@SpringBootTest(properties = {
|
||||
"zeromq.consumer.topic='test-topic'"
|
||||
})
|
||||
@DirtiesContext
|
||||
public class ZeroMqConsumerConfigurationTests {
|
||||
|
||||
private static final ZContext CONTEXT = new ZContext();
|
||||
|
||||
private static ZMQ.Socket socket;
|
||||
|
||||
@Autowired
|
||||
Function<Flux<Message<?>>, Mono<Void>> subject;
|
||||
|
||||
@BeforeAll
|
||||
static void setup() {
|
||||
|
||||
socket = CONTEXT.createSocket(SocketType.SUB);
|
||||
socket.setReceiveTimeOut(10_000);
|
||||
int bindPort = socket.bindToRandomPort("tcp://*");
|
||||
socket.subscribe("test-topic");
|
||||
|
||||
System.setProperty("zeromq.consumer.connectUrl", "tcp://localhost:" + bindPort);
|
||||
|
||||
}
|
||||
|
||||
@AfterAll
|
||||
static void tearDown() {
|
||||
socket.close();
|
||||
CONTEXT.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testMessageHandlerConfiguration() throws InterruptedException {
|
||||
|
||||
Thread.sleep(2000);
|
||||
|
||||
Message<?> testMessage = MessageBuilder.withPayload("test").setHeader("topic", "test-topic").build();
|
||||
subject.apply(Flux.just(testMessage))
|
||||
.subscribe();
|
||||
|
||||
String topic = socket.recvStr();
|
||||
assertThat(topic).isEqualTo("test-topic");
|
||||
assertThat(socket.recvStr()).isEmpty();
|
||||
assertThat(socket.recvStr()).isEqualTo("test");
|
||||
|
||||
}
|
||||
|
||||
@SpringBootApplication
|
||||
public static class ZeroMqConsumerTestApplication {
|
||||
}
|
||||
|
||||
}
|
||||
@@ -106,6 +106,11 @@
|
||||
<artifactId>websocket-supplier</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>zeromq-supplier</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>analytics-consumer</artifactId>
|
||||
@@ -201,6 +206,11 @@
|
||||
<artifactId>websocket-consumer</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>zeromq-consumer</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>aggregator-function</artifactId>
|
||||
|
||||
2
pom.xml
2
pom.xml
@@ -55,6 +55,7 @@
|
||||
<module>consumer/twitter-consumer</module>
|
||||
<module>consumer/wavefront-consumer</module>
|
||||
<module>consumer/rsocket-consumer</module>
|
||||
<module>consumer/zeromq-consumer</module>
|
||||
|
||||
<module>function/aggregator-function</module>
|
||||
<module>function/filter-function</module>
|
||||
@@ -88,6 +89,7 @@
|
||||
<module>supplier/twitter-supplier</module>
|
||||
<module>supplier/cdc-debezium-supplier</module>
|
||||
<module>supplier/syslog-supplier</module>
|
||||
<module>supplier/zeromq-supplier</module>
|
||||
|
||||
<module>spring-functions-parent</module>
|
||||
<module>function-dependencies</module>
|
||||
|
||||
32
supplier/zeromq-supplier/README.adoc
Normal file
32
supplier/zeromq-supplier/README.adoc
Normal file
@@ -0,0 +1,32 @@
|
||||
# ZeroMQ Supplier
|
||||
|
||||
A basic ZeroMQ supplier that produced messages through TCP connection.
|
||||
The `Supplier` uses the `ZeroMqMessageProducer` from Spring Integration.
|
||||
This supplier gives you a reactive stream of messages and the supplier has a signature of `Supplier<Flux<Message<?>>>`.
|
||||
Users have to subscribe to this `Flux` and receive the data.
|
||||
|
||||
## Beans for injection
|
||||
|
||||
You can import the `ZeroMqSupplierConfiguration` in the application and then inject the following bean.
|
||||
|
||||
`zeromqSupplier`
|
||||
|
||||
You need to inject this as `Supplier<Flux<Message<?>>>`.
|
||||
|
||||
You can use `zeromqSupplier` as a qualifier when injecting.
|
||||
|
||||
Once injected, you can use the `get` method of the `Supplier` to invoke it and then subscribe to the returned `Flux`.
|
||||
|
||||
## Configuration Options
|
||||
|
||||
All configuration properties are prefixed with `zeromq.supplier`.
|
||||
|
||||
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierProperties.java[ZeroMqSupplierProperties].
|
||||
|
||||
## Tests
|
||||
|
||||
See this link:src/test/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierTests.java[test suite] for the various ways, this supplier is used.
|
||||
|
||||
## Other usage
|
||||
|
||||
See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/zeromq-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application where it makes a ZeroMQ Source.
|
||||
45
supplier/zeromq-supplier/pom.xml
Normal file
45
supplier/zeromq-supplier/pom.xml
Normal file
@@ -0,0 +1,45 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>spring-functions-parent</artifactId>
|
||||
<version>${revision}</version>
|
||||
<relativePath>../../spring-functions-parent</relativePath>
|
||||
</parent>
|
||||
|
||||
<artifactId>zeromq-supplier</artifactId>
|
||||
<name>zeromq-supplier</name>
|
||||
<description>ZeroMQ supplier</description>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-zeromq</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-validation</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.projectreactor</groupId>
|
||||
<artifactId>reactor-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,84 @@
|
||||
/*
|
||||
* Copyright 2016-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.cloud.fn.supplier.zeromq;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.zeromq.SocketType;
|
||||
import org.zeromq.ZContext;
|
||||
import org.zeromq.ZMQ;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.channel.FluxMessageChannel;
|
||||
import org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
/**
|
||||
* A source module that receives data from ZeroMQ.
|
||||
*
|
||||
* @author Daniel Frey
|
||||
* @since 3.1.0
|
||||
*/
|
||||
@Configuration
|
||||
@EnableConfigurationProperties(ZeroMqSupplierProperties.class)
|
||||
public class ZeroMqSupplierConfiguration {
|
||||
|
||||
private FluxMessageChannel output = new FluxMessageChannel();
|
||||
|
||||
@Bean
|
||||
public ZContext zContext() {
|
||||
return new ZContext();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ZeroMqMessageProducer adapter(ZeroMqSupplierProperties properties, ZContext zContext,
|
||||
@Autowired(required = false) Consumer<ZMQ.Socket> socketConfigurer) {
|
||||
|
||||
ZeroMqMessageProducer zeroMqMessageProducer = new ZeroMqMessageProducer(zContext, properties.getSocketType());
|
||||
|
||||
if (properties.getConnectUrl() != null) {
|
||||
zeroMqMessageProducer.setConnectUrl(properties.getConnectUrl());
|
||||
}
|
||||
else if (properties.getBindPort() > 0) {
|
||||
zeroMqMessageProducer.setBindPort(properties.getBindPort());
|
||||
}
|
||||
zeroMqMessageProducer.setConsumeDelay(properties.getConsumeDelay());
|
||||
if (SocketType.SUB.equals(properties.getSocketType())) {
|
||||
zeroMqMessageProducer.setTopics(properties.getTopics());
|
||||
}
|
||||
zeroMqMessageProducer.setMessageMapper(GenericMessage::new);
|
||||
if (socketConfigurer != null) {
|
||||
zeroMqMessageProducer.setSocketConfigurer(socketConfigurer);
|
||||
}
|
||||
zeroMqMessageProducer.setOutputChannel(output);
|
||||
zeroMqMessageProducer.setAutoStartup(false);
|
||||
|
||||
return zeroMqMessageProducer;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Supplier<Flux<Message<?>>> zeromqSupplier(ZeroMqMessageProducer adapter) {
|
||||
return () -> Flux.from(output).doOnSubscribe(subscription -> adapter.start());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
/*
|
||||
* Copyright 2016-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.cloud.fn.supplier.zeromq;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
import javax.validation.constraints.NotEmpty;
|
||||
import javax.validation.constraints.NotNull;
|
||||
|
||||
import org.hibernate.validator.constraints.Range;
|
||||
import org.zeromq.SocketType;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Daniel Frey
|
||||
* @since 3.1.0
|
||||
*/
|
||||
@ConfigurationProperties("zeromq.supplier")
|
||||
@Validated
|
||||
public class ZeroMqSupplierProperties {
|
||||
|
||||
/**
|
||||
* The Socket Type the connection should make.
|
||||
*/
|
||||
private SocketType socketType = SocketType.SUB;
|
||||
|
||||
/**
|
||||
* Connection URL for to the ZeroMQ Socket.
|
||||
*/
|
||||
private String connectUrl;
|
||||
|
||||
/**
|
||||
* Bind Port for creating a ZeroMQ Socket; 0 selects a random port.
|
||||
*/
|
||||
private int bindPort;
|
||||
|
||||
/**
|
||||
* The delay to consume from the ZeroMQ Socket when no data received.
|
||||
*/
|
||||
private Duration consumeDelay = Duration.ofSeconds(1);
|
||||
|
||||
/**
|
||||
* The Topics to subscribe to.
|
||||
*/
|
||||
private String[] topics = {""};
|
||||
|
||||
/**
|
||||
* @param socketType the {@link SocketType} to establish.
|
||||
*/
|
||||
public void setSocketType(SocketType socketType) {
|
||||
this.socketType = socketType;
|
||||
}
|
||||
|
||||
@NotNull(message = "'socketType' is required")
|
||||
public SocketType getSocketType() {
|
||||
return socketType;
|
||||
}
|
||||
|
||||
@NotEmpty(message = "connectUrl is required like tcp://server:port")
|
||||
public String getConnectUrl() {
|
||||
return connectUrl;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param connectUrl The ZeroMQ server connect url
|
||||
*
|
||||
* @see org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer#setConnectUrl(String)
|
||||
*/
|
||||
public void setConnectUrl(String connectUrl) {
|
||||
this.connectUrl = connectUrl;
|
||||
}
|
||||
|
||||
@Range(min = 0, message = "'bindPort' must not be negative")
|
||||
public int getBindPort() {
|
||||
return bindPort;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param bindPort The Port to bind to on all interfaces
|
||||
*
|
||||
* @see org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer#setBindPort(int)
|
||||
*/
|
||||
public void setBindPort(int bindPort) {
|
||||
this.bindPort = bindPort;
|
||||
}
|
||||
|
||||
@NotNull(message = "'consumeDelay' is required")
|
||||
public Duration getConsumeDelay() {
|
||||
return consumeDelay;
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a {@link Duration} to delay consumption when no data received.
|
||||
* @param consumeDelay the {@link Duration} to delay consumption when empty.
|
||||
*/
|
||||
public void setConsumeDelay(Duration consumeDelay) {
|
||||
this.consumeDelay = consumeDelay;
|
||||
}
|
||||
|
||||
public String[] getTopics() {
|
||||
return topics;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param topics The ZeroMQ Topics to subscribe to
|
||||
*
|
||||
* @see org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer#setTopics(String...)
|
||||
*/
|
||||
public void setTopics(String... topics) {
|
||||
this.topics = topics;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
/*
|
||||
* Copyright 2016-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.cloud.fn.supplier.zeromq;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.assertj.core.api.InstanceOfAssertFactories;
|
||||
import org.junit.jupiter.api.AfterAll;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
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 reactor.core.publisher.Flux;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Daniel Frey
|
||||
* since 3.1.0
|
||||
*/
|
||||
@SpringBootTest(properties = {"zeromq.supplier.topics=test-topic"})
|
||||
@DirtiesContext
|
||||
public class ZeroMqSupplierConfigurationTests {
|
||||
|
||||
private static final ZContext CONTEXT = new ZContext();
|
||||
private static ZMQ.Socket socket;
|
||||
|
||||
@Autowired
|
||||
Supplier<Flux<Message<?>>> subject;
|
||||
|
||||
@BeforeAll
|
||||
static void setup() {
|
||||
|
||||
String socketAddress = "tcp://*";
|
||||
socket = CONTEXT.createSocket(SocketType.PUB);
|
||||
int bindPort = socket.bindToRandomPort(socketAddress);
|
||||
|
||||
System.setProperty("zeromq.supplier.connectUrl", "tcp://localhost:" + bindPort);
|
||||
|
||||
}
|
||||
|
||||
@AfterAll
|
||||
static void tearDown() {
|
||||
socket.close();
|
||||
CONTEXT.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testSubscriptionConfiguration() throws InterruptedException {
|
||||
|
||||
StepVerifier stepVerifier =
|
||||
StepVerifier.create(subject.get())
|
||||
.assertNext((message) ->
|
||||
assertThat(message.getPayload())
|
||||
.asInstanceOf(InstanceOfAssertFactories.type(byte[].class))
|
||||
.isEqualTo("test".getBytes(ZMQ.CHARSET))
|
||||
)
|
||||
.thenCancel()
|
||||
.verifyLater();
|
||||
|
||||
Thread.sleep(2000);
|
||||
|
||||
ZMsg msg = ZMsg.newStringMsg("test");
|
||||
msg.wrap(new ZFrame("test-topic"));
|
||||
msg.send(socket);
|
||||
|
||||
stepVerifier.verify(Duration.ofSeconds(10));
|
||||
|
||||
}
|
||||
|
||||
@SpringBootApplication
|
||||
public static class ZeroMqSourceTestApplication { }
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user