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

eliminated all javadoc warnings; no API changes, with the exception of adding generics metadata to GatewayProxyFactoryBean (now 'implements FactoryBean<Object>').

Browse Source
This commit is contained in:
Chris Beams
2010-05-20 17:50:32 +00:00
parent cf6027e3b7
commit 233994116c
38 changed files with 453 additions and 499 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java
View File

@@ -16,7 +16,9 @@
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;
@@ -27,7 +29,7 @@ import org.springframework.util.Assert;
/**
* Base class for all TCP/UDP MessageHandlers.
*
*
* @author Gary Russell
* @since 2.0
*/
@@ -36,9 +38,9 @@ public abstract class AbstractInternetProtocolSendingMessageHandler implements M
protected final Log logger = LogFactory.getLog(getClass());
protected final SocketAddress destinationAddress;
protected final String host;
protected final int port;
protected volatile int soSendBufferSize = -1;
@@ -57,7 +59,8 @@ public abstract class AbstractInternetProtocolSendingMessageHandler implements M
/**
* @see {@link Socket#setSoTimeout(int)} and {@link DatagramSocket#setSoTimeout(int)}
* @see Socket#setSoTimeout(int)
* @see DatagramSocket#setSoTimeout(int)
* @param timeout
*/
public void setSoTimeout(int timeout) {
@@ -65,21 +68,22 @@ public abstract class AbstractInternetProtocolSendingMessageHandler implements M
}
/**
* @see {@link Socket#setReceiveBufferSize(int)} and {@link DatagramSocket#setReceiveBufferSize(int)}
* @see Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
* @param size
*/
public void setSoReceiveBufferSize(int size) {
}
/**
* @see {@link Socket#setSendBufferSize(int)} and {@link DatagramSocket#setSendBufferSize(int)}
* @see Socket#setSendBufferSize(int)
* @see DatagramSocket#setSendBufferSize(int)
* @param size
*/
public void setSoSendBufferSize(int size) {
this.soSendBufferSize = size;
}
/**
* @return the port
*/

8
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -89,11 +89,11 @@ public abstract class IpAdapterParserUtils {
/**
* Adds a constructor-arg to the bean definition with the value
* of the attribute whose name is provided if that attribute is
* defined in the given element.
* Adds a constructor-arg to the provided bean definition builder
* with the value of the attribute whose name is provided if that
* attribute is defined in the given element.
*
* @param beanDefinition the bean definition to be configured
* @param builder the bean definition builder to be configured
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be
* used to populate the property

4
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java
View File

@@ -100,7 +100,7 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
}
/**
* @see {@link Socket#setKeepAlive(boolean)}.
* @see Socket#setKeepAlive(boolean)
* @param soKeepAlive the soKeepAlive to set
*/
public void setSoKeepAlive(boolean soKeepAlive) {
@@ -108,7 +108,7 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
}
/**
* @See {@link MessageFormats}
* @see MessageFormats
* @param messageFormat the messageFormat to set
*/
public void setMessageFormat(int messageFormat) {

3
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java
View File

@@ -25,13 +25,14 @@ import java.nio.ByteBuffer;
* data is wrapped in a wire protocol based on the messageFormat property.
*
* @author Gary Russell
*
*/
public class NetSocketWriter extends AbstractSocketWriter {
protected Socket socket;
/**
* Constructs a NetSocketWriter for the Socket.
*
* @param socket The socket.
*/
public NetSocketWriter(Socket socket) {

7
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java
View File

@@ -255,7 +255,7 @@ public class NioSocketReader extends AbstractSocketReader {
/**
* Reads data into the rawBuffer for non-deterministic algorithms.
* @return true If data is available.
* @return true if data is available.
* @throws IOException
*/
protected boolean readChannelNonDeterministic() throws IOException {
@@ -287,8 +287,6 @@ public class NioSocketReader extends AbstractSocketReader {
/**
* Allocates a ByteBuffer of the requested length using normal or
* direct buffers, depending on the usingDirectBuffers field.
* @param length
* @return
*/
protected ByteBuffer allocate(int length) {
ByteBuffer buffer;
@@ -319,9 +317,6 @@ public class NioSocketReader extends AbstractSocketReader {
return this.channel.socket().getInetAddress();
}
/**
* @return the usingeDirectBuffers
*/
public boolean isUsingDirectBuffers() {
return usingDirectBuffers;
}

53
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp;
import java.io.IOException;
@@ -22,46 +23,45 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
/**
* A {@link SocketWriter} that writes to a {@link java.nio.channels.SocketChannel}. The
* A {@link SocketWriter} that writes to a {@link SocketChannel}. The
* data is wrapped in a wire protocol based on the messageFormat property.
*
* @author Gary Russell
*
*/
public class NioSocketWriter extends AbstractSocketWriter {
protected SocketChannel channel;
/**
* If true, direct buffers are used.
* @see {@link ByteBuffer} for more information.
* If true, direct buffers are used.
* @see ByteBuffer for more information
*/
protected boolean usingDirectBuffers;
/**
* A buffer containing the length part when the messageFormat is
* A buffer containing the length part when the messageFormat is
* {@link MessageFormats#FORMAT_LENGTH_HEADER}.
*/
protected ByteBuffer lengthPart;
/**
* A buffer containing the STX for when the messageFormat is
* A buffer containing the STX for when the messageFormat is
* {@link MessageFormats#FORMAT_STX_ETX}.
*/
protected ByteBuffer stxPart;
/**
* A buffer containing the ETX for when the messageFormat is
* A buffer containing the ETX for when the messageFormat is
* {@link MessageFormats#FORMAT_STX_ETX}.
*/
protected ByteBuffer etxPart;
/**
* A buffer containing the CRLF for when the messageFormat is
* A buffer containing the CRLF for when the messageFormat is
* {@link MessageFormats#FORMAT_CRLF}.
*/
protected ByteBuffer crLfPart;
/**
* If we are using direct buffers, we don't want to churn them using
* normal heap management. But,
@@ -70,19 +70,16 @@ public class NioSocketWriter extends AbstractSocketWriter {
* We handle this with a blocking queue.
*/
protected BlockingQueue<ByteBuffer> buffers;
protected int maxBuffers = 2;
protected int bufferCount = 0;
private int sendBufferSize;
/**
* @param socket
*/
public NioSocketWriter(SocketChannel channel,
int maxBuffers,
int sendBufferSize) {
public NioSocketWriter(SocketChannel channel,
int maxBuffers,
int sendBufferSize) {
this.channel = channel;
this.maxBuffers = maxBuffers;
if (sendBufferSize <= 0) {
@@ -91,9 +88,9 @@ public class NioSocketWriter extends AbstractSocketWriter {
this.sendBufferSize = sendBufferSize;
buffers = new LinkedBlockingQueue<ByteBuffer>(maxBuffers);
}
/**
* @param usingDirectBuffers the usingDirectBuffers to set
* @param usingDirectBuffers whether direct buffers are to be used
*/
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
@@ -115,13 +112,13 @@ public class NioSocketWriter extends AbstractSocketWriter {
buffer.clear();
return buffer;
}
protected void returnBuffer(ByteBuffer buffer) {
if (buffer != null) {
buffers.offer(buffer);
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCrLfFormat(byte[])
*/
@@ -218,7 +215,6 @@ public class NioSocketWriter extends AbstractSocketWriter {
} finally {
returnBuffer(buffer);
}
}
synchronized (channel) {
if (stxPart == null) {
@@ -257,5 +253,4 @@ public class NioSocketWriter extends AbstractSocketWriter {
} catch (IOException e) {}
}
}

11
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java
View File

@@ -84,25 +84,24 @@ public class SimpleTcpNetOutboundGateway extends
}
/**
* @param obj
* @return
* @see java.lang.Object#equals(java.lang.Object)
* @see java.lang.Object#equals(Object)
* @return whether the MessageHandler delegate for this Gateway is equal to the provided object
*/
public boolean equals(Object obj) {
return handler.equals(obj);
}
/**
* @return
* @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#getPort()
* @return the port number of the MessageHandler delegate for this Gateway
*/
public int getPort() {
return handler.getPort();
}
/**
* @return
* @see java.lang.Object#hashCode()
* @return hashcode value of the MessageHandler delegate for this Gateway
*/
public int hashCode() {
return handler.hashCode();
@@ -184,7 +183,7 @@ public class SimpleTcpNetOutboundGateway extends
}
/**
* @param customSocketReaderClass the customSocketReader to set
* @param customSocketReaderClassName the {@link NetSocketReader} class to use
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")

17
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java
View File

@@ -25,7 +25,7 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.ip.util.SocketIoUtils;
/**
* Tcp Receiving Channel adapter that uses a {@link java.net.Socket}. Each
* Tcp Receiving Channel adapter that uses a {@link Socket}. Each
* connected socket uses a dedicated thread so the pool size must be set
* accordingly.
*
@@ -38,8 +38,8 @@ public class TcpNetReceivingChannelAdapter extends
protected ServerSocket serverSocket;
protected Class<NetSocketReader> customSocketReaderClass;
/**
* Constructs a TcpNetReceivingChannelAdapter that listens on the port.
* @param port The port.
* Constructs a TcpNetReceivingChannelAdapter that listens on the provided port.
* @param port the port on which to listen
*/
public TcpNetReceivingChannelAdapter(int port) {
super(port);
@@ -87,8 +87,6 @@ public class TcpNetReceivingChannelAdapter extends
* Constructs a {@link NetSocketReader} and calls its {@link NetSocketReader#assembledData}
* method repeatedly; for each assembled message, calls {@link #sendMessage(Message)} with
* the mapped message.
*
* @param socket
*/
protected void handleSocket(Socket socket) {
NetSocketReader reader = SocketIoUtils.createNetReader(messageFormat,
@@ -106,11 +104,6 @@ public class TcpNetReceivingChannelAdapter extends
}
}
/**
* @param reader
* @return
* @throws Exception
*/
protected void processMessage(NetSocketReader reader)
throws Exception {
Message<byte[]> message = mapper.toMessage(reader);
@@ -131,8 +124,8 @@ public class TcpNetReceivingChannelAdapter extends
}
/**
* @param customSocketReaderClass the customSocketReader to set
* @throws ClassNotFoundException
* @param customSocketReaderClassName the {@link NetSocketReader} class to use
* @throws ClassNotFoundException if the named class cannot be loaded
*/
@SuppressWarnings("unchecked")
public void setCustomSocketReaderClassName(

2
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java
View File

@@ -104,7 +104,7 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
/**
* Set the underlying {@link MulticastSocket} time to live property.
* @param timeToLive {@see MulticastSocket#setTimeToLive(int)}
* @param timeToLive {@link MulticastSocket#setTimeToLive(int)}
*/
public void setTimeToLive(int timeToLive) {
this.timeToLive = timeToLive;

617
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -1,308 +1,309 @@
/*
* 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.BindException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.util.Assert;
/**
* A {@link MessageHandler} implementation that maps a Message into
* a UDP datagram packet and sends that to the specified host and port.
*
* Messages can be basic, with no support for reliability, can be prefixed
* by a length so the receiving end can detect truncation, and can require
* a UDP acknowledgment to confirm delivery.
*
* @author Gary Russell
* @since 2.0
*/
public class UnicastSendingMessageHandler extends
AbstractInternetProtocolSendingMessageHandler implements Runnable {
protected final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
protected volatile DatagramSocket socket;
/**
* If true adds headers to instruct receiving adapter to return an ack.
*/
protected volatile boolean waitForAck = false;
protected volatile int ackPort;
protected volatile int ackTimeout = 5000;
protected volatile int ackCounter = 1;
protected volatile Map<String, CountDownLatch> ackControl = Collections
.synchronizedMap(new HashMap<String, CountDownLatch>());
protected volatile DatagramSocket ackSocket;
protected volatile Exception fatalException;
protected int soReceiveBufferSize = -1;
/**
* Basic constructor; no reliability; no acknowledgment.
* @param host Destination host.
* @param port Destination port.
*/
public UnicastSendingMessageHandler(String host, int port) {
super(host, port);
this.mapper.setLengthCheck(false);
this.mapper.setAcknowledge(false);
}
/**
* Can used to add a length to each packet which can be checked at the destination.
* @param host Destination Host.
* @param port Destination Port.
* @param lengthCheck If true, packets will contain a length.
*/
public UnicastSendingMessageHandler(String host, int port, boolean lengthCheck) {
super(host, port);
this.mapper.setLengthCheck(lengthCheck);
this.mapper.setAcknowledge(false);
}
/**
* Add an acknowledgment request to packets.
* @param host Destination Host.
* @param port Destination Port.
* @param acknowledge If true, packets will request acknowledgment.
* @param ackHost The host to which acks should be sent. Required if ack true.
* @param ackPort The port to which acks should be sent.
* @param ackTimeout How long we will wait (milliseconds) for the ack.
*/
public UnicastSendingMessageHandler(String host,
int port,
boolean acknowledge,
String ackHost,
int ackPort,
int ackTimeout) {
super(host, port);
setReliabilityAttributes(false, acknowledge, ackHost, ackPort,
ackTimeout);
}
/**
* Add a length and/or acknowledgment request to packets.
* @param host Destination Host.
* @param port Destination Port.
* @param lengthCheck If true, packets will contain a length.
* @param acknowledge If true, packets will request acknowledgment.
* @param ackHost The host to which acks should be sent. Required if ack true.
* @param ackPort The port to which acks should be sent.
* @param ackTimeout How long we will wait (milliseconds) for the ack.
*/
public UnicastSendingMessageHandler(String host,
int port,
boolean lengthCheck,
boolean acknowledge,
String ackHost,
int ackPort,
int ackTimeout) {
super(host, port);
setReliabilityAttributes(lengthCheck, acknowledge, ackHost, ackPort,
ackTimeout);
}
protected void setReliabilityAttributes(boolean lengthCheck,
boolean acknowledge, String ackHost, int ackPort, int ackTimeout) {
this.mapper.setLengthCheck(lengthCheck);
this.waitForAck = acknowledge;
this.mapper.setAcknowledge(acknowledge);
this.mapper.setAckAddress(ackHost + ":" + ackPort);
this.ackPort = ackPort;
if (ackTimeout > 0) {
this.ackTimeout = ackTimeout;
}
if (acknowledge) {
Assert.hasLength(ackHost);
this.executorService = Executors
.newSingleThreadExecutor(new ThreadFactory() {
private AtomicInteger n = new AtomicInteger();
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("UDP-Ack-Handler-" + n.getAndIncrement());
thread.setDaemon(true);
return thread;
}
});
this.executorService.execute(this);
}
}
public void handleMessage(Message<?> message)
throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
CountDownLatch countdownLatch = null;
String messageId = message.getHeaders().getId().toString();
try {
DatagramPacket packet;
if (this.waitForAck) {
if (this.fatalException != null) {
throw new MessagingException(message, "Acknowledgment failure", fatalException);
}
countdownLatch = new CountDownLatch(ackCounter);
this.ackControl.put(messageId, countdownLatch);
}
packet = this.mapper.fromMessage(message);
this.send(packet);
logger.debug("Sent packet for message id " + message.getHeaders().getId());
if (this.waitForAck) {
if (!countdownLatch.await(this.ackTimeout, TimeUnit.MILLISECONDS)) {
throw new MessagingException(message, "Failed to receive UDP Ack in " + ackTimeout + " millis");
}
}
}
catch (MessagingException e) {
throw e;
}
catch (Exception e) {
try{
socket.close();
}
catch (Exception e1) { }
socket = null;
throw new MessageHandlingException(message, "failed to send UDP packet", e);
}
finally {
if (countdownLatch != null)
this.ackControl.remove(messageId);
}
}
protected void send(DatagramPacket packet) throws Exception {
DatagramSocket socket = this.getSocket();
packet.setSocketAddress(this.destinationAddress);
socket.send(packet);
}
protected synchronized DatagramSocket getSocket() throws IOException {
if (this.socket == null) {
this.socket = new DatagramSocket();
setSocketAttributes(this.socket);
}
return this.socket;
}
protected void setSocketAttributes(DatagramSocket socket) throws SocketException {
if (this.soTimeout >= 0) {
socket.setSoTimeout(this.soTimeout);
}
if (this.soSendBufferSize > 0) {
socket.setSendBufferSize(this.soSendBufferSize);
}
}
/**
* 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);
}
DatagramPacket ackPack = new DatagramPacket(new byte[100], 100);
while(true) {
this.ackSocket.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());
}
CountDownLatch latch = this.ackControl.get(id);
if (latch != null) {
latch.countDown();
}
}
}
catch (IOException e) {
logger.error("Error on UDP Acknowledge thread" + e.getMessage());
fatalException = e;
}
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);
}
}
}
/**
* If exposed as an MBean, can be used to restart the ack thread if a fatal
* (bind) error occurred, without bouncing the JVM.
*/
public void restartAckThread() {
if (fatalException == null) {
return;
}
this.fatalException = null;
this.executorService.execute(this);
}
public void shutDown() {
DatagramSocket socket = this.ackSocket;
this.ackSocket = null;
if (socket != null) {
socket.close();
}
}
/**
* @see {@link Socket#setReceiveBufferSize(int)} and {@link DatagramSocket#setReceiveBufferSize(int)}
* @param size
*/
public void setSoReceiveBufferSize(int size) {
this.soReceiveBufferSize = size;
}
}
/*
* 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.BindException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.util.Assert;
/**
* A {@link MessageHandler} implementation that maps a Message into
* a UDP datagram packet and sends that to the specified host and port.
*
* Messages can be basic, with no support for reliability, can be prefixed
* by a length so the receiving end can detect truncation, and can require
* a UDP acknowledgment to confirm delivery.
*
* @author Gary Russell
* @since 2.0
*/
public class UnicastSendingMessageHandler extends
AbstractInternetProtocolSendingMessageHandler implements Runnable {
protected final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
protected volatile DatagramSocket socket;
/**
* If true adds headers to instruct receiving adapter to return an ack.
*/
protected volatile boolean waitForAck = false;
protected volatile int ackPort;
protected volatile int ackTimeout = 5000;
protected volatile int ackCounter = 1;
protected volatile Map<String, CountDownLatch> ackControl = Collections
.synchronizedMap(new HashMap<String, CountDownLatch>());
protected volatile DatagramSocket ackSocket;
protected volatile Exception fatalException;
protected int soReceiveBufferSize = -1;
/**
* Basic constructor; no reliability; no acknowledgment.
* @param host Destination host.
* @param port Destination port.
*/
public UnicastSendingMessageHandler(String host, int port) {
super(host, port);
this.mapper.setLengthCheck(false);
this.mapper.setAcknowledge(false);
}
/**
* Can used to add a length to each packet which can be checked at the destination.
* @param host Destination Host.
* @param port Destination Port.
* @param lengthCheck If true, packets will contain a length.
*/
public UnicastSendingMessageHandler(String host, int port, boolean lengthCheck) {
super(host, port);
this.mapper.setLengthCheck(lengthCheck);
this.mapper.setAcknowledge(false);
}
/**
* Add an acknowledgment request to packets.
* @param host Destination Host.
* @param port Destination Port.
* @param acknowledge If true, packets will request acknowledgment.
* @param ackHost The host to which acks should be sent. Required if ack true.
* @param ackPort The port to which acks should be sent.
* @param ackTimeout How long we will wait (milliseconds) for the ack.
*/
public UnicastSendingMessageHandler(String host,
int port,
boolean acknowledge,
String ackHost,
int ackPort,
int ackTimeout) {
super(host, port);
setReliabilityAttributes(false, acknowledge, ackHost, ackPort,
ackTimeout);
}
/**
* Add a length and/or acknowledgment request to packets.
* @param host Destination Host.
* @param port Destination Port.
* @param lengthCheck If true, packets will contain a length.
* @param acknowledge If true, packets will request acknowledgment.
* @param ackHost The host to which acks should be sent. Required if ack true.
* @param ackPort The port to which acks should be sent.
* @param ackTimeout How long we will wait (milliseconds) for the ack.
*/
public UnicastSendingMessageHandler(String host,
int port,
boolean lengthCheck,
boolean acknowledge,
String ackHost,
int ackPort,
int ackTimeout) {
super(host, port);
setReliabilityAttributes(lengthCheck, acknowledge, ackHost, ackPort,
ackTimeout);
}
protected void setReliabilityAttributes(boolean lengthCheck,
boolean acknowledge, String ackHost, int ackPort, int ackTimeout) {
this.mapper.setLengthCheck(lengthCheck);
this.waitForAck = acknowledge;
this.mapper.setAcknowledge(acknowledge);
this.mapper.setAckAddress(ackHost + ":" + ackPort);
this.ackPort = ackPort;
if (ackTimeout > 0) {
this.ackTimeout = ackTimeout;
}
if (acknowledge) {
Assert.hasLength(ackHost);
this.executorService = Executors
.newSingleThreadExecutor(new ThreadFactory() {
private AtomicInteger n = new AtomicInteger();
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("UDP-Ack-Handler-" + n.getAndIncrement());
thread.setDaemon(true);
return thread;
}
});
this.executorService.execute(this);
}
}
public void handleMessage(Message<?> message)
throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
CountDownLatch countdownLatch = null;
String messageId = message.getHeaders().getId().toString();
try {
DatagramPacket packet;
if (this.waitForAck) {
if (this.fatalException != null) {
throw new MessagingException(message, "Acknowledgment failure", fatalException);
}
countdownLatch = new CountDownLatch(ackCounter);
this.ackControl.put(messageId, countdownLatch);
}
packet = this.mapper.fromMessage(message);
this.send(packet);
logger.debug("Sent packet for message id " + message.getHeaders().getId());
if (this.waitForAck) {
if (!countdownLatch.await(this.ackTimeout, TimeUnit.MILLISECONDS)) {
throw new MessagingException(message, "Failed to receive UDP Ack in " + ackTimeout + " millis");
}
}
}
catch (MessagingException e) {
throw e;
}
catch (Exception e) {
try{
socket.close();
}
catch (Exception e1) { }
socket = null;
throw new MessageHandlingException(message, "failed to send UDP packet", e);
}
finally {
if (countdownLatch != null)
this.ackControl.remove(messageId);
}
}
protected void send(DatagramPacket packet) throws Exception {
DatagramSocket socket = this.getSocket();
packet.setSocketAddress(this.destinationAddress);
socket.send(packet);
}
protected synchronized DatagramSocket getSocket() throws IOException {
if (this.socket == null) {
this.socket = new DatagramSocket();
setSocketAttributes(this.socket);
}
return this.socket;
}
protected void setSocketAttributes(DatagramSocket socket) throws SocketException {
if (this.soTimeout >= 0) {
socket.setSoTimeout(this.soTimeout);
}
if (this.soSendBufferSize > 0) {
socket.setSendBufferSize(this.soSendBufferSize);
}
}
/**
* 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);
}
DatagramPacket ackPack = new DatagramPacket(new byte[100], 100);
while(true) {
this.ackSocket.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());
}
CountDownLatch latch = this.ackControl.get(id);
if (latch != null) {
latch.countDown();
}
}
}
catch (IOException e) {
logger.error("Error on UDP Acknowledge thread" + e.getMessage());
fatalException = e;
}
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);
}
}
}
/**
* If exposed as an MBean, can be used to restart the ack thread if a fatal
* (bind) error occurred, without bouncing the JVM.
*/
public void restartAckThread() {
if (fatalException == null) {
return;
}
this.fatalException = null;
this.executorService.execute(this);
}
public void shutDown() {
DatagramSocket socket = this.ackSocket;
this.ackSocket = null;
if (socket != null) {
socket.close();
}
}
/**
* @see Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
*/
public void setSoReceiveBufferSize(int size) {
this.soReceiveBufferSize = size;
}
}

32
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/util/SocketIoUtils.java
View File

@@ -33,14 +33,6 @@ import org.springframework.integration.message.MessageMappingException;
*/
public class SocketIoUtils {
/**
* @param messageFormat
* @param customSocketReaderClass
* @param socket
* @param receiveBufferSize
* @param soReceiveBufferSize
* @return
*/
public static NetSocketReader createNetReader(int messageFormat,
Class<NetSocketReader> customSocketReaderClass,
Socket socket,
@@ -67,12 +59,6 @@ public class SocketIoUtils {
return reader;
}
/**
* @param messageFormat
* @param socket
* @param customSocketWriterClass
* @return
*/
public static NetSocketWriter createNetWriter(int messageFormat,
Class<NetSocketWriter> customSocketWriterClass, Socket socket) {
NetSocketWriter writer;
@@ -90,15 +76,6 @@ public class SocketIoUtils {
return writer;
}
/**
* @param messageFormat
* @param customSocketReaderClass
* @param socket
* @param receiveBufferSize
* @param usingDirectBuffers
* @param soReceiveBufferSize
* @return
*/
public static NioSocketReader createNioReader(int messageFormat,
Class<NioSocketReader> customSocketReaderClass,
SocketChannel channel,
@@ -127,15 +104,6 @@ public class SocketIoUtils {
return reader;
}
/**
* @param messageFormat
* @param socket
* @param customSocketWriter
* @param maxBuffers
* @param sendBufferSize
* @param usingDirectBuffers
* @return
*/
public static NioSocketWriter createNioWriter(int messageFormat,
Class<NioSocketWriter> customSocketWriterClass,
SocketChannel channel,