diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java index f5ae7370d7..0d35979a3f 100644 --- a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -168,7 +168,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl ? SocketType.PAIR : (this.pubSub ? SocketType.PUB : SocketType.PUSH)) )) - .doOnNext(this.sendSocketConfigurer) + .doOnNext((socket) -> this.sendSocketConfigurer.accept(socket)) .doOnNext((socket) -> socket.connect(this.connectSendUrl != null ? this.connectSendUrl @@ -184,7 +184,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl this.connectSubscribeUrl == null ? SocketType.PAIR : (this.pubSub ? SocketType.SUB : SocketType.PULL)))) - .doOnNext(this.subscribeSocketConfigurer) + .doOnNext((socket) -> this.subscribeSocketConfigurer.accept(socket)) .doOnNext((socket) -> { if (this.connectSubscribeUrl != null) { if (this.pubSub) { @@ -213,7 +213,7 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl return Mono.empty(); }) .publishOn(Schedulers.parallel()) - .map(this.messageMapper::toMessage) + .map((data) -> this.messageMapper.toMessage(data)) .doOnError((error) -> logger.error(error, () -> "Error processing ZeroMQ message in the " + this)) .repeatWhenEmpty((repeat) -> diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java index 7e55ac4a00..5a9e38389c 100644 --- a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -104,7 +104,7 @@ public class ZeroMqMessageHandler extends AbstractReactiveMessageHandler { this.socketMono = Mono.just(context.createSocket(socketType)) .publishOn(this.publisherScheduler) - .doOnNext(this.socketConfigurer) + .doOnNext((socket) -> this.socketConfigurer.accept(socket)) .doOnNext((socket) -> socket.connect(connectUrl)) .cache() .publishOn(this.publisherScheduler); diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java index 66a22ce0d3..612885195f 100644 --- a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/channel/ZeroMqChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -20,9 +20,11 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; import java.time.Duration; +import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; @@ -31,10 +33,13 @@ import org.zeromq.ZContext; import org.zeromq.ZMQ; import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper; +import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.zeromq.ZeroMqProxy; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; +import reactor.core.publisher.Mono; + /** * @author Artem Bilan * @@ -54,8 +59,29 @@ public class ZeroMqChannelTests { ZeroMqChannel channel = new ZeroMqChannel(CONTEXT); channel.setBeanName("testChannel1"); channel.setConsumeDelay(Duration.ofMillis(10)); + channel.setSendSocketConfigurer(socket -> socket.setZapDomain("global")); + channel.setSubscribeSocketConfigurer(socket -> socket.setZapDomain("local")); + AtomicBoolean customMessageMapperCalled = new AtomicBoolean(); + channel.setMessageMapper(new EmbeddedJsonHeadersMessageMapper() { + + @Override public Message toMessage(byte[] bytes, Map headers) { + customMessageMapperCalled.set(true); + return super.toMessage(bytes, headers); + } + + }); channel.afterPropertiesSet(); + @SuppressWarnings("unchecked") + Mono sendSocketMono = TestUtils.getPropertyValue(channel, "sendSocket", Mono.class); + ZMQ.Socket sendSocket = sendSocketMono.block(Duration.ofSeconds(10)); + assertThat(sendSocket.getZapDomain()).isEqualTo("global"); + + @SuppressWarnings("unchecked") + Mono subscribeSocketMono = TestUtils.getPropertyValue(channel, "subscribeSocket", Mono.class); + ZMQ.Socket subscribeSocket = subscribeSocketMono.block(Duration.ofSeconds(10)); + assertThat(subscribeSocket.getZapDomain()).isEqualTo("local"); + BlockingQueue> received = new LinkedBlockingQueue<>(); channel.subscribe(received::offer); @@ -78,6 +104,8 @@ public class ZeroMqChannelTests { assertThat(received.poll(100, TimeUnit.MILLISECONDS)).isNull(); channel.destroy(); + + assertThat(customMessageMapperCalled.get()).isTrue(); } @Test diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java index d367d85926..58f7385eee 100644 --- a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2021 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. @@ -33,8 +33,10 @@ import org.zeromq.ZMsg; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.support.GenericMessage; +import reactor.core.publisher.Mono; import reactor.test.StepVerifier; /** @@ -67,9 +69,15 @@ public class ZeroMqMessageProducerTests { messageProducer.setMessageMapper((object, headers) -> new GenericMessage<>(new String(object))); messageProducer.setConsumeDelay(Duration.ofMillis(10)); messageProducer.setBeanFactory(mock(BeanFactory.class)); + messageProducer.setSocketConfigurer(s -> s.setZapDomain("global")); messageProducer.afterPropertiesSet(); messageProducer.start(); + @SuppressWarnings("unchecked") + Mono socketMono = TestUtils.getPropertyValue(messageProducer, "socketMono", Mono.class); + ZMQ.Socket socketInUse = socketMono.block(Duration.ofSeconds(10)); + assertThat(socketInUse.getZapDomain()).isEqualTo("global"); + ZMQ.Socket socket = CONTEXT.createSocket(SocketType.PAIR); await().until(() -> messageProducer.getBoundPort() > 0); diff --git a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java index a77a98d509..8793c5b26c 100644 --- a/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java +++ b/spring-integration-zeromq/src/test/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandlerTests.java @@ -21,6 +21,8 @@ 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.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; import org.zeromq.SocketType; @@ -32,11 +34,14 @@ 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.test.util.TestUtils; import org.springframework.integration.zeromq.ZeroMqProxy; import org.springframework.messaging.Message; import org.springframework.messaging.converter.ByteArrayMessageConverter; import org.springframework.messaging.support.GenericMessage; +import reactor.core.publisher.Mono; + /** * @author Artem Bilan * @@ -59,8 +64,14 @@ public class ZeroMqMessageHandlerTests { ZeroMqMessageHandler messageHandler = new ZeroMqMessageHandler(CONTEXT, socketAddress); messageHandler.setBeanFactory(mock(BeanFactory.class)); + messageHandler.setSocketConfigurer(s -> s.setZapDomain("global")); messageHandler.afterPropertiesSet(); + @SuppressWarnings("unchecked") + Mono socketMono = TestUtils.getPropertyValue(messageHandler, "socketMono", Mono.class); + ZMQ.Socket socketInUse = socketMono.block(Duration.ofSeconds(10)); + assertThat(socketInUse.getZapDomain()).isEqualTo("global"); + Message testMessage = new GenericMessage<>("test"); messageHandler.handleMessage(testMessage).subscribe();