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

INT-4162: TCP/UDP DSL

Browse Source 
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
This commit is contained in:
Gary Russell
2016-11-05 14:28:05 -04:00
committed by Artem Bilan
parent 46348291ff
commit 65ec234697
30 changed files with 1758 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java
View File

@@ -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<AbstractConnectionFactory> implements SmartLifecycle, BeanNameAware,
BeanFactoryAware, ApplicationEventPublisherAware {
public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<AbstractConnectionFactory>
implements Lifecycle, BeanNameAware, BeanFactoryAware, ApplicationEventPublisherAware {
private volatile AbstractConnectionFactory connectionFactory;
@@ -446,33 +446,6 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
this.connectionFactory.stop();
}
/**
* @return phase
* @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#getPhase()
*/
@Override
public int getPhase() {
return this.connectionFactory.getPhase();
}
/**
* @return isAutoStartup
* @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#isAutoStartup()
*/
@Override
public boolean isAutoStartup() {
return this.connectionFactory.isAutoStartup();
}
/**
* @param callback The Runnable to invoke.
* @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#stop(java.lang.Runnable)
*/
@Override
public void stop(Runnable callback) {
this.connectionFactory.stop(callback);
}
@Override
public boolean isRunning() {
return this.connectionFactory.isRunning();

222
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractConnectionFactorySpec.java Normal file
View File

@@ -0,0 +1,222 @@
/*
* 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.concurrent.Executor;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.dsl.IntegrationComponentSpec;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain;
import org.springframework.integration.ip.tcp.connection.TcpMessageMapper;
import org.springframework.integration.ip.tcp.connection.TcpSocketSupport;
/**
* An {@link IntegrationComponentSpec} for {@link AbstractConnectionFactory}s.
* @param <S> the target {@link AbstractConnectionFactorySpec} implementation type.
* @param <C> the target {@link AbstractConnectionFactory} implementation type.
*
* @author Gary Russell
* @since 5.0
*
*/
public abstract class AbstractConnectionFactorySpec
<S extends AbstractConnectionFactorySpec<S, C>, C extends AbstractConnectionFactory>
extends IntegrationComponentSpec<S, C> {
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();
}
}

131
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/AbstractUdpOutboundChannelAdapterSpec.java Normal file
View File

@@ -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 <S> the target {@link AbstractUdpOutboundChannelAdapterSpec} implementation type.
*
* @author Gary Russell
* @since 5.0
*
*/
public abstract class AbstractUdpOutboundChannelAdapterSpec<S extends AbstractUdpOutboundChannelAdapterSpec<S>>
extends MessageHandlerSpec<S, UnicastSendingMessageHandler> {
protected AbstractUdpOutboundChannelAdapterSpec() {
super();
}
AbstractUdpOutboundChannelAdapterSpec(String destinationExpression) {
this.target = new UnicastSendingMessageHandler(destinationExpression);
}
AbstractUdpOutboundChannelAdapterSpec(Function<Message<?>, ?> 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<Message<?>, 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();
}
}

162
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java Normal file
View File

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

41
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpClientConnectionFactorySpec.java Normal file
View File

@@ -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, AbstractClientConnectionFactory> {
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));
}
}

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

@@ -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<TcpInboundChannelAdapterSpec, TcpReceivingChannelAdapter>
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<Object> getComponentsToRegister() {
return this.connectionFactory == null ? Collections.emptyList()
: Collections.singletonList(this.connectionFactory);
}
}

96
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundGatewaySpec.java Normal file
View File

@@ -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<TcpInboundGatewaySpec, TcpInboundGateway>
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<Object> getComponentsToRegister() {
return this.connectionFactory == null ? Collections.emptyList()
: Collections.singletonList(this.connectionFactory);
}
}

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

@@ -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<TcpOutboundChannelAdapterSpec, TcpSendingMessageHandler>
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<Object> getComponentsToRegister() {
return this.connectionFactory == null ? Collections.emptyList()
: Collections.singletonList(this.connectionFactory);
}
}

96
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundGatewaySpec.java Normal file
View File

@@ -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<TcpOutboundGatewaySpec, TcpOutboundGateway>
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:
* <pre class="code">
* {@code
* .remoteTimeout(m -> m.getHeaders().get('rto'))
* }
* </pre>
* @param remoteTimeoutFunction the function.
* @param <P> the message payload type.
* @return the spec.
* @see TcpOutboundGateway#setRemoteTimeoutExpression(org.springframework.expression.Expression)
*/
public <P> TcpOutboundGatewaySpec remoteTimeout(Function<Message<P>, ?> remoteTimeoutFunction) {
this.target.setRemoteTimeoutExpression(new FunctionExpression<>(remoteTimeoutFunction));
return _this();
}
@Override
public Collection<Object> getComponentsToRegister() {
return this.connectionFactory == null ? Collections.emptyList()
: Collections.singletonList(this.connectionFactory);
}
}

61
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpServerConnectionFactorySpec.java Normal file
View File

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

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

@@ -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<Message<?>, ?> 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<Message<?>, ?>
destinationFunction) {
return new UdpMulticastOutboundChannelAdapterSpec(destinationFunction);
}
}

155
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpInboundChannelAdapterSpec.java Normal file
View File

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

54
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpMulticastOutboundChannelAdapterSpec.java Normal file
View File

@@ -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> {
UdpMulticastOutboundChannelAdapterSpec(String destinationExpression) {
this.target = new MulticastSendingMessageHandler(destinationExpression);
}
UdpMulticastOutboundChannelAdapterSpec(Function<Message<?>, ?> 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();
}
}

43
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/UdpUnicastOutboundChannelAdapterSpec.java Normal file
View File

@@ -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> {
UdpUnicastOutboundChannelAdapterSpec(Function<Message<?>, ?> destinationFunction) {
super(destinationFunction);
}
UdpUnicastOutboundChannelAdapterSpec(String destinationExpression) {
super(destinationExpression);
}
}

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

@@ -0,0 +1,4 @@
/**
* Provides TCP/UDP Component support for the Java DSL.
*/
package org.springframework.integration.ip.dsl;

25
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java
View File

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

52
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

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

15
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
View File

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

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java
View File

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

25
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java
View File

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

9
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
View File

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

25
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

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

32
spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java
View File

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

69
spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/ConnectionFacforyTests.java Normal file
View File

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

186
spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java Normal file
View File

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

3
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/AutoStartTests.java
View File

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

10
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java
View File

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

2
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundGatewaySpec.java
View File

@@ -108,7 +108,7 @@ public class JmsOutboundGatewaySpec extends MessageHandlerSpec<JmsOutboundGatewa
}
/**
* Configure a {@link Function} that will be invoked at run time to determine the destination to
* 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:
* <pre class="code">
* {@code

14
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests-context.xml
View File

@@ -18,7 +18,7 @@
query="{'name' : 'Bob'}"
entity-class="java.lang.Object"
auto-startup="false">
<int:poller fixed-rate="100" />
<int:poller fixed-delay="100" />
</int-mongodb:inbound-channel-adapter>
<int-mongodb:inbound-channel-adapter id="mongoInboundAdapterNamedFactory"
@@ -26,7 +26,7 @@
channel="replyChannel"
query="{'name' : 'Bob'}"
auto-startup="false">
<int:poller fixed-rate="5000" />
<int:poller fixed-delay="5000" />
</int-mongodb:inbound-channel-adapter>
<int-mongodb:inbound-channel-adapter id="mongoInboundAdapterWithTemplate"
@@ -36,7 +36,7 @@
expect-single-result="true"
entity-class="org.springframework.integration.mongodb.rules.MongoDbAvailableTests.Person"
auto-startup="false">
<int:poller fixed-rate="5000" />
<int:poller fixed-delay="5000" />
</int-mongodb:inbound-channel-adapter>
<int-mongodb:inbound-channel-adapter id="mongoInboundAdapterWithNamedCollection"
@@ -46,7 +46,7 @@
query="{'name' : 'Bob'}"
entity-class="java.lang.Object"
auto-startup="false">
<int:poller fixed-rate="5000" />
<int:poller fixed-delay="5000" />
</int-mongodb:inbound-channel-adapter>
<int-mongodb:inbound-channel-adapter id="mongoInboundAdapterWithNamedCollectionExpression"
@@ -56,7 +56,7 @@
query="{'name' : 'Bob'}"
entity-class="java.lang.Object"
auto-startup="false">
<int:poller fixed-rate="5000" />
<int:poller fixed-delay="5000" />
</int-mongodb:inbound-channel-adapter>
<int-mongodb:inbound-channel-adapter id="inboundAdapterWithOnSuccessDisposition"
@@ -64,7 +64,7 @@
query="{'name' : 'Bob'}"
auto-startup="false">
<int:poller fixed-rate="200" max-messages-per-poll="1">
<int:poller fixed-delay="200" max-messages-per-poll="1">
<int:advice-chain synchronization-factory="syncFactory">
<bean
class="org.springframework.integration.mongodb.config.MongoDbInboundChannelAdapterIntegrationTests.TestMessageSourceAdvice" />
@@ -87,7 +87,7 @@
entity-class="java.lang.Object"
mongo-converter="mongoConverter"
auto-startup="false">
<int:poller fixed-rate="100" />
<int:poller fixed-delay="100" />
</int-mongodb:inbound-channel-adapter>
<bean id="documentCleaner"

11
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java
View File

@@ -35,12 +35,12 @@ import org.springframework.data.mongodb.core.query.BasicQuery;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.integration.aop.AbstractMessageSourceAdvice;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.mongodb.rules.MongoDbAvailable;
import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -62,7 +62,7 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
private MongoTemplate mongoTemplate;
@Autowired
private PollableChannel replyChannel;
private QueueChannel replyChannel;
@Autowired
@Qualifier("mongoInboundAdapter")
@@ -106,6 +106,7 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
assertNotNull(this.replyChannel.receive(10000));
this.mongoInboundAdapter.stop();
this.replyChannel.purge(null);
}
@Test
@@ -121,6 +122,7 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
assertEquals("Bob", message.getPayload().get(0).get("name"));
this.mongoInboundAdapterNamedFactory.stop();
this.replyChannel.purge(null);
}
@Test
@@ -136,6 +138,7 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
assertEquals("Bob", message.getPayload().getName());
this.mongoInboundAdapterWithTemplate.stop();
this.replyChannel.purge(null);
}
@Test
@@ -151,6 +154,7 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
assertEquals("Bob", message.getPayload().get(0).getName());
this.mongoInboundAdapterWithNamedCollection.stop();
this.replyChannel.purge(null);
}
@Test
@@ -166,6 +170,7 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
assertEquals("Bob", message.getPayload().get(0).getName());
this.mongoInboundAdapterWithNamedCollectionExpression.stop();
this.replyChannel.purge(null);
}
@Test
@@ -181,6 +186,7 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
this.inboundAdapterWithOnSuccessDisposition.stop();
assertNull(this.mongoTemplate.findOne(new Query(Criteria.where("name").is("Bob")), Person.class, "data"));
this.replyChannel.purge(null);
}
@Test
@@ -197,6 +203,7 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
assertNotNull(replyChannel.receive(10000));
this.mongoInboundAdapterWithConverter.stop();
this.replyChannel.purge(null);
}
@Test(expected = BeanDefinitionParsingException.class)