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

INT-1149 Support the specification of a network interface via local-address attribute on inbound endpoints and outbound UDP adapter. Docs to follow.

Browse Source
This commit is contained in:
Gary Russell
2010-05-30 18:54:50 +00:00
parent 4fa1483e42
commit 9fef40f179
23 changed files with 667 additions and 334 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java
View File

@@ -44,6 +44,8 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
protected volatile boolean listening;
protected volatile String localAddress;
public AbstractInternetProtocolReceivingChannelAdapter(int port) {
this.port = port;
@@ -101,4 +103,12 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
return listening;
}
public String getLocalAddress() {
return localAddress;
}
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
}

187
spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java
View File

@@ -1,94 +1,93 @@
/*
* Copyright 2002-2010 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;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.concurrent.ExecutorService;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
/**
* Base class for all TCP/UDP MessageHandlers.
*
* @author Gary Russell
* @since 2.0
*/
public abstract class AbstractInternetProtocolSendingMessageHandler implements MessageHandler, CommonSocketOptions {
protected final Log logger = LogFactory.getLog(getClass());
protected final SocketAddress destinationAddress;
protected final String host;
protected final int port;
protected volatile int soSendBufferSize = -1;
protected volatile int soTimeout = -1;
protected volatile ExecutorService executorService;
public AbstractInternetProtocolSendingMessageHandler(String host, int port) {
Assert.notNull(host, "host must not be null");
this.destinationAddress = new InetSocketAddress(host, port);
this.host = host;
this.port = port;
}
/**
* @see Socket#setSoTimeout(int)
* @see DatagramSocket#setSoTimeout(int)
* @param timeout
*/
public void setSoTimeout(int timeout) {
this.soTimeout = timeout;
}
/**
* @see Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
* @param size
*/
public void setSoReceiveBufferSize(int size) {
}
/**
* @see Socket#setSendBufferSize(int)
* @see DatagramSocket#setSendBufferSize(int)
* @param size
*/
public void setSoSendBufferSize(int size) {
this.soSendBufferSize = size;
}
/**
* @return the port
*/
public int getPort() {
return port;
}
}
/*
* Copyright 2002-2010 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;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.concurrent.ExecutorService;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
/**
* Base class for all TCP/UDP MessageHandlers.
*
* @author Gary Russell
* @since 2.0
*/
public abstract class AbstractInternetProtocolSendingMessageHandler implements MessageHandler, CommonSocketOptions {
protected final Log logger = LogFactory.getLog(getClass());
protected final SocketAddress destinationAddress;
protected final String host;
protected final int port;
protected volatile int soSendBufferSize = -1;
protected volatile int soTimeout = -1;
protected volatile ExecutorService executorService;
public AbstractInternetProtocolSendingMessageHandler(String host, int port) {
Assert.notNull(host, "host must not be null");
this.destinationAddress = new InetSocketAddress(host, port);
this.host = host;
this.port = port;
}
/**
* @see Socket#setSoTimeout(int)
* @see DatagramSocket#setSoTimeout(int)
* @param timeout
*/
public void setSoTimeout(int timeout) {
this.soTimeout = timeout;
}
/**
* @see Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
* @param size
*/
public void setSoReceiveBufferSize(int size) {
}
/**
* @see Socket#setSendBufferSize(int)
* @see DatagramSocket#setSendBufferSize(int)
* @param size
*/
public void setSoSendBufferSize(int size) {
this.soSendBufferSize = size;
}
/**
* @return the port
*/
public int getPort() {
return port;
}
}

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

@@ -22,10 +22,37 @@ package org.springframework.integration.ip;
*/
public interface CommonSocketOptions {
void setSoTimeout(int soTimeout);
/**
* @see Socket#setSoTimeout(int)
* @see DatagramSocket#setSoTimeout(int)
* @param timeout
*/
public void setSoTimeout(int soTimeout);
void setSoReceiveBufferSize(int soReceiveBufferSize);
/**
* @see Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
* @param size
*/
public void setSoReceiveBufferSize(int soReceiveBufferSize);
void setSoSendBufferSize(int soSendBufferSize);
/**
* @see Socket#setSendBufferSize(int)
* @see DatagramSocket#setSendBufferSize(int)
* @param size
*/
public void setSoSendBufferSize(int soSendBufferSize);
/**
* On a multi-homed system, specifies the ip address of the network interface used to communicate.
* For inbound adapters and gateways, specifies the interface used to listed for incoming connections.
* If omitted, the endpoint will listen on all available adapters. For the UDP multicast outbound adapter
* specifies the interface to which multicast packets will be sent. For UDP unicast and multicast
* adapters, specifies which interface to which the acknowledgment socket will be bound. Does not
* apply to TCP outbound adapters and gateways.
*
* @param localAddress
*/
public void setLocalAddress(String localAddress);
}

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -88,6 +88,8 @@ public abstract class IpAdapterParserUtils {
static final String SO_TRAFFIC_CLASS = "so-traffic-class";
static final String CLOSE = "close";
static final String LOCAL_ADDRESS = "local-address";
/**
@@ -265,6 +267,7 @@ public abstract class IpAdapterParserUtils {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SO_TIMEOUT);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SO_RECEIVE_BUFFER_SIZE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SO_SEND_BUFFER_SIZE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, LOCAL_ADDRESS);
}
}

7
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java
View File

@@ -66,6 +66,8 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
protected boolean close;
protected String localAddress;
@Override
protected void doStart() {
super.doStart();
@@ -91,6 +93,7 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
this.delegate.setTaskScheduler(getTaskScheduler());
this.delegate.setCustomSocketReaderClassName(customSocketReaderClassName);
this.delegate.setClose(close);
this.delegate.setLocalAddress(localAddress);
super.onInit();
}
@@ -206,6 +209,10 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
return delegate.isListening();
}
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
private class WriteCapableTcpNetReceivingChannelAdapter extends TcpNetReceivingChannelAdapter {
/**

26
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.ip.tcp;
import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
@@ -37,6 +38,7 @@ public class TcpNetReceivingChannelAdapter extends
AbstractTcpReceivingChannelAdapter {
protected ServerSocket serverSocket;
protected Class<NetSocketReader> customSocketReaderClass;
/**
* Constructs a TcpNetReceivingChannelAdapter that listens on the provided port.
@@ -55,11 +57,17 @@ public class TcpNetReceivingChannelAdapter extends
*/
@Override
protected void server() {
while (active) {
while (this.active) {
try {
serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(port, Math.abs(poolSize));
listening = true;
if (this.localAddress == null) {
this.serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(this.port, Math.abs(this.poolSize));
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(port, Math.abs(poolSize), whichNic);
}
this.listening = true;
while (true) {
final Socket socket = serverSocket.accept();
setSocketOptions(socket);
@@ -69,14 +77,14 @@ public class TcpNetReceivingChannelAdapter extends
}});
}
} catch (IOException e) {
if (serverSocket != null) {
if (this.serverSocket != null) {
try {
serverSocket.close();
this.serverSocket.close();
} catch (IOException e1) {}
}
listening = false;
serverSocket = null;
if (active) {
this.listening = false;
this.serverSocket = null;
if (this.active) {
logger.error("Error on ServerSocket", e);
}

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

@@ -95,4 +95,8 @@ public class TcpNetSendingMessageHandler extends
this.writer.doClose();
this.writer = null;
}
public void setLocalAddress(String localAddress) {
logger.warn("localAddress not used on tcp outbound endpoints");
}
}

27
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java
View File

@@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
@@ -65,22 +66,28 @@ public class TcpNioReceivingChannelAdapter extends
@Override
protected void server() {
try {
serverChannel = ServerSocketChannel.open();
listening = true;
serverChannel.configureBlocking(false);
serverChannel.socket().bind(new InetSocketAddress(port),
Math.abs(poolSize));
this.serverChannel = ServerSocketChannel.open();
this.listening = true;
this.serverChannel.configureBlocking(false);
if (this.localAddress == null) {
this.serverChannel.socket().bind(new InetSocketAddress(this.port),
Math.abs(this.poolSize));
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.serverChannel.socket().bind(new InetSocketAddress(whichNic, this.port),
Math.abs(this.poolSize));
}
final Selector selector = Selector.open();
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
doSelect(serverChannel, selector);
this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
doSelect(this.serverChannel, selector);
} catch (IOException e) {
try {
serverChannel.close();
} catch (IOException e1) { }
listening = false;
serverChannel = null;
if (active) {
this.listening = false;
this.serverChannel = null;
if (this.active) {
logger.error("Error on ServerSocketChannel", e);
}
}

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java
View File

@@ -102,4 +102,7 @@ public class TcpNioSendingMessageHandler extends
this.buffsPerConnection = buffsPerConnection;
}
public void setLocalAddress(String localAddress) {
logger.warn("localAddress not used on tcp outbound endpoints");
}
}

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastReceivingChannelAdapter.java
View File

@@ -59,12 +59,15 @@ public class MulticastReceivingChannelAdapter extends UnicastReceivingChannelAda
this.group = group;
}
@Override
protected synchronized DatagramSocket getSocket() {
if (this.socket == null) {
try {
MulticastSocket socket = new MulticastSocket(this.port);
if (localAddress != null) {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
socket.setInterface(whichNic);
}
socket.setSoTimeout(this.soTimeout);
if (this.soReceiveBufferSize > 0) {
socket.setReceiveBufferSize(this.soReceiveBufferSize);

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

@@ -1,126 +1,156 @@
/*
* Copyright 2002-2010 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.udp;
import java.io.IOException;
import java.net.DatagramSocket;
import java.net.MulticastSocket;
import org.springframework.integration.message.MessageHandler;
/**
* A {@link MessageHandler} implementation that maps a Message into
* a UDP datagram packet and sends that to the specified multicast address
* (224.0.0.0 to 239.255.255.255) and port.
*
* The only difference between this and its super class is the
* ability to specify how many acknowledgments are required to
* determine success.
*
* @author Gary Russell
* @since 2.0
*/
public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler {
protected int timeToLive = -1;
/**
* Constructs a MulticastSendingMessageHandler to send data to the multicast address/port.
* @param address The multicast address.
* @param port The port.
*/
public MulticastSendingMessageHandler(String address, int port) {
super(address, port);
}
/**
* Constructs a MulticastSendingMessageHandler to send data to the multicast address/port
* and enables setting the lengthCheck option (if set, a length is prepended to the packet and checked
* at the destination).
* @param address The multicast address.
* @param port The port.
* @param lengthCheck Enable the lengthCheck option.
*/
public MulticastSendingMessageHandler(String address, int port, boolean lengthCheck) {
super(address, port, lengthCheck);
}
/**
* Constructs a MulticastSendingMessageHandler to send data to the multicast address/port
* and enables setting the acknowledge option, where the destination sends a receipt acknowledgment.
* @param address The multicast address.
* @param port The port.
* @param acknowledge Whether or not acknowledgments are required.
* @param ackHost The host to which acknowledgments should be sent; required if acknowledge is true.
* @param ackPort The port to which acknowledgments should be sent; required if acknowledge is true.
* @param ackTimeout How long to wait (milliseconds) for an acknowledgment.
*/
public MulticastSendingMessageHandler(String address, int port,
boolean acknowledge, String ackHost, int ackPort, int ackTimeout) {
super(address, port, acknowledge, ackHost, ackPort, ackTimeout);
}
/**
* Constructs a MulticastSendingMessageHandler to send data to the multicast address/port
* and enables setting the acknowledge option, where the destination sends a receipt acknowledgment.
* @param address The multicast address.
* @param port The port.
* @param lengthCheck Enable the lengthCheck option.
* @param acknowledge Whether or not acknowledgments are required.
* @param ackHost The host to which acknowledgments should be sent; required if acknowledge is true.
* @param ackPort The port to which acknowledgments should be sent; required if acknowledge is true.
* @param ackTimeout How long to wait (milliseconds) for an acknowledgment.
*/
public MulticastSendingMessageHandler(String address, int port,
boolean lengthCheck, boolean acknowledge, String ackHost,
int ackPort, int ackTimeout) {
super(address, port, lengthCheck, acknowledge, ackHost, ackPort, ackTimeout);
}
/**
* If acknowledge = true; how many acks needed for success.
* @param minAcksForSuccess
*/
public void setMinAcksForSuccess(int minAcksForSuccess) {
this.ackCounter = minAcksForSuccess;
}
/**
* Set the underlying {@link MulticastSocket} time to live property.
* @param timeToLive {@link MulticastSocket#setTimeToLive(int)}
*/
public void setTimeToLive(int timeToLive) {
this.timeToLive = timeToLive;
}
protected synchronized DatagramSocket getSocket() throws IOException {
if (this.socket == null) {
MulticastSocket socket = new MulticastSocket();
if (this.timeToLive >= 0) {
socket.setTimeToLive(this.timeToLive);
}
socket.setLoopbackMode(true); // disable loopback to the local port
setSocketAttributes(socket);
this.socket = socket;
}
return this.socket;
}
}
/*
* Copyright 2002-2010 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.udp;
import java.io.IOException;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.NetworkInterface;
import java.net.SocketAddress;
import org.springframework.integration.message.MessageHandler;
/**
* A {@link MessageHandler} implementation that maps a Message into
* a UDP datagram packet and sends that to the specified multicast address
* (224.0.0.0 to 239.255.255.255) and port.
*
* The only difference between this and its super class is the
* ability to specify how many acknowledgments are required to
* determine success.
*
* @author Gary Russell
* @since 2.0
*/
public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler {
protected int timeToLive = -1;
protected String localAddress;
/**
* Constructs a MulticastSendingMessageHandler to send data to the multicast address/port.
* @param address The multicast address.
* @param port The port.
*/
public MulticastSendingMessageHandler(String address, int port) {
super(address, port);
}
/**
* Constructs a MulticastSendingMessageHandler to send data to the multicast address/port
* and enables setting the lengthCheck option (if set, a length is prepended to the packet and checked
* at the destination).
* @param address The multicast address.
* @param port The port.
* @param lengthCheck Enable the lengthCheck option.
*/
public MulticastSendingMessageHandler(String address, int port, boolean lengthCheck) {
super(address, port, lengthCheck);
}
/**
* Constructs a MulticastSendingMessageHandler to send data to the multicast address/port
* and enables setting the acknowledge option, where the destination sends a receipt acknowledgment.
* @param address The multicast address.
* @param port The port.
* @param acknowledge Whether or not acknowledgments are required.
* @param ackHost The host to which acknowledgments should be sent; required if acknowledge is true.
* @param ackPort The port to which acknowledgments should be sent; required if acknowledge is true.
* @param ackTimeout How long to wait (milliseconds) for an acknowledgment.
*/
public MulticastSendingMessageHandler(String address, int port,
boolean acknowledge, String ackHost, int ackPort, int ackTimeout) {
super(address, port, acknowledge, ackHost, ackPort, ackTimeout);
}
/**
* Constructs a MulticastSendingMessageHandler to send data to the multicast address/port
* and enables setting the acknowledge option, where the destination sends a receipt acknowledgment.
* @param address The multicast address.
* @param port The port.
* @param lengthCheck Enable the lengthCheck option.
* @param acknowledge Whether or not acknowledgments are required.
* @param ackHost The host to which acknowledgments should be sent; required if acknowledge is true.
* @param ackPort The port to which acknowledgments should be sent; required if acknowledge is true.
* @param ackTimeout How long to wait (milliseconds) for an acknowledgment.
*/
public MulticastSendingMessageHandler(String address, int port,
boolean lengthCheck, boolean acknowledge, String ackHost,
int ackPort, int ackTimeout) {
super(address, port, lengthCheck, acknowledge, ackHost, ackPort, ackTimeout);
}
@Override
protected synchronized DatagramSocket getSocket() throws IOException {
if (this.socket == null) {
MulticastSocket socket;
if (acknowledge) {
if (logger.isDebugEnabled()) {
logger.debug("Listening for acks on port: " + ackPort);
}
if (localAddress == null) {
socket = new MulticastSocket(this.ackPort);
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
socket = new MulticastSocket(new InetSocketAddress(whichNic, this.ackPort));
}
if (this.soReceiveBufferSize > 0) {
socket.setReceiveBufferSize(this.soReceiveBufferSize);
}
} else {
socket = new MulticastSocket();
}
if (this.timeToLive >= 0) {
socket.setTimeToLive(this.timeToLive);
}
setSocketAttributes(socket);
if (localAddress != null) {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
NetworkInterface intfce = NetworkInterface.getByInetAddress(whichNic);
socket.setNetworkInterface(intfce);
}
this.socket = socket;
}
return this.socket;
}
/**
* If acknowledge = true; how many acks needed for success.
* @param minAcksForSuccess
*/
public void setMinAcksForSuccess(int minAcksForSuccess) {
this.ackCounter = minAcksForSuccess;
}
/**
* Set the underlying {@link MulticastSocket} time to live property.
* @param timeToLive {@link MulticastSocket#setTimeToLive(int)}
*/
public void setTimeToLive(int timeToLive) {
this.timeToLive = timeToLive;
}
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
}

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

@@ -19,6 +19,7 @@ package org.springframework.integration.ip.udp;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
@@ -161,6 +162,8 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
Message<byte[]> message = null;
try {
message = mapper.toMessage(packet);
if (logger.isDebugEnabled())
logger.debug("Received:" + message);
}
catch (Exception e) {
logger.error("Failed to map packet to message ", e);
@@ -186,13 +189,19 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
protected synchronized DatagramSocket getSocket() {
if (this.socket == null) {
try {
this.socket = new DatagramSocket(this.port);
if (localAddress == null) {
this.socket = new DatagramSocket(this.port);
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.socket = new DatagramSocket(this.port, whichNic);
}
this.socket.setSoTimeout(this.soTimeout);
if (this.soReceiveBufferSize > 0) {
this.socket.setReceiveBufferSize(this.soReceiveBufferSize);
}
}
catch (SocketException e) {
catch (IOException e) {
throw new MessagingException("failed to create DatagramSocket", e);
}
}

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

@@ -17,9 +17,9 @@
package org.springframework.integration.ip.udp;
import java.io.IOException;
import java.net.BindException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.util.Collections;
@@ -63,6 +63,8 @@ public class UnicastSendingMessageHandler extends
* If true adds headers to instruct receiving adapter to return an ack.
*/
protected volatile boolean waitForAck = false;
protected volatile boolean acknowledge = false;
protected volatile int ackPort;
@@ -73,13 +75,16 @@ public class UnicastSendingMessageHandler extends
protected volatile Map<String, CountDownLatch> ackControl = Collections
.synchronizedMap(new HashMap<String, CountDownLatch>());
protected volatile DatagramSocket ackSocket;
protected volatile Exception fatalException;
protected int soReceiveBufferSize = -1;
protected String localAddress;
private CountDownLatch ackLatch;
private boolean ackThreadRunning;
/**
* Basic constructor; no reliability; no acknowledgment.
* @param host Destination host.
@@ -157,6 +162,7 @@ public class UnicastSendingMessageHandler extends
}
if (acknowledge) {
Assert.hasLength(ackHost);
this.acknowledge = true;
this.executorService = Executors
.newSingleThreadExecutor(new ThreadFactory() {
private AtomicInteger n = new AtomicInteger();
@@ -167,13 +173,25 @@ public class UnicastSendingMessageHandler extends
return thread;
}
});
this.executorService.execute(this);
}
}
public void handleMessage(Message<?> message)
throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
if (this.acknowledge) {
if (!this.ackThreadRunning) {
synchronized(this) {
if (!this.ackThreadRunning) {
ackLatch = new CountDownLatch(1);
this.executorService.execute(this);
try {
ackLatch.await(10000, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) { }
}
}
}
}
CountDownLatch countdownLatch = null;
String messageId = message.getHeaders().getId().toString();
try {
@@ -187,7 +205,7 @@ public class UnicastSendingMessageHandler extends
}
packet = this.mapper.fromMessage(message);
this.send(packet);
logger.debug("Sent packet for message id " + message.getHeaders().getId());
logger.debug("Sent packet for message " + message);
if (this.waitForAck) {
if (!countdownLatch.await(this.ackTimeout, TimeUnit.MILLISECONDS)) {
throw new MessagingException(message, "Failed to receive UDP Ack in " + ackTimeout + " millis");
@@ -219,7 +237,22 @@ public class UnicastSendingMessageHandler extends
protected synchronized DatagramSocket getSocket() throws IOException {
if (this.socket == null) {
this.socket = new DatagramSocket();
if (acknowledge) {
if (logger.isDebugEnabled()) {
logger.debug("Listening for acks on port: " + ackPort);
}
if (localAddress == null) {
this.socket = new DatagramSocket(this.ackPort);
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.socket = new DatagramSocket(this.ackPort, whichNic);
}
if (this.soReceiveBufferSize > 0) {
socket.setReceiveBufferSize(this.soReceiveBufferSize);
}
} else {
this.socket = new DatagramSocket();
}
setSocketAttributes(this.socket);
}
return this.socket;
@@ -238,18 +271,12 @@ public class UnicastSendingMessageHandler extends
* Process acknowledgments, if requested.
*/
public void run() {
Exception fatalException = null;
try {
if (logger.isDebugEnabled()) {
logger.debug("Listening for acks on port: " + ackPort);
}
this.ackSocket = new DatagramSocket(this.ackPort);
if (this.soReceiveBufferSize > 0) {
ackSocket.setReceiveBufferSize(this.soReceiveBufferSize);
}
this.ackThreadRunning = true;
ackLatch.countDown();
DatagramPacket ackPack = new DatagramPacket(new byte[100], 100);
while(true) {
this.ackSocket.receive(ackPack);
this.getSocket().receive(ackPack);
String id = new String(ackPack.getData(), ackPack.getOffset(), ackPack.getLength());
if (logger.isDebugEnabled()) {
logger.debug("Received ack for " + id + " from " + ackPack.getAddress().getHostAddress());
@@ -261,20 +288,12 @@ public class UnicastSendingMessageHandler extends
}
}
catch (IOException e) {
logger.error("Error on UDP Acknowledge thread" + e.getMessage());
fatalException = e;
if (this.socket != null && !this.socket.isClosed()) {
logger.error("Error on UDP Acknowledge thread:" + e.getMessage());
}
}
finally {
if (this.ackSocket != null) {
this.ackSocket.close();
}
if (fatalException instanceof BindException) {
logger.fatal("Failed to bind to acknowledge port: " + ackPort);
this.fatalException = fatalException;
}
else {
this.executorService.execute(this);
}
this.ackThreadRunning = false;
}
}
@@ -291,10 +310,9 @@ public class UnicastSendingMessageHandler extends
}
public void shutDown() {
DatagramSocket socket = this.ackSocket;
this.ackSocket = null;
if (socket != null) {
socket.close();
socket = null;
}
}
@@ -306,4 +324,9 @@ public class UnicastSendingMessageHandler extends
this.soReceiveBufferSize = size;
}
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
}

12
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd
View File

@@ -245,6 +245,18 @@ receive the next message.
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="so-keep-alive" type="xsd:string" />
<xsd:attribute name="local-address" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
On a multi-homed system, specifies the ip address of the network interface used to communicate.
For inbound adapters and gateways, specifies the interface used to listed for incoming connections.
If omitted, the endpoint will listen on all available adapters. For the UDP multicast outbound adapter
specifies the interface to which multicast packets will be sent. For UDP unicast and multicast
adapters, specifies which interface to which the acknowledgment socket will be bound. Does not
apply to TCP outbound adapters and gateways.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:schema>