From 65ec234697e3577b5b19e6d2aca6fc04db7d501c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sat, 5 Nov 2016 14:28:05 -0400 Subject: [PATCH] INT-4162: TCP/UDP DSL JIRA: https://jira.spring.io/browse/INT-4162 INT-4162: IP DSL - Phase I Connection Factories Phase 2 - Adapters/Gateways Phase 3 - UDP Polishing - PR Comments Remove unnecessary generics. More Polishing - PR Comments Checkstyle/Javadoc Fixes ComponentsRegistration Improvements Add ...Spec CTORs to avoid the issue described here: https://github.com/spring-projects/spring-integration-java-dsl/issues/137 Also other PR comments. Fix UDP Spec Inheritance Jdbc Lock Test Log Adjuster Fix Mongo Test Race Conditions * Fox typos in the `AbstractUdpOutboundChannelAdapterSpec` * Improve `IpIntegrationTests.testUdpInheritance()` * Change `fixed-rate` to `fixed-delay` in the `MongoDbInboundChannelAdapterIntegrationTests-context` to pursue single-threaded environment from poller for test --- .../TcpConnectionFactoryFactoryBean.java | 33 +-- .../ip/dsl/AbstractConnectionFactorySpec.java | 222 ++++++++++++++++++ ...AbstractUdpOutboundChannelAdapterSpec.java | 131 +++++++++++ .../integration/ip/dsl/Tcp.java | 162 +++++++++++++ .../dsl/TcpClientConnectionFactorySpec.java | 41 ++++ .../ip/dsl/TcpInboundChannelAdapterSpec.java | 97 ++++++++ .../ip/dsl/TcpInboundGatewaySpec.java | 96 ++++++++ .../ip/dsl/TcpOutboundChannelAdapterSpec.java | 97 ++++++++ .../ip/dsl/TcpOutboundGatewaySpec.java | 96 ++++++++ .../dsl/TcpServerConnectionFactorySpec.java | 61 +++++ .../integration/ip/dsl/Udp.java | 97 ++++++++ .../ip/dsl/UdpInboundChannelAdapterSpec.java | 155 ++++++++++++ ...dpMulticastOutboundChannelAdapterSpec.java | 54 +++++ .../UdpUnicastOutboundChannelAdapterSpec.java | 43 ++++ .../integration/ip/dsl/package-info.java | 4 + .../ip/tcp/TcpSendingMessageHandler.java | 25 +- .../connection/AbstractConnectionFactory.java | 52 ++-- .../CachingClientConnectionFactory.java | 15 -- .../ip/tcp/connection/ConnectionFactory.java | 4 +- .../udp/MulticastSendingMessageHandler.java | 25 ++ .../udp/UnicastReceivingChannelAdapter.java | 9 + .../ip/udp/UnicastSendingMessageHandler.java | 25 +- .../integration/ip/util/TestingUtilities.java | 32 +++ .../ip/dsl/ConnectionFacforyTests.java | 69 ++++++ .../ip/dsl/IpIntegrationTests.java | 186 +++++++++++++++ .../integration/ip/tcp/AutoStartTests.java | 3 +- .../JdbcLockRegistryLeaderInitiatorTests.java | 10 +- .../jms/dsl/JmsOutboundGatewaySpec.java | 2 +- ...ChannelAdapterIntegrationTests-context.xml | 14 +- ...InboundChannelAdapterIntegrationTests.java | 11 +- 30 files changed, 1758 insertions(+), 113 deletions(-) create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractConnectionFactorySpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpClientConnectionFactorySpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundChannelAdapterSpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundGatewaySpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundChannelAdapterSpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundGatewaySpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpServerConnectionFactorySpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Udp.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpInboundChannelAdapterSpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpMulticastOutboundChannelAdapterSpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpUnicastOutboundChannelAdapterSpec.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/package-info.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/ConnectionFacforyTests.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java index c5c251eae5..f8bec67fda 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java @@ -24,7 +24,7 @@ import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.config.AbstractFactoryBean; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; -import org.springframework.context.SmartLifecycle; +import org.springframework.context.Lifecycle; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; @@ -55,8 +55,8 @@ import org.springframework.util.Assert; * @author Gary Russell * @since 2.0.5 */ -public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean implements SmartLifecycle, BeanNameAware, - BeanFactoryAware, ApplicationEventPublisherAware { +public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean + implements Lifecycle, BeanNameAware, BeanFactoryAware, ApplicationEventPublisherAware { private volatile AbstractConnectionFactory connectionFactory; @@ -446,33 +446,6 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean the target {@link AbstractConnectionFactorySpec} implementation type. + * @param the target {@link AbstractConnectionFactory} implementation type. + * + * @author Gary Russell + * @since 5.0 + * + */ +public abstract class AbstractConnectionFactorySpec + , C extends AbstractConnectionFactory> + extends IntegrationComponentSpec { + + AbstractConnectionFactorySpec(C connectionFactory) { + this.target = connectionFactory; + } + + @Override + public S id(String id) { + this.target.setBeanName(id); + return _this(); + } + + /** + * @param soTimeout the timeout socket option. + * @return the spec. + * @see AbstractConnectionFactory#setSoTimeout(int) + */ + public S soTimeout(int soTimeout) { + this.target.setSoTimeout(soTimeout); + return _this(); + } + + /** + * @param soReceiveBufferSize the receive buffer size socket option. + * @return the spec. + * @see AbstractConnectionFactory#setSoReceiveBufferSize(int) + */ + public S soReceiveBufferSize(int soReceiveBufferSize) { + this.target.setSoReceiveBufferSize(soReceiveBufferSize); + return _this(); + } + + /** + * @param soSendBufferSize the send buffer size socket option. + * @return the spec. + * @see AbstractConnectionFactory#setSoSendBufferSize(int) + */ + public S soSendBufferSize(int soSendBufferSize) { + this.target.setSoSendBufferSize(soSendBufferSize); + return _this(); + } + + /** + * @param soTcpNoDelay the TCP no delay socket option (disable Nagle's algorithm). + * @return the spec. + * @see AbstractConnectionFactory#setSoTcpNoDelay(boolean) + */ + public S soTcpNoDelay(boolean soTcpNoDelay) { + this.target.setSoTcpNoDelay(soTcpNoDelay); + return _this(); + } + + /** + * @param soLinger the linger socket option. + * @return the spec. + * @see AbstractConnectionFactory#setSoLinger(int) + */ + public S soLinger(int soLinger) { + this.target.setSoLinger(soLinger); + return _this(); + } + + /** + * @param soKeepAlive the keep alive socket option. + * @return the spec. + * @see AbstractConnectionFactory#setSoKeepAlive(boolean) + */ + public S soKeepAlive(boolean soKeepAlive) { + this.target.setSoKeepAlive(soKeepAlive); + return _this(); + } + + /** + * @param soTrafficClass the traffic class socket option. + * @return the spec. + * @see AbstractConnectionFactory#setSoTrafficClass(int) + */ + public S soTrafficClass(int soTrafficClass) { + this.target.setSoTrafficClass(soTrafficClass); + return _this(); + } + + /** + * @param taskExecutor the task executor. + * @return the spec. + * @see AbstractConnectionFactory#setTaskExecutor(Executor) + */ + public S taskExecutor(Executor taskExecutor) { + this.target.setTaskExecutor(taskExecutor); + return _this(); + } + + /** + * @param deserializer the deserializer. + * @return the spec. + * @see AbstractConnectionFactory#setDeserializer(Deserializer) + */ + public S deserializer(Deserializer deserializer) { + this.target.setDeserializer(deserializer); + return _this(); + } + + /** + * @param serializer the serializer. + * @return the spec. + * @see AbstractConnectionFactory#setSerializer(Serializer) + */ + public S serializer(Serializer serializer) { + this.target.setSerializer(serializer); + return _this(); + } + + /** + * @param mapper the message mapper. + * @return the spec. + * @see AbstractConnectionFactory#setMapper(TcpMessageMapper) + */ + public S mapper(TcpMessageMapper mapper) { + this.target.setMapper(mapper); + return _this(); + } + + /** + * @param leaveOpen true to leave the socket open for additional messages. + * @return the spec. + * @see AbstractConnectionFactory#setLeaveOpen(boolean) + */ + public S leaveOpen(boolean leaveOpen) { + this.target.setLeaveOpen(leaveOpen); + return _this(); + } + + /** + * @param interceptorFactoryChain the interceptor factory chain. + * @return the spec. + * @see AbstractConnectionFactory#setInterceptorFactoryChain(TcpConnectionInterceptorFactoryChain) + */ + public S interceptorFactoryChain(TcpConnectionInterceptorFactoryChain interceptorFactoryChain) { + this.target.setInterceptorFactoryChain(interceptorFactoryChain); + return _this(); + } + + /** + * @param lookupHost true to reverse lookup the host. + * @return the spec. + * @see AbstractConnectionFactory#setLookupHost(boolean) + */ + public S lookupHost(boolean lookupHost) { + this.target.setLookupHost(lookupHost); + return _this(); + } + + /** + * @param nioHarvestInterval the harvest interval when using NIO. + * @return the spec. + * @see AbstractConnectionFactory#setNioHarvestInterval(int) + */ + public S nioHarvestInterval(int nioHarvestInterval) { + this.target.setNioHarvestInterval(nioHarvestInterval); + return _this(); + } + + /** + * @param readDelay the read delay. + * @return the spec. + * @see AbstractConnectionFactory#setReadDelay(long) + */ + public S readDelay(long readDelay) { + this.target.setReadDelay(readDelay); + return _this(); + } + + /** + * @param tcpSocketSupport the {@link TcpSocketSupport}. + * @return the spec. + * @see AbstractConnectionFactory#setTcpSocketSupport(TcpSocketSupport) + */ + public S tcpSocketSupport(TcpSocketSupport tcpSocketSupport) { + this.target.setTcpSocketSupport(tcpSocketSupport); + return _this(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java new file mode 100644 index 0000000000..a2ee0b6ac8 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java @@ -0,0 +1,131 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import java.net.DatagramSocket; +import java.util.function.Function; + +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.ip.udp.UnicastSendingMessageHandler; +import org.springframework.messaging.Message; + +/** + * A {@link MessageHandlerSpec} for UDP {@link org.springframework.messaging.MessageHandler}s. + * + * @param the target {@link AbstractUdpOutboundChannelAdapterSpec} implementation type. + * + * @author Gary Russell + * @since 5.0 + * + */ +public abstract class AbstractUdpOutboundChannelAdapterSpec> + extends MessageHandlerSpec { + + protected AbstractUdpOutboundChannelAdapterSpec() { + super(); + } + + AbstractUdpOutboundChannelAdapterSpec(String destinationExpression) { + this.target = new UnicastSendingMessageHandler(destinationExpression); + } + + AbstractUdpOutboundChannelAdapterSpec(Function, ?> destinationFunction) { + this.target = new UnicastSendingMessageHandler(new FunctionExpression<>(destinationFunction)); + } + + /** + * @param timeout the timeout socket option. + * @return the spec. + * @see UnicastSendingMessageHandler#setSoTimeout(int) + */ + public S soTimeout(int timeout) { + this.target.setSoTimeout(timeout); + return _this(); + } + + /** + * @param size the send buffer size socket option. + * @return the spec. + * @see UnicastSendingMessageHandler#setSoSendBufferSize(int) + */ + public S soSendBufferSize(int size) { + this.target.setSoSendBufferSize(size); + return _this(); + } + + /** + * @param localAddress the local address. + * @return the spec. + * @see UnicastSendingMessageHandler#setLocalAddress(String) + */ + public S localAddress(String localAddress) { + this.target.setLocalAddress(localAddress); + return _this(); + } + + /** + * @param lengthCheck the length check boolean. + * @return the spec. + * @see UnicastSendingMessageHandler#setLengthCheck(boolean) + */ + public S lengthCheck(boolean lengthCheck) { + this.target.setLengthCheck(lengthCheck); + return _this(); + } + + /** + * @param size the receive buffer size socket option. + * @return the spec. + * @see UnicastSendingMessageHandler#setSoReceiveBufferSize(int) + */ + public S soReceiveBufferSize(int size) { + this.target.setSoReceiveBufferSize(size); + return _this(); + } + + /** + * @param ackCounter the ack counter. + * @return the spec. + * @see UnicastSendingMessageHandler#setAckCounter(int) + */ + public S ackCounter(int ackCounter) { + this.target.setAckCounter(ackCounter); + return _this(); + } + + /** + * @param socketFunction the socket function. + * @return the spec. + * @see UnicastSendingMessageHandler#setSocketExpression(org.springframework.expression.Expression) + */ + public S socketFunction(Function, DatagramSocket> socketFunction) { + this.target.setSocketExpression(new FunctionExpression<>(socketFunction)); + return _this(); + } + + /** + * @param socketExpression the socket expression. + * @return the spec. + * @see UnicastSendingMessageHandler#setSocketExpressionString(String) + */ + public S socketExpression(String socketExpression) { + this.target.setSocketExpressionString(socketExpression); + return _this(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java new file mode 100644 index 0000000000..e28de38537 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java @@ -0,0 +1,162 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; + +/** + * Factory methods for TCP. + * + * @author Gary Russell + * @since 5.0 + * + */ +public final class Tcp { + + /** + * Boolean indicating the connection factory should use NIO. + */ + public static final boolean NIO = true; + + /** + * Boolean indicating the connection factory should not use NIO + * (default). + */ + public static final boolean NET = true; + + private Tcp() { + super(); + } + + /** + * Create a server spec that uses NIO. + * @param port the port to listen on. + * @return the spec. + */ + public static TcpServerConnectionFactorySpec nioServer(int port) { + return new TcpServerConnectionFactorySpec(port, NIO); + } + + /** + * Create a server spec that does not use NIO. + * @param port the port to listen on. + * @return the spec. + */ + public static TcpServerConnectionFactorySpec netServer(int port) { + return new TcpServerConnectionFactorySpec(port, NET); + } + + /** + * Create a client spec that uses NIO. + * @param host the host to connect to. + * @param port the port to connect to. + * @return the spec. + */ + public static TcpClientConnectionFactorySpec nioClient(String host, int port) { + return new TcpClientConnectionFactorySpec(host, port, NIO); + } + + /** + * Create a client spec that does not use NIO. + * @param host the host to connect to. + * @param port the port to connect to. + * @return the spec. + */ + public static TcpClientConnectionFactorySpec netClient(String host, int port) { + return new TcpClientConnectionFactorySpec(host, port, NET); + } + + /** + * Create an inbound gateway using the supplied connection factory. + * @param connectionFactory the connection factory - must be an existing bean - it + * will not be initialized. + * @return the spec. + */ + public static TcpInboundGatewaySpec inboundGateway(AbstractConnectionFactory connectionFactory) { + return new TcpInboundGatewaySpec(connectionFactory); + } + + /** + * Create an inbound gateway using the supplied connection factory. + * @param connectionFactorySpec the connection factory spec. + * @return the spec. + */ + public static TcpInboundGatewaySpec inboundGateway(AbstractConnectionFactorySpec connectionFactorySpec) { + return new TcpInboundGatewaySpec(connectionFactorySpec); + } + + /** + * Create an inbound channel adapter using the supplied connection factory. + * @param connectionFactory the connection factory - must be an existing bean - it + * will not be initialized. + * @return the spec. + */ + public static TcpInboundChannelAdapterSpec inboundAdapter(AbstractConnectionFactory connectionFactory) { + return new TcpInboundChannelAdapterSpec(connectionFactory); + } + + /** + * Create an inbound channel adapter using the supplied connection factory. + * @param connectionFactorySpec the connection factory spec. + * @return the spec. + */ + public static TcpInboundChannelAdapterSpec inboundAdapter( + AbstractConnectionFactorySpec connectionFactorySpec) { + return new TcpInboundChannelAdapterSpec(connectionFactorySpec); + } + + /** + * Create an outbound gateway using the supplied client connection factory. + * @param connectionFactory the connection factory - must be an existing bean - it + * will not be initialized. + * @return the spec. + */ + public static TcpOutboundGatewaySpec outboundGateway(AbstractClientConnectionFactory connectionFactory) { + return new TcpOutboundGatewaySpec(connectionFactory); + } + + /** + * Create an outbound gateway using the supplied client connection factory. + * @param connectionFactory the connection factory spec. + * @return the spec. + */ + public static TcpOutboundGatewaySpec outboundGateway(TcpClientConnectionFactorySpec connectionFactory) { + return new TcpOutboundGatewaySpec(connectionFactory); + } + + /** + * Create an outbound gateway using the supplied connection factory. + * @param connectionFactory the connection factory - must be an existing bean - it + * will not be initialized. + * @return the spec. + */ + public static TcpOutboundChannelAdapterSpec outboundAdapter(AbstractConnectionFactory connectionFactory) { + return new TcpOutboundChannelAdapterSpec(connectionFactory); + } + + /** + * Create an outbound gateway using the supplied connection factory. + * @param connectionFactorySpec the connection factory. + * @return the spec. + */ + public static TcpOutboundChannelAdapterSpec outboundAdapter( + AbstractConnectionFactorySpec connectionFactorySpec) { + return new TcpOutboundChannelAdapterSpec(connectionFactorySpec); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpClientConnectionFactorySpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpClientConnectionFactorySpec.java new file mode 100644 index 0000000000..7b29328698 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpClientConnectionFactorySpec.java @@ -0,0 +1,41 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; + +/** + * An {@link AbstractConnectionFactorySpec} for {@link AbstractClientConnectionFactory}s. + * @author Gary Russell + * + * @since 5.0 + * + */ +public class TcpClientConnectionFactorySpec + extends AbstractConnectionFactorySpec { + + TcpClientConnectionFactorySpec(String host, int port) { + this(host, port, false); + } + + TcpClientConnectionFactorySpec(String host, int port, boolean nio) { + super(nio ? new TcpNioClientConnectionFactory(host, port) : new TcpNetClientConnectionFactory(host, port)); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundChannelAdapterSpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundChannelAdapterSpec.java new file mode 100644 index 0000000000..bb7d928350 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundChannelAdapterSpec.java @@ -0,0 +1,97 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import java.util.Collection; +import java.util.Collections; + +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.MessageProducerSpec; +import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.scheduling.TaskScheduler; + +/** + * A {@link MessageProducerSpec} for {@link TcpReceivingChannelAdapter}s. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class TcpInboundChannelAdapterSpec + extends MessageProducerSpec + implements ComponentsRegistration { + + private final AbstractConnectionFactory connectionFactory; + + /** + * Construct an instance using an existing spring-managed connection factory. + * @param connectionFactoryBean the spring-managed bean. + */ + TcpInboundChannelAdapterSpec(AbstractConnectionFactory connectionFactoryBean) { + super(new TcpReceivingChannelAdapter()); + this.connectionFactory = null; + this.target.setConnectionFactory(connectionFactoryBean); + } + + /** + * Construct an instance using the provided connection factory spec. + * @param connectionFactorySpec the spec. + */ + TcpInboundChannelAdapterSpec(AbstractConnectionFactorySpec connectionFactorySpec) { + super(new TcpReceivingChannelAdapter()); + this.connectionFactory = connectionFactorySpec.get(); + this.target.setConnectionFactory(this.connectionFactory); + } + + /** + * @param isClientMode true to connect in client mode + * @return the spec. + * @see TcpReceivingChannelAdapter#setClientMode(boolean) + */ + public TcpInboundChannelAdapterSpec clientMode(boolean isClientMode) { + this.target.setClientMode(isClientMode); + return _this(); + } + + /** + * @param retryInterval the client mode retry interval to set. + * @return the spec. + * @see TcpReceivingChannelAdapter#setRetryInterval(long) + */ + public TcpInboundChannelAdapterSpec retryInterval(long retryInterval) { + this.target.setRetryInterval(retryInterval); + return _this(); + } + + /** + * @param taskScheduler the scheduler for connecting in client mode. + * @return the spec. + * @see TcpReceivingChannelAdapter#setTaskScheduler(TaskScheduler) + */ + public TcpInboundChannelAdapterSpec taskScheduler(TaskScheduler taskScheduler) { + this.target.setTaskScheduler(taskScheduler); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + return this.connectionFactory == null ? Collections.emptyList() + : Collections.singletonList(this.connectionFactory); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundGatewaySpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundGatewaySpec.java new file mode 100644 index 0000000000..f632352c97 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundGatewaySpec.java @@ -0,0 +1,96 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import java.util.Collection; +import java.util.Collections; + +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.MessagingGatewaySpec; +import org.springframework.integration.ip.tcp.TcpInboundGateway; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.scheduling.TaskScheduler; + +/** + * A {@link MessagingGatewaySpec} for {@link TcpInboundGateway}s. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class TcpInboundGatewaySpec extends MessagingGatewaySpec + implements ComponentsRegistration { + + private final AbstractConnectionFactory connectionFactory; + + /** + * Construct an instance using an existing spring-managed connection factory. + * @param connectionFactoryBean the spring-managed bean. + */ + TcpInboundGatewaySpec(AbstractConnectionFactory connectionFactoryBean) { + super(new TcpInboundGateway()); + this.connectionFactory = null; + this.target.setConnectionFactory(connectionFactoryBean); + } + + /** + * Construct an instance using a connection factory spec. + * @param connectionFactorySpec the spec. + */ + TcpInboundGatewaySpec(AbstractConnectionFactorySpec connectionFactorySpec) { + super(new TcpInboundGateway()); + this.connectionFactory = connectionFactorySpec.get(); + this.target.setConnectionFactory(this.connectionFactory); + } + + /** + * @param isClientMode true to connect in client mode + * @return the spec. + * @see TcpInboundGateway#setClientMode(boolean) + */ + public TcpInboundGatewaySpec clientMode(boolean isClientMode) { + this.target.setClientMode(isClientMode); + return _this(); + } + + /** + * @param retryInterval the client mode retry interval to set. + * @return the spec. + * @see TcpInboundGateway#setRetryInterval(long) + */ + public TcpInboundGatewaySpec retryInterval(long retryInterval) { + this.target.setRetryInterval(retryInterval); + return _this(); + } + + /** + * @param taskScheduler the scheduler for connecting in client mode. + * @return the spec. + * @see TcpInboundGateway#setTaskScheduler(TaskScheduler) + */ + public TcpInboundGatewaySpec taskScheduler(TaskScheduler taskScheduler) { + this.target.setTaskScheduler(taskScheduler); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + return this.connectionFactory == null ? Collections.emptyList() + : Collections.singletonList(this.connectionFactory); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundChannelAdapterSpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundChannelAdapterSpec.java new file mode 100644 index 0000000000..17df57766a --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundChannelAdapterSpec.java @@ -0,0 +1,97 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import java.util.Collection; +import java.util.Collections; + +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.ip.tcp.TcpSendingMessageHandler; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.scheduling.TaskScheduler; + +/** + * A {@link MessageHandlerSpec} for {@link TcpSendingMessageHandler}s. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class TcpOutboundChannelAdapterSpec + extends MessageHandlerSpec + implements ComponentsRegistration { + + private final AbstractConnectionFactory connectionFactory; + + /** + * Construct an instance using an existing spring-managed connection factory. + * @param connectionFactoryBean the spring-managed bean. + */ + TcpOutboundChannelAdapterSpec(AbstractConnectionFactory connectionFactoryBean) { + this.target = new TcpSendingMessageHandler(); + this.connectionFactory = null; + this.target.setConnectionFactory(connectionFactoryBean); + } + + /** + * Construct an instance using the provided connection factory spec. + * @param connectionFactorySpec the spec. + */ + TcpOutboundChannelAdapterSpec(AbstractConnectionFactorySpec connectionFactorySpec) { + this.target = new TcpSendingMessageHandler(); + this.connectionFactory = connectionFactorySpec.get(); + this.target.setConnectionFactory(this.connectionFactory); + } + + /** + * @param isClientMode true to connect in client mode + * @return the spec. + * @see TcpSendingMessageHandler#setClientMode(boolean) + */ + public TcpOutboundChannelAdapterSpec clientMode(boolean isClientMode) { + this.target.setClientMode(isClientMode); + return _this(); + } + + /** + * @param retryInterval the client mode retry interval to set. + * @return the spec. + * @see TcpSendingMessageHandler#setRetryInterval(long) + */ + public TcpOutboundChannelAdapterSpec retryInterval(long retryInterval) { + this.target.setRetryInterval(retryInterval); + return _this(); + } + + /** + * @param taskScheduler the scheduler for connecting in client mode. + * @return the spec. + * @see TcpSendingMessageHandler#setTaskScheduler(TaskScheduler) + */ + public TcpOutboundChannelAdapterSpec taskScheduler(TaskScheduler taskScheduler) { + this.target.setTaskScheduler(taskScheduler); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + return this.connectionFactory == null ? Collections.emptyList() + : Collections.singletonList(this.connectionFactory); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundGatewaySpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundGatewaySpec.java new file mode 100644 index 0000000000..bbb5d6c2e9 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundGatewaySpec.java @@ -0,0 +1,96 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import java.util.Collection; +import java.util.Collections; +import java.util.function.Function; + +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.ip.tcp.TcpOutboundGateway; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.messaging.Message; + +/** + * A {@link MessageHandlerSpec} for {@link TcpOutboundGateway}s. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class TcpOutboundGatewaySpec extends MessageHandlerSpec + implements ComponentsRegistration { + + private final AbstractClientConnectionFactory connectionFactory; + + /** + * Construct an instance using an existing spring-managed connection factory. + * @param connectionFactoryBean the spring-managed bean. + */ + public TcpOutboundGatewaySpec(AbstractClientConnectionFactory connectionFactoryBean) { + this.target = new TcpOutboundGateway(); + this.connectionFactory = null; + this.target.setConnectionFactory(connectionFactoryBean); + } + + /** + * Construct an instance using the supplied connection factory spec. + * @param connectionFactorySpec the spec. + */ + public TcpOutboundGatewaySpec(TcpClientConnectionFactorySpec connectionFactorySpec) { + this.target = new TcpOutboundGateway(); + this.connectionFactory = connectionFactorySpec.get(); + this.target.setConnectionFactory(this.connectionFactory); + } + + /** + * @param remoteTimeout the remote timeout to set. + * @return the spec. + * @see TcpOutboundGateway#setRemoteTimeout(long) + */ + public TcpOutboundGatewaySpec remoteTimeout(long remoteTimeout) { + this.target.setRemoteTimeout(remoteTimeout); + return _this(); + } + + /** + * Configure a {@link Function} that will be invoked at runtime to determine the destination to + * which a message will be sent. Typically used with a Java 8 Lambda expression: + * 
+	 * {@code
+	 * .remoteTimeout(m -> m.getHeaders().get('rto'))
+	 * }
+	 *
+ * @param remoteTimeoutFunction the function. + * @param

the message payload type. + * @return the spec. + * @see TcpOutboundGateway#setRemoteTimeoutExpression(org.springframework.expression.Expression) + */ + public

TcpOutboundGatewaySpec remoteTimeout(Function, ?> remoteTimeoutFunction) { + this.target.setRemoteTimeoutExpression(new FunctionExpression<>(remoteTimeoutFunction)); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + return this.connectionFactory == null ? Collections.emptyList() + : Collections.singletonList(this.connectionFactory); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpServerConnectionFactorySpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpServerConnectionFactorySpec.java new file mode 100644 index 0000000000..36a030331c --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpServerConnectionFactorySpec.java @@ -0,0 +1,61 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; + +/** + * An {@link AbstractConnectionFactorySpec} for {@link AbstractServerConnectionFactory}s. + * @author Gary Russell + * + * @since 5.0 + * + */ +public class TcpServerConnectionFactorySpec + extends AbstractConnectionFactorySpec { + + TcpServerConnectionFactorySpec(int port) { + this(port, false); + } + + TcpServerConnectionFactorySpec(int port, boolean nio) { + super(nio ? new TcpNioServerConnectionFactory(port) : new TcpNetServerConnectionFactory(port)); + } + + /** + * @param localAddress the local address. + * @return the spec. + * @see AbstractServerConnectionFactory#setLocalAddress(String) + */ + public TcpServerConnectionFactorySpec localAddress(String localAddress) { + this.target.setLocalAddress(localAddress); + return _this(); + } + + /** + * @param backlog the backlog. + * @return the spec. + * @see AbstractServerConnectionFactory#setBacklog(int) + */ + public TcpServerConnectionFactorySpec backlog(int backlog) { + this.target.setBacklog(backlog); + return _this(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Udp.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Udp.java new file mode 100644 index 0000000000..8b2263537f --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Udp.java @@ -0,0 +1,97 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import java.util.function.Function; + +import org.springframework.messaging.Message; + +/** + * Factory methods for UDP. + * + * @author Gary Russell + * @since 5.0 + * + */ +public final class Udp { + + private Udp() { + super(); + } + + /** + * Create an inbound unicast channel adapter using the supplied port. + * @param port the port. + * @return the spec. + */ + public static UdpInboundChannelAdapterSpec inboundAdapter(int port) { + return new UdpInboundChannelAdapterSpec(port); + } + + /** + * Create an inbound multicast channel adapter using the supplied port and + * group. + * @param port the port. + * @param multicastGroup the group. + * @return the spec. + */ + public static UdpInboundChannelAdapterSpec inboundMulticastAdapter(int port, String multicastGroup) { + return new UdpInboundChannelAdapterSpec(port, multicastGroup); + } + + /** + * Create an inbound unicast channel adapter using the supplied destination + * expression. + * @param destinationExpression destination expression. + * @return the spec. + */ + public static UdpUnicastOutboundChannelAdapterSpec outboundAdapter(String destinationExpression) { + return new UdpUnicastOutboundChannelAdapterSpec(destinationExpression); + } + + /** + * Create an inbound unicast channel adapter using the supplied destination + * expression. + * @param destinationFunction function that will provide the destination based on the message. + * @return the spec. + */ + public static UdpUnicastOutboundChannelAdapterSpec outboundAdapter(Function, ?> destinationFunction) { + return new UdpUnicastOutboundChannelAdapterSpec(destinationFunction); + } + + /** + * Create an inbound multicast channel adapter using the supplied destination + * expression. + * @param destinationExpression destination expression. + * @return the spec. + */ + public static UdpMulticastOutboundChannelAdapterSpec outboundMulticastAdapter(String destinationExpression) { + return new UdpMulticastOutboundChannelAdapterSpec(destinationExpression); + } + + /** + * Create an inbound multicast channel adapter using the supplied destination + * expression. + * @param destinationFunction function that will provide the destination based on the message. + * @return the spec. + */ + public static UdpMulticastOutboundChannelAdapterSpec outboundMulticastAdapter(Function, ?> + destinationFunction) { + return new UdpMulticastOutboundChannelAdapterSpec(destinationFunction); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpInboundChannelAdapterSpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpInboundChannelAdapterSpec.java new file mode 100644 index 0000000000..537a2d0ebf --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpInboundChannelAdapterSpec.java @@ -0,0 +1,155 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import java.net.DatagramSocket; +import java.util.concurrent.Executor; + +import org.springframework.integration.dsl.MessageProducerSpec; +import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter; +import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter; +import org.springframework.scheduling.TaskScheduler; + +/** + * A {@link MessageProducerSpec} for {@link UnicastReceivingChannelAdapter}s. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class UdpInboundChannelAdapterSpec + extends MessageProducerSpec { + + UdpInboundChannelAdapterSpec(int port) { + super(new UnicastReceivingChannelAdapter(port)); + } + + UdpInboundChannelAdapterSpec(int port, String multicastGroup) { + super(new MulticastReceivingChannelAdapter(multicastGroup, port)); + } + + /** + * @param soTimeout set the timeout socket option. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setSoTimeout(int) + */ + public UdpInboundChannelAdapterSpec soTimeout(int soTimeout) { + this.target.setSoTimeout(soTimeout); + return _this(); + } + + /** + * @param taskScheduler set the task scheduler. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setTaskScheduler(TaskScheduler) + */ + public UdpInboundChannelAdapterSpec taskScheduler(TaskScheduler taskScheduler) { + this.target.setTaskScheduler(taskScheduler); + return _this(); + } + + /** + * @param soReceiveBufferSize set the receive buffer size socket option. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setSoReceiveBufferSize(int) + */ + public UdpInboundChannelAdapterSpec soReceiveBufferSize(int soReceiveBufferSize) { + this.target.setSoReceiveBufferSize(soReceiveBufferSize); + return _this(); + } + + /** + * @param receiveBufferSize set the receive buffer size. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setReceiveBufferSize(int) + */ + public UdpInboundChannelAdapterSpec receiveBufferSize(int receiveBufferSize) { + this.target.setReceiveBufferSize(receiveBufferSize); + return _this(); + } + + /** + * @param lengthCheck set the length check boolean. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setLengthCheck(boolean) + */ + public UdpInboundChannelAdapterSpec lengthCheck(boolean lengthCheck) { + this.target.setLengthCheck(lengthCheck); + return _this(); + } + + /** + * @param localAddress set the local address. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setLocalAddress(String) + */ + public UdpInboundChannelAdapterSpec localAddress(String localAddress) { + this.target.setLocalAddress(localAddress); + return _this(); + } + + /** + * @param poolSize set the pool size. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setPoolSize(int) + */ + public UdpInboundChannelAdapterSpec poolSize(int poolSize) { + this.target.setPoolSize(poolSize); + return _this(); + } + + /** + * @param taskExecutor set the task executor. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setTaskExecutor(Executor) + */ + public UdpInboundChannelAdapterSpec taskExecutor(Executor taskExecutor) { + this.target.setTaskExecutor(taskExecutor); + return _this(); + } + + /** + * @param socket set the socket. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setSocket(DatagramSocket) + */ + public UdpInboundChannelAdapterSpec socket(DatagramSocket socket) { + this.target.setSocket(socket); + return _this(); + } + + /** + * @param soSendBufferSize set the send buffer size socket option. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setSoSendBufferSize(int) + */ + public UdpInboundChannelAdapterSpec soSendBufferSize(int soSendBufferSize) { + this.target.setSoSendBufferSize(soSendBufferSize); + return _this(); + } + + /** + * @param lookupHost set true to reverse lookup the host. + * @return the spec. + * @see UnicastReceivingChannelAdapter#setLookupHost(boolean) + */ + public UdpInboundChannelAdapterSpec lookupHost(boolean lookupHost) { + this.target.setLookupHost(lookupHost); + return _this(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpMulticastOutboundChannelAdapterSpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpMulticastOutboundChannelAdapterSpec.java new file mode 100644 index 0000000000..c8ad0b2f02 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpMulticastOutboundChannelAdapterSpec.java @@ -0,0 +1,54 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import java.util.function.Function; + +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.ip.udp.MulticastSendingMessageHandler; +import org.springframework.messaging.Message; + +/** + * A {@link MessageHandlerSpec} for {@link MulticastSendingMessageHandler}s. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class UdpMulticastOutboundChannelAdapterSpec + extends AbstractUdpOutboundChannelAdapterSpec { + + UdpMulticastOutboundChannelAdapterSpec(String destinationExpression) { + this.target = new MulticastSendingMessageHandler(destinationExpression); + } + + UdpMulticastOutboundChannelAdapterSpec(Function, ?> destinationFunction) { + this.target = new MulticastSendingMessageHandler(new FunctionExpression<>(destinationFunction)); + } + + /** + * @param timeToLive the timeToLive. + * @return the spec. + * @see MulticastSendingMessageHandler#setTimeToLive(int) + */ + public UdpMulticastOutboundChannelAdapterSpec timeToLive(int timeToLive) { + ((MulticastSendingMessageHandler) this.target).setTimeToLive(timeToLive); + return _this(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpUnicastOutboundChannelAdapterSpec.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpUnicastOutboundChannelAdapterSpec.java new file mode 100644 index 0000000000..5988d7dad8 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpUnicastOutboundChannelAdapterSpec.java @@ -0,0 +1,43 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import java.util.function.Function; + +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.messaging.Message; + +/** + * A {@link MessageHandlerSpec} for + * {@link org.springframework.integration.ip.udp.UnicastSendingMessageHandler}s. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class UdpUnicastOutboundChannelAdapterSpec + extends AbstractUdpOutboundChannelAdapterSpec { + + UdpUnicastOutboundChannelAdapterSpec(Function, ?> destinationFunction) { + super(destinationFunction); + } + + UdpUnicastOutboundChannelAdapterSpec(String destinationExpression) { + super(destinationExpression); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/package-info.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/package-info.java new file mode 100644 index 0000000000..f93424236d --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides TCP/UDP Component support for the Java DSL. + */ +package org.springframework.integration.ip.dsl; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index 595d74b1ea..5858fd310d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -267,6 +267,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements if (this.scheduledFuture != null) { this.scheduledFuture.cancel(true); } + this.clientModeConnectionManager = null; if (this.clientConnectionFactory != null) { this.clientConnectionFactory.stop(); } @@ -282,24 +283,6 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements return this.active; } - public void stop(Runnable callback) { - synchronized (this.lifecycleMonitor) { - if (this.active) { - this.active = false; - if (this.scheduledFuture != null) { - this.scheduledFuture.cancel(true); - } - this.clientModeConnectionManager = null; - if (this.clientConnectionFactory != null) { - this.clientConnectionFactory.stop(callback); - } - if (this.serverConnectionFactory != null) { - this.serverConnectionFactory.stop(callback); - } - } - } - } - /** * @return the clientConnectionFactory */ @@ -330,8 +313,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements } /** - * @param isClientMode - * the isClientMode to set + * @param isClientMode the isClientMode to set */ public void setClientMode(boolean isClientMode) { this.isClientMode = isClientMode; @@ -350,8 +332,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements } /** - * @param retryInterval - * the retryInterval to set + * @param retryInterval the retryInterval to set */ public void setRetryInterval(long retryInterval) { this.retryInterval = retryInterval; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 3d5c4d4a0e..acb3156a0c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -44,7 +44,6 @@ import java.util.concurrent.TimeUnit; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; -import org.springframework.context.SmartLifecycle; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.context.IntegrationObjectSupport; @@ -60,7 +59,7 @@ import org.springframework.util.Assert; * */ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport - implements ConnectionFactory, SmartLifecycle, ApplicationEventPublisherAware { + implements ConnectionFactory, ApplicationEventPublisherAware { protected static final int DEFAULT_REPLY_TIMEOUT = 10000; @@ -275,6 +274,16 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport this.soTrafficClass = soTrafficClass; } + /** + * Set the host; requires the factory to be stopped. + * @param host the host. + * @since 5.0 + */ + public void setHost(String host) { + Assert.state(!isRunning(), "Cannot change the host while running"); + this.host = host; + } + /** * @return the host */ @@ -282,6 +291,16 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport return this.host; } + /** + * Set the port; requires the factory to be stopped. + * @param port the port. + * @since 5.0 + */ + public void setPort(int port) { + Assert.state(!isRunning(), "Cannot change the host while running"); + this.port = port; + } + /** * @return the port */ @@ -395,6 +414,15 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport this.singleUse = singleUse; } + /** + * If true, sockets created by this factory will be reused. + * Inverse of {@link #setSingleUse(boolean)}. + * @param leaveOpen The keepOpen to set. + * @since 5.0 + */ + public void setLeaveOpen(boolean leaveOpen) { + this.singleUse = !leaveOpen; + } public void setInterceptorFactoryChain(TcpConnectionInterceptorFactoryChain interceptorFactoryChain) { this.interceptorFactoryChain = interceptorFactoryChain; @@ -752,26 +780,6 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport throw new UnsupportedOperationException("Nio server factory must override this method"); } - @Override - public int getPhase() { - return 0; - } - - /** - * We are controlled by the startup options of - * the bound endpoint. - */ - @Override - public boolean isAutoStartup() { - return false; - } - - @Override - public void stop(Runnable callback) { - stop(); - callback.run(); - } - protected void addConnection(TcpConnectionSupport connection) { synchronized (this.connections) { if (!this.active) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index ba87ca838a..949cf6d3e6 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -388,21 +388,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact this.pool.removeAllIdleItems(); } - @Override - public int getPhase() { - return this.targetConnectionFactory.getPhase(); - } - - @Override - public boolean isAutoStartup() { - return this.targetConnectionFactory.isAutoStartup(); - } - - @Override - public void stop(Runnable callback) { - this.targetConnectionFactory.stop(callback); - } - private final class CachedConnection extends TcpConnectionInterceptorSupport { private final AtomicBoolean released = new AtomicBoolean(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java index ca74042ef4..b506dc168e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java @@ -16,7 +16,7 @@ package org.springframework.integration.ip.tcp.connection; -import org.springframework.context.SmartLifecycle; +import org.springframework.context.Lifecycle; @@ -27,7 +27,7 @@ import org.springframework.context.SmartLifecycle; * @since 2.0 * */ -public interface ConnectionFactory extends SmartLifecycle { +public interface ConnectionFactory extends Lifecycle { TcpConnection getConnection() throws Exception; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java index b07633d4c5..36c92f137a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java @@ -21,7 +21,10 @@ import java.net.DatagramSocket; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.MulticastSocket; +import java.net.SocketAddress; +import java.net.URI; +import org.springframework.expression.Expression; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; @@ -99,6 +102,28 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler super(address, port, lengthCheck, acknowledge, ackHost, ackPort, ackTimeout); } + /** + * Construct MulticastSendingMessageHandler based on the destination SpEL expression to + * determine the target destination at runtime against requestMessage. + * @param destinationExpression the SpEL expression to evaluate the target destination + * at runtime. Must evaluate to {@link String}, {@link URI} or {@link SocketAddress}. + * @since 5.0 + */ + public MulticastSendingMessageHandler(Expression destinationExpression) { + super(destinationExpression); + } + + /** + * Construct MulticastSendingMessageHandler based on the destination SpEL expression to + * determine the target destination at runtime against requestMessage. + * @param destinationExpression the SpEL expression to evaluate the target destination + * at runtime. Must evaluate to {@link String}, {@link URI} or {@link SocketAddress}. + * @since 5.0 + */ + public MulticastSendingMessageHandler(String destinationExpression) { + super(destinationExpression); + } + @Override protected DatagramSocket getSocket() throws IOException { if (this.getTheSocket() == null) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 3af9ba2bf8..9d77730379 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -74,6 +74,15 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece this.mapper.setLengthCheck(lengthCheck); } + /** + * @param lengthCheck if true, the incoming packet is expected to have a four + * byte binary length header. + * @since 5.0 + */ + public void setLengthCheck(boolean lengthCheck) { + this.mapper.setLengthCheck(lengthCheck); + } + @Override public boolean isLongLived() { return true; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index 9d62a3df3a..bac767bb98 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -114,10 +114,10 @@ public class UnicastSendingMessageHandler extends } /** - * Construct UnicastSendingMessageHandler based on the destination SpEL expression to determine - * the target destination at runtime against requestMessage. - * @param destinationExpression the SpEL expression to evaluate the target destination at runtime. - * Must evaluate to {@link String}, {@link URI} or {@link SocketAddress}. + * Construct UnicastSendingMessageHandler based on the destination SpEL expression to + * determine the target destination at runtime against requestMessage. + * @param destinationExpression the SpEL expression to evaluate the target destination + * at runtime. Must evaluate to {@link String}, {@link URI} or {@link SocketAddress}. * @since 4.3 */ public UnicastSendingMessageHandler(String destinationExpression) { @@ -129,10 +129,10 @@ public class UnicastSendingMessageHandler extends } /** - * Construct UnicastSendingMessageHandler based on the destination SpEL expression to determine - * the target destination at runtime against requestMessage. - * @param destinationExpression the SpEL expression to evaluate the target destination at runtime. - * Must evaluate to {@link String}, {@link URI} or {@link SocketAddress}. + * Construct UnicastSendingMessageHandler based on the destination SpEL expression to + * determine the target destination at runtime against requestMessage. + * @param destinationExpression the SpEL expression to evaluate the target destination + * at runtime. Must evaluate to {@link String}, {@link URI} or {@link SocketAddress}. * @since 4.3 */ public UnicastSendingMessageHandler(Expression destinationExpression) { @@ -217,6 +217,15 @@ public class UnicastSendingMessageHandler extends } } + /** + * @param lengthCheck if true, a four byte binary length header is added to the + * packet, allowing the receiver to check for data truncation. + * @since 5.0 + */ + public void setLengthCheck(boolean lengthCheck) { + this.mapper.setLengthCheck(lengthCheck); + } + @Override public void doStart() { if (this.acknowledge) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java index a4a24a816f..d2c8e69419 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java @@ -18,6 +18,7 @@ package org.springframework.integration.ip.util; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter; /** * Convenience class providing methods for testing IP components. @@ -65,6 +66,37 @@ public final class TestingUtilities { } } + /** + * Wait for a server connection factory to actually start listening before + * starting a test. Waits for up to 10 seconds by default. + * @param adapter The server connection factory. + * @param delay How long to wait in milliseconds; default 10000 (10 seconds) if null. + * @throws IllegalStateException If the server does not start listening in time. + */ + public static void waitListening(UnicastReceivingChannelAdapter adapter, Long delay) + throws IllegalStateException { + if (delay == null) { + delay = 100L; + } + else { + delay = delay / 100; + } + int n = 0; + while (!adapter.isListening()) { + try { + Thread.sleep(100); + } + catch (InterruptedException e1) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e1); + } + + if (n++ > delay) { + throw new IllegalStateException("Server didn't start listening."); + } + } + } + /** * Wait for a server connection factory to stop listening. * Waits for up to 10 seconds by default. diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/ConnectionFacforyTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/ConnectionFacforyTests.java new file mode 100644 index 0000000000..b343b0988d --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/ConnectionFacforyTests.java @@ -0,0 +1,69 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; + +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.util.TestingUtilities; +import org.springframework.integration.transformer.ObjectToStringTransformer; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Gary Russell + * @since 5.0 + * + */ +public class ConnectionFacforyTests { + + @Test + public void test() throws Exception { + ApplicationEventPublisher publisher = e -> { }; + AbstractServerConnectionFactory server = Tcp.netServer(0).backlog(2).soTimeout(5000).get(); + final AtomicReference> received = new AtomicReference<>(); + final CountDownLatch latch = new CountDownLatch(1); + server.registerListener(m -> { + received.set(new ObjectToStringTransformer().transform(m)); + latch.countDown(); + return false; + }); + server.setApplicationEventPublisher(publisher); + server.afterPropertiesSet(); + server.start(); + TestingUtilities.waitListening(server, null); + AbstractClientConnectionFactory client = Tcp.netClient("localhost", server.getPort()).get(); + client.setApplicationEventPublisher(publisher); + client.afterPropertiesSet(); + client.start(); + client.getConnection().send(new GenericMessage<>("foo")); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertEquals("foo", received.get().getPayload()); + client.stop(); + server.stop(); + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java new file mode 100644 index 0000000000..7fecc668d5 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java @@ -0,0 +1,186 @@ +/* + * Copyright 2016 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 + * + * http://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.ip.dsl; + +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.channel.QueueChannel; +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.dsl.context.IntegrationFlowRegistration; +import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.TcpSendingMessageHandler; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.serializer.TcpCodecs; +import org.springframework.integration.ip.udp.MulticastSendingMessageHandler; +import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter; +import org.springframework.integration.ip.util.TestingUtilities; +import org.springframework.integration.support.MessageBuilder; +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.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 5.0 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class IpIntegrationTests { + + @Autowired + private AbstractServerConnectionFactory server1; + + @Autowired + private IntegrationFlowContext flowContext; + + @Autowired + @Qualifier("outUdpAdapter.input") + private MessageChannel udpOut; + + @Autowired + private UnicastReceivingChannelAdapter udpInbound; + + @Autowired + private QueueChannel udpIn; + + @Test + public void testTcpAdapters() throws Exception { + ApplicationEventPublisher publisher = e -> { }; + AbstractServerConnectionFactory server = Tcp.netServer(0).backlog(2).soTimeout(5000).id("server").get(); + assertEquals("server", server.getComponentName()); + server.setApplicationEventPublisher(publisher); + server.afterPropertiesSet(); + TcpReceivingChannelAdapter inbound = Tcp.inboundAdapter(server).get(); + QueueChannel received = new QueueChannel(); + inbound.setOutputChannel(received); + inbound.afterPropertiesSet(); + inbound.start(); + TestingUtilities.waitListening(server, null); + AbstractClientConnectionFactory client = Tcp.netClient("localhost", server.getPort()).id("client").get(); + assertEquals("client", client.getComponentName()); + client.setApplicationEventPublisher(publisher); + client.afterPropertiesSet(); + TcpSendingMessageHandler handler = Tcp.outboundAdapter(client).get(); + handler.start(); + handler.handleMessage(new GenericMessage<>("foo")); + Message receivedMessage = received.receive(10000); + assertNotNull(receivedMessage); + assertEquals("foo", Transformers.objectToString().transform(receivedMessage).getPayload()); + client.stop(); + server.stop(); + } + + @Test + public void testTcpGateways() { + TestingUtilities.waitListening(this.server1, null); + IntegrationFlow flow = f -> f + .handle(Tcp.outboundGateway(Tcp.netClient("localhost", this.server1.getPort()) + .serializer(TcpCodecs.crlf()) + .deserializer(TcpCodecs.lengthHeader1()) + .id("client1")) + .remoteTimeout(m -> 5000)) + .transform(Transformers.objectToString()); + IntegrationFlowRegistration theFlow = this.flowContext.registration(flow).register(); + assertThat(theFlow.getMessagingTemplate().convertSendAndReceive("foo", String.class), equalTo("FOO")); + } + + @Test + public void testUdp() { + TestingUtilities.waitListening(this.udpInbound, null); + Message outMessage = MessageBuilder.withPayload("foo") + .setHeader("udp_dest", "udp://localhost:" + this.udpInbound.getPort()) + .build(); + this.udpOut.send(outMessage); + Message received = this.udpIn.receive(10000); + assertNotNull(received); + assertEquals("foo", Transformers.objectToString().transform(received).getPayload()); + } + + @Test + public void testUdpInheritance() { + UdpMulticastOutboundChannelAdapterSpec udpMulticastOutboundChannelAdapterSpec = + Udp.outboundMulticastAdapter("headers['udp_dest']"); + + UdpMulticastOutboundChannelAdapterSpec udpMulticastOutboundChannelAdapterSpec1 = + udpMulticastOutboundChannelAdapterSpec.lengthCheck(true); + + UdpMulticastOutboundChannelAdapterSpec udpMulticastOutboundChannelAdapterSpec2 = + udpMulticastOutboundChannelAdapterSpec1.timeToLive(10); + + assertThat(udpMulticastOutboundChannelAdapterSpec2.get(), instanceOf(MulticastSendingMessageHandler.class)); + } + + @Configuration + @EnableIntegration + public static class Config { + + @Bean + public AbstractServerConnectionFactory server1() { + return Tcp.netServer(0) + .serializer(TcpCodecs.lengthHeader1()) + .deserializer(TcpCodecs.crlf()) + .get(); + } + + @Bean + public IntegrationFlow inTcpGateway() { + return IntegrationFlows.from(Tcp.inboundGateway(server1())) + .transform(Transformers.objectToString()) + .transform(String::toUpperCase) + .get(); + } + + @Bean + public IntegrationFlow inUdpAdapter() { + return IntegrationFlows.from(Udp.inboundAdapter(0)) + .channel(udpIn()) + .get(); + } + + @Bean + public QueueChannel udpIn() { + return new QueueChannel(); + } + + @Bean + public IntegrationFlow outUdpAdapter() { + return f -> f.handle(Udp.outboundAdapter(m -> m.getHeaders().get("udp_dest"))); + } + + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests.java index 9c549569ed..85ade45c71 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests.java @@ -16,11 +16,11 @@ package org.springframework.integration.ip.tcp; -import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; @@ -45,7 +45,6 @@ public class AutoStartTests { @Test public void testNetIn() throws Exception { - assertFalse(cfS1.isAutoStartup()); DirectFieldAccessor dfa = new DirectFieldAccessor(cfS1); assertNull(dfa.getPropertyValue("serverSocket")); startAndStop(); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java index 68304f142f..450b490299 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java @@ -25,8 +25,10 @@ import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.apache.log4j.Level; import org.junit.AfterClass; import org.junit.BeforeClass; +import org.junit.Rule; import org.junit.Test; import org.springframework.integration.jdbc.lock.DefaultLockRepository; @@ -35,6 +37,7 @@ import org.springframework.integration.leader.Context; import org.springframework.integration.leader.DefaultCandidate; import org.springframework.integration.leader.event.LeaderEventPublisher; import org.springframework.integration.support.leader.LockRegistryLeaderInitiator; +import org.springframework.integration.test.rule.Log4jLevelAdjuster; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; @@ -45,6 +48,9 @@ import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; */ public class JdbcLockRegistryLeaderInitiatorTests { + @Rule + public Log4jLevelAdjuster logAdjuster = new Log4jLevelAdjuster(Level.DEBUG, "org.springframework.integration"); + public static EmbeddedDatabase dataSource; @BeforeClass @@ -162,9 +168,9 @@ public class JdbcLockRegistryLeaderInitiatorTests { private static class CountingPublisher implements LeaderEventPublisher { - private CountDownLatch granted; + private final CountDownLatch granted; - private CountDownLatch revoked; + private final CountDownLatch revoked; private volatile LockRegistryLeaderInitiator initiator; diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundGatewaySpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundGatewaySpec.java index 8c25f36e41..a0810b608e 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundGatewaySpec.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundGatewaySpec.java @@ -108,7 +108,7 @@ public class JmsOutboundGatewaySpec extends MessageHandlerSpec * {@code diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests-context.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests-context.xml index 7c8b331424..434178bccc 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests-context.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests-context.xml @@ -18,7 +18,7 @@ query="{'name' : 'Bob'}" entity-class="java.lang.Object" auto-startup="false"> - + - + - + - + - + - + @@ -87,7 +87,7 @@ entity-class="java.lang.Object" mongo-converter="mongoConverter" auto-startup="false"> - +