diff --git a/consumer/zeromq-consumer/README.adoc b/consumer/zeromq-consumer/README.adoc new file mode 100644 index 00000000..58976d0b --- /dev/null +++ b/consumer/zeromq-consumer/README.adoc @@ -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>, Mono> zeromqConsumer` + +You need to inject this as `Function>, Mono> 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. diff --git a/consumer/zeromq-consumer/pom.xml b/consumer/zeromq-consumer/pom.xml new file mode 100644 index 00000000..d9fc0f26 --- /dev/null +++ b/consumer/zeromq-consumer/pom.xml @@ -0,0 +1,59 @@ + + + 4.0.0 + + + org.springframework.cloud.fn + spring-functions-parent + ${revision} + ../../spring-functions-parent + + + zeromq-consumer + zeromq-consumer + ZeroMQ consumer + + + + + org.springframework.cloud.fn + config-common + ${project.version} + + + + + org.springframework.integration + spring-integration-zeromq + + + + + org.springframework.boot + spring-boot-starter-json + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + + + diff --git a/consumer/zeromq-consumer/src/main/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerConfiguration.java b/consumer/zeromq-consumer/src/main/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerConfiguration.java new file mode 100644 index 00000000..035ba3f7 --- /dev/null +++ b/consumer/zeromq-consumer/src/main/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerConfiguration.java @@ -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 socketConfigurer, + @Autowired(required = false) OutboundMessageMapper 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>, Mono> zeromqConsumer(ZeroMqMessageHandler zeromqMessageHandler) { + return input -> input.flatMap(zeromqMessageHandler::handleMessage) + .ignoreElements(); + } + +} diff --git a/consumer/zeromq-consumer/src/main/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerProperties.java b/consumer/zeromq-consumer/src/main/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerProperties.java new file mode 100644 index 00000000..ac5eda08 --- /dev/null +++ b/consumer/zeromq-consumer/src/main/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerProperties.java @@ -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; + } + +} diff --git a/consumer/zeromq-consumer/src/test/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerConfigurationTests.java b/consumer/zeromq-consumer/src/test/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerConfigurationTests.java new file mode 100644 index 00000000..e3b00b35 --- /dev/null +++ b/consumer/zeromq-consumer/src/test/java/org/springframework/cloud/fn/consumer/zeromq/ZeroMqConsumerConfigurationTests.java @@ -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>, Mono> 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 { + } + +} diff --git a/function-dependencies/pom.xml b/function-dependencies/pom.xml index 50653c2c..6c4ca103 100644 --- a/function-dependencies/pom.xml +++ b/function-dependencies/pom.xml @@ -106,6 +106,11 @@ websocket-supplier ${revision} + + org.springframework.cloud.fn + zeromq-supplier + ${revision} + org.springframework.cloud.fn analytics-consumer @@ -201,6 +206,11 @@ websocket-consumer ${revision} + + org.springframework.cloud.fn + zeromq-consumer + ${revision} + org.springframework.cloud.fn aggregator-function diff --git a/pom.xml b/pom.xml index 1ebaf85c..af72c93d 100644 --- a/pom.xml +++ b/pom.xml @@ -55,6 +55,7 @@ consumer/twitter-consumer consumer/wavefront-consumer consumer/rsocket-consumer + consumer/zeromq-consumer function/aggregator-function function/filter-function @@ -88,6 +89,7 @@ supplier/twitter-supplier supplier/cdc-debezium-supplier supplier/syslog-supplier + supplier/zeromq-supplier spring-functions-parent function-dependencies diff --git a/supplier/zeromq-supplier/README.adoc b/supplier/zeromq-supplier/README.adoc new file mode 100644 index 00000000..bacc875a --- /dev/null +++ b/supplier/zeromq-supplier/README.adoc @@ -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>>`. +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>>`. + +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. \ No newline at end of file diff --git a/supplier/zeromq-supplier/pom.xml b/supplier/zeromq-supplier/pom.xml new file mode 100644 index 00000000..3555d4c4 --- /dev/null +++ b/supplier/zeromq-supplier/pom.xml @@ -0,0 +1,45 @@ + + + 4.0.0 + + + org.springframework.cloud.fn + spring-functions-parent + ${revision} + ../../spring-functions-parent + + + zeromq-supplier + zeromq-supplier + ZeroMQ supplier + + + + org.springframework.integration + spring-integration-zeromq + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-validation + + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + + + diff --git a/supplier/zeromq-supplier/src/main/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierConfiguration.java b/supplier/zeromq-supplier/src/main/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierConfiguration.java new file mode 100644 index 00000000..c55bc008 --- /dev/null +++ b/supplier/zeromq-supplier/src/main/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierConfiguration.java @@ -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 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>> zeromqSupplier(ZeroMqMessageProducer adapter) { + return () -> Flux.from(output).doOnSubscribe(subscription -> adapter.start()); + } + +} diff --git a/supplier/zeromq-supplier/src/main/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierProperties.java b/supplier/zeromq-supplier/src/main/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierProperties.java new file mode 100644 index 00000000..f4875807 --- /dev/null +++ b/supplier/zeromq-supplier/src/main/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierProperties.java @@ -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; + } + +} diff --git a/supplier/zeromq-supplier/src/test/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierConfigurationTests.java b/supplier/zeromq-supplier/src/test/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierConfigurationTests.java new file mode 100644 index 00000000..7239fab8 --- /dev/null +++ b/supplier/zeromq-supplier/src/test/java/org/springframework/cloud/fn/supplier/zeromq/ZeroMqSupplierConfigurationTests.java @@ -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>> 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 { } + +}