Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fixed the UDP and multicast test cases by merging the ResponseCorrelator into the UDP transport implementation 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@385794 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
James Strachan 2006-03-14 12:31:44 +00:00
parent daf48bb297
commit 72b382d0bd
23 changed files with 471 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
activemq-core/project.xml
View File

@ -360,10 +360,6 @@
<exclude>**/MultipleTestsWithSpringFactoryBeanTest.*</exclude>
<exclude>**/MultipleTestsWithXBeanFactoryBeanTest.*</exclude>
<exclude>**/MultipleTestsWithSpringXBeanFactoryBeanTest.*</exclude>
<!-- TODO FIX ME -->
<exclude>**/UdpSendReceiveWithTwoConnectionsTest.*</exclude>
<exclude>**/MulticastTransportTest.*</exclude>
</excludes>
</unitTest>
<resources>

5
activemq-core/src/main/java/org/apache/activemq/command/BaseEndpoint.java
View File

@ -33,4 +33,9 @@ public class BaseEndpoint implements Endpoint {
return name;
}
public String toString() {
return "Endpoint[" + name + "]";
}
}

23
activemq-core/src/main/java/org/apache/activemq/command/LastPartialCommand.java
View File

@ -31,9 +31,8 @@ public class LastPartialCommand extends BaseCommand {
public LastPartialCommand() {
}
public LastPartialCommand(Command command) {
setCommandId(command.getCommandId());
setResponseRequired(command.isResponseRequired());
public LastPartialCommand(boolean responseRequired) {
setResponseRequired(responseRequired);
}
public byte getDataStructureType() {
@ -44,4 +43,22 @@ public class LastPartialCommand extends BaseCommand {
throw new IllegalStateException("The transport layer should filter out LastPartialCommand instances but received: " + this);
}
/**
* Lets copy across the required fields from this last partial command to
* the newly unmarshalled complete command
*
* @param completeCommand the newly unmarshalled complete command
*/
public void configure(Command completeCommand) {
// overwrite the commandId as the numbers change when we introduce
// fragmentation commands
completeCommand.setCommandId(getCommandId());
// copy across the transient properties
completeCommand.setFrom(getFrom());
// TODO should not be required as the large command would be marshalled with this property
//completeCommand.setResponseRequired(isResponseRequired());
}
}

6
activemq-core/src/main/java/org/apache/activemq/command/PartialCommand.java
View File

@ -128,4 +128,10 @@ public class PartialCommand implements Command {
public boolean isMarshallAware() {
return false;
}
public String toString() {
return "PartialCommand[id: " + commandId + " data: " + data.length + " byte(s)]";
}
}

18
activemq-core/src/main/java/org/apache/activemq/transport/CommandJoiner.java
View File

@ -21,6 +21,8 @@ import org.apache.activemq.command.Command;
import org.apache.activemq.command.LastPartialCommand;
import org.apache.activemq.command.PartialCommand;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
@ -33,6 +35,7 @@ import java.io.IOException;
* @version $Revision$
*/
public class CommandJoiner extends TransportFilter {
private static final Log log = LogFactory.getLog(CommandJoiner.class);
private ByteArrayOutputStream out = new ByteArrayOutputStream();
private OpenWireFormat wireFormat;
@ -57,13 +60,16 @@ public class CommandJoiner extends TransportFilter {
else if (type == LastPartialCommand.DATA_STRUCTURE_TYPE) {
try {
byte[] fullData = out.toByteArray();
out.reset();
Command completeCommand = (Command) wireFormat.unmarshal(new DataInputStream(new ByteArrayInputStream(fullData)));
completeCommand.setCommandId(command.getCommandId());
completeCommand.setResponseRequired(command.isResponseRequired());
resetBuffer();
LastPartialCommand lastCommand = (LastPartialCommand) command;
lastCommand.configure(completeCommand);
getTransportListener().onCommand(completeCommand);
}
catch (IOException e) {
log.warn("Failed to unmarshal partial command: " + command);
getTransportListener().onException(e);
}
}
@ -74,14 +80,10 @@ public class CommandJoiner extends TransportFilter {
public void stop() throws Exception {
super.stop();
resetBuffer();
out = null;
}
public String toString() {
return next.toString();
}
protected void resetBuffer() {
out.reset();
}
}

14
activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java
View File

@ -32,14 +32,14 @@ import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
*
* @version $Revision: 1.4 $
*/
final public class ResponseCorrelator extends TransportFilter {
public class ResponseCorrelator extends TransportFilter {
private static final Log log = LogFactory.getLog(ResponseCorrelator.class);
private final ConcurrentHashMap requestMap = new ConcurrentHashMap();
private int lastCommandId = 0;
synchronized int getNextCommandId() {
public synchronized int getNextCommandId() {
return ++lastCommandId;
}
@ -48,13 +48,19 @@ final public class ResponseCorrelator extends TransportFilter {
}
public void oneway(Command command) throws IOException {
command.setCommandId(getNextCommandId());
// a parent transport could have set the ID
if (command.getCommandId() == 0) {
command.setCommandId(getNextCommandId());
}
command.setResponseRequired(false);
next.oneway(command);
}
public FutureResponse asyncRequest(Command command) throws IOException {
command.setCommandId(getNextCommandId());
// a parent transport could have set the ID
if (command.getCommandId() == 0) {
command.setCommandId(getNextCommandId());
}
command.setResponseRequired(true);
FutureResponse future = new FutureResponse();
requestMap.put(new Integer(command.getCommandId()), future);

6
activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java
View File

@ -88,15 +88,15 @@ public class WireFormatNegotiator extends TransportFilter {
try {
wireInfoSentDownLatch.await();
if (log.isDebugEnabled()) {
log.debug(this + " before negotiation: " + wireFormat);
}
if( !info.isValid() ) {
onException(new IOException("Remote wire format magic is invalid"));
} else if( info.getVersion() < minimumVersion ) {
onException(new IOException("Remote wire format ("+info.getVersion()+") is lower the minimum version required ("+minimumVersion+")"));
}
if (log.isDebugEnabled()) {
log.debug(this + " before negotiation: " + wireFormat);
}
wireFormat.renegociatWireFormat(info);
if (log.isDebugEnabled()) {

2
activemq-core/src/main/java/org/apache/activemq/transport/multicast/MulticastTransport.java
View File

@ -93,7 +93,7 @@ public class MulticastTransport extends UdpTransport {
socket.joinGroup(mcastAddress);
socket.setSoTimeout((int) keepAliveInterval);
return new CommandDatagramSocket(toString(), socket, getWireFormat(), getDatagramSize(), mcastAddress, mcastPort, createDatagramHeaderMarshaller());
return new CommandDatagramSocket(this, socket, getWireFormat(), getDatagramSize(), mcastAddress, mcastPort, createDatagramHeaderMarshaller());
}
protected InetSocketAddress createAddress(URI remoteLocation) throws UnknownHostException, IOException {

9
activemq-core/src/main/java/org/apache/activemq/transport/reliable/ExceptionIfDroppedReplayStrategy.java
View File

@ -28,13 +28,20 @@ public class ExceptionIfDroppedReplayStrategy implements ReplayStrategy {
private int maximumDifference = 5;
public ExceptionIfDroppedReplayStrategy() {
}
public ExceptionIfDroppedReplayStrategy(int maximumDifference) {
this.maximumDifference = maximumDifference;
}
public boolean onDroppedPackets(ReliableTransport transport, int expectedCounter, int actualCounter) throws IOException {
int difference = actualCounter - expectedCounter;
long count = Math.abs(difference);
if (count > maximumDifference) {
throw new IOException("Packets dropped on: " + transport + " count: " + count + " expected: " + expectedCounter + " but was: " + actualCounter);
}
// lets discard old commands
return difference > 0;
}

63
activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReliableTransport.java
View File

@ -17,9 +17,13 @@
package org.apache.activemq.transport.reliable;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.Response;
import org.apache.activemq.openwire.CommandIdComparator;
import org.apache.activemq.transport.FutureResponse;
import org.apache.activemq.transport.ResponseCorrelator;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFilter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.io.IOException;
import java.util.SortedSet;
@ -31,17 +35,54 @@ import java.util.TreeSet;
*
* @version $Revision$
*/
public class ReliableTransport extends TransportFilter {
public class ReliableTransport extends ResponseCorrelator {
private static final Log log = LogFactory.getLog(ReliableTransport.class);
private ReplayStrategy replayStrategy;
private SortedSet commands = new TreeSet(new CommandIdComparator());
private int expectedCounter = 1;
private int requestTimeout = 2000;
public ReliableTransport(Transport next, ReplayStrategy replayStrategy) {
super(next);
this.replayStrategy = replayStrategy;
}
public Response request(Command command) throws IOException {
FutureResponse response = asyncRequest(command);
while (true) {
Response result = response.getResult(requestTimeout);
if (result != null) {
return result;
}
replayRequest(command, response);
}
}
public Response request(Command command, int timeout) throws IOException {
FutureResponse response = asyncRequest(command);
while (timeout > 0) {
int time = timeout;
if (timeout > requestTimeout) {
time = requestTimeout;
}
Response result = response.getResult(time);
if (result != null) {
return result;
}
replayRequest(command, response);
timeout -= time;
}
return response.getResult(0);
}
public void onCommand(Command command) {
// lets pass wireformat through
if (command.isWireFormatInfo()) {
super.onCommand(command);
return;
}
int actualCounter = command.getCommandId();
boolean valid = expectedCounter == actualCounter;
@ -49,9 +90,12 @@ public class ReliableTransport extends TransportFilter {
synchronized (commands) {
try {
boolean keep = replayStrategy.onDroppedPackets(this, expectedCounter, actualCounter);
if (keep) {
// lets add it to the list for later on
if (log.isDebugEnabled()) {
log.debug("Received out of order command which is being buffered for later: " + command);
}
commands.add(command);
}
}
@ -75,7 +119,7 @@ public class ReliableTransport extends TransportFilter {
// we've got a valid header so increment counter
replayStrategy.onReceivedPacket(this, expectedCounter);
expectedCounter++;
getTransportListener().onCommand(command);
super.onCommand(command);
synchronized (commands) {
// we could have more commands left
@ -98,13 +142,14 @@ public class ReliableTransport extends TransportFilter {
return commands.size();
}
}
public int getExpectedCounter() {
return expectedCounter;
}
/**
* This property should never really be set - but is mutable primarily for test cases
* This property should never really be set - but is mutable primarily for
* test cases
*/
public void setExpectedCounter(int expectedCounter) {
this.expectedCounter = expectedCounter;
@ -114,4 +159,10 @@ public class ReliableTransport extends TransportFilter {
return next.toString();
}
/**
* Lets attempt to replay the request as a command may have disappeared
*/
protected void replayRequest(Command command, FutureResponse response) {
}
}

9
activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandChannel.java
View File

@ -16,11 +16,14 @@
*/
package org.apache.activemq.transport.udp;
import edu.emory.mathcs.backport.java.util.concurrent.Future;
import org.apache.activemq.Service;
import org.apache.activemq.command.Command;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.Map;
/**
*
@ -30,9 +33,7 @@ public interface CommandChannel extends Service {
public abstract Command read() throws IOException;
public abstract void write(Command command) throws IOException;
public abstract void write(Command command, SocketAddress address) throws IOException;
public abstract void write(Command command, SocketAddress address, Map requestMap, Future future) throws IOException;
public abstract int getDatagramSize();
@ -45,4 +46,6 @@ public interface CommandChannel extends Service {
public abstract void setHeaderMarshaller(DatagramHeaderMarshaller headerMarshaller);
public abstract void setTargetAddress(SocketAddress address);
}

88
activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandDatagramChannel.java
View File

@ -16,9 +16,10 @@
*/
package org.apache.activemq.transport.udp;
import edu.emory.mathcs.backport.java.util.concurrent.Future;
import org.activeio.ByteArrayInputStream;
import org.activeio.ByteArrayOutputStream;
import org.apache.activemq.Service;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.Endpoint;
import org.apache.activemq.command.LastPartialCommand;
@ -34,6 +35,7 @@ import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.Map;
/**
* A strategy for reading datagrams and de-fragmenting them together.
@ -44,6 +46,7 @@ public class CommandDatagramChannel implements CommandChannel {
private static final Log log = LogFactory.getLog(CommandDatagramChannel.class);
private final UdpTransport transport;
private final String name;
private DatagramChannel channel;
private OpenWireFormat wireFormat;
@ -61,15 +64,17 @@ public class CommandDatagramChannel implements CommandChannel {
private ByteBuffer writeBuffer;
private int defaultMarshalBufferSize = 64 * 1024;
public CommandDatagramChannel(String name, DatagramChannel channel, OpenWireFormat wireFormat, ByteBufferPool bufferPool, int datagramSize,
public CommandDatagramChannel(UdpTransport transport, DatagramChannel channel, OpenWireFormat wireFormat, ByteBufferPool bufferPool, int datagramSize,
SocketAddress targetAddress, DatagramHeaderMarshaller headerMarshaller) {
this.name = name;
this.transport = transport;
this.channel = channel;
this.wireFormat = wireFormat;
this.bufferPool = bufferPool;
this.datagramSize = datagramSize;
this.targetAddress = targetAddress;
this.headerMarshaller = headerMarshaller;
this.name = transport.toString();
}
public String toString() {
@ -87,34 +92,21 @@ public class CommandDatagramChannel implements CommandChannel {
bufferPool.stop();
}
/* (non-Javadoc)
* @see org.apache.activemq.transport.udp.CommandChannel#read()
*/
public Command read() throws IOException {
Command answer = null;
Endpoint from = null;
synchronized (readLock) {
while (true) {
readBuffer.clear();
SocketAddress address = channel.receive(readBuffer);
/*
if (address == null) {
System.out.println("No address on packet: " + readBuffer);
// continue;
}
*/
readBuffer.flip();
if (readBuffer.limit() == 0) {
//System.out.println("Empty packet!");
continue;
}
//log.debug("buffer: " + readBuffer + " has remaining: " + readBuffer.remaining());
Endpoint from = headerMarshaller.createEndpoint(readBuffer, address);
from = headerMarshaller.createEndpoint(readBuffer, address);
int remaining = readBuffer.remaining();
byte[] data = new byte[remaining];
readBuffer.get(data);
@ -125,33 +117,25 @@ public class CommandDatagramChannel implements CommandChannel {
// buffering?
DataInputStream dataIn = new DataInputStream(new ByteArrayInputStream(data));
answer = (Command) wireFormat.unmarshal(dataIn);
if (answer != null) {
answer.setFrom(from);
}
break;
}
}
if (answer != null) {
answer.setFrom(from);
if (log.isDebugEnabled()) {
log.debug("Channel: " + name + " about to process: " + answer);
log.debug("Channel: " + name + " received from: " + from + " about to process: " + answer);
}
}
return answer;
}
/* (non-Javadoc)
* @see org.apache.activemq.transport.udp.CommandChannel#write(org.apache.activemq.command.Command)
*/
public void write(Command command) throws IOException {
write(command, targetAddress);
}
/* (non-Javadoc)
* @see org.apache.activemq.transport.udp.CommandChannel#write(org.apache.activemq.command.Command, java.net.SocketAddress)
*/
public void write(Command command, SocketAddress address) throws IOException {
public void write(Command command, SocketAddress address, Map requestMap, Future future) throws IOException {
synchronized (writeLock) {
if (!command.isWireFormatInfo() && command.getCommandId() == 0) {
command.setCommandId(transport.getNextCommandId());
}
ByteArrayOutputStream largeBuffer = new ByteArrayOutputStream(defaultMarshalBufferSize);
wireFormat.marshal(command, new DataOutputStream(largeBuffer));
byte[] data = largeBuffer.toByteArray();
@ -160,7 +144,7 @@ public class CommandDatagramChannel implements CommandChannel {
writeBuffer.clear();
headerMarshaller.writeHeader(command, writeBuffer);
if (size >= writeBuffer.remaining()) {
if (size > writeBuffer.remaining()) {
// lets split the command up into chunks
int offset = 0;
boolean lastFragment = false;
@ -216,7 +200,11 @@ public class CommandDatagramChannel implements CommandChannel {
bs.marshal(writeBuffer);
}
writeBuffer.putInt(command.getCommandId());
int commandId = command.getCommandId();
if (fragment > 0) {
commandId = transport.getNextCommandId();
}
writeBuffer.putInt(commandId);
if (bs == null) {
writeBuffer.put((byte) 1);
}
@ -232,7 +220,9 @@ public class CommandDatagramChannel implements CommandChannel {
}
// now lets write the last partial command
command = new LastPartialCommand(command);
command = new LastPartialCommand(command.isResponseRequired());
command.setCommandId(transport.getNextCommandId());
largeBuffer = new ByteArrayOutputStream(defaultMarshalBufferSize);
wireFormat.marshal(command, new DataOutputStream(largeBuffer));
data = largeBuffer.toByteArray();
@ -243,6 +233,9 @@ public class CommandDatagramChannel implements CommandChannel {
writeBuffer.put(data);
if (command.isResponseRequired()) {
requestMap.put(new Integer(command.getCommandId()), future);
}
sendWriteBuffer(address);
}
}
@ -250,16 +243,10 @@ public class CommandDatagramChannel implements CommandChannel {
// Properties
// -------------------------------------------------------------------------
/* (non-Javadoc)
* @see org.apache.activemq.transport.udp.CommandChannel#getDatagramSize()
*/
public int getDatagramSize() {
return datagramSize;
}
/* (non-Javadoc)
* @see org.apache.activemq.transport.udp.CommandChannel#setDatagramSize(int)
*/
public void setDatagramSize(int datagramSize) {
this.datagramSize = datagramSize;
}
@ -275,20 +262,23 @@ public class CommandDatagramChannel implements CommandChannel {
this.bufferPool = bufferPool;
}
/* (non-Javadoc)
* @see org.apache.activemq.transport.udp.CommandChannel#getHeaderMarshaller()
*/
public DatagramHeaderMarshaller getHeaderMarshaller() {
return headerMarshaller;
}
/* (non-Javadoc)
* @see org.apache.activemq.transport.udp.CommandChannel#setHeaderMarshaller(org.apache.activemq.transport.udp.DatagramHeaderMarshaller)
*/
public void setHeaderMarshaller(DatagramHeaderMarshaller headerMarshaller) {
this.headerMarshaller = headerMarshaller;
}
public SocketAddress getTargetAddress() {
return targetAddress;
}
public void setTargetAddress(SocketAddress targetAddress) {
this.targetAddress = targetAddress;
}
// Implementation methods
// -------------------------------------------------------------------------
protected void sendWriteBuffer(SocketAddress address) throws IOException {

66
activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandDatagramSocket.java
View File

@ -16,6 +16,9 @@
*/
package org.apache.activemq.transport.udp;
import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
import edu.emory.mathcs.backport.java.util.concurrent.Future;
import org.activeio.ByteArrayInputStream;
import org.activeio.ByteArrayOutputStream;
import org.apache.activemq.command.Command;
@ -24,6 +27,8 @@ import org.apache.activemq.command.LastPartialCommand;
import org.apache.activemq.command.PartialCommand;
import org.apache.activemq.openwire.BooleanStream;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.transport.FutureResponse;
import org.apache.activemq.transport.ResponseCorrelator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -35,6 +40,7 @@ import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Map;
/**
* A strategy for reading datagrams and de-fragmenting them together.
@ -45,6 +51,7 @@ public class CommandDatagramSocket implements CommandChannel {
private static final Log log = LogFactory.getLog(CommandDatagramSocket.class);
private final UdpTransport transport;
private final String name;
private DatagramSocket channel;
private InetAddress targetAddress;
@ -59,16 +66,16 @@ public class CommandDatagramSocket implements CommandChannel {
// writing
private Object writeLock = new Object();
public CommandDatagramSocket(String name, DatagramSocket channel, OpenWireFormat wireFormat, int datagramSize, InetAddress targetAddress, int targetPort,
DatagramHeaderMarshaller headerMarshaller) {
this.name = name;
public CommandDatagramSocket(UdpTransport transport, DatagramSocket channel, OpenWireFormat wireFormat, int datagramSize, InetAddress targetAddress,
int targetPort, DatagramHeaderMarshaller headerMarshaller) {
this.transport = transport;
this.channel = channel;
this.wireFormat = wireFormat;
this.datagramSize = datagramSize;
this.targetAddress = targetAddress;
this.targetPort = targetPort;
this.headerMarshaller = headerMarshaller;
this.name = transport.toString();
}
public String toString() {
@ -100,7 +107,7 @@ public class CommandDatagramSocket implements CommandChannel {
}
if (answer != null) {
answer.setFrom(from);
if (log.isDebugEnabled()) {
log.debug("Channel: " + name + " about to process: " + answer);
}
@ -108,23 +115,15 @@ public class CommandDatagramSocket implements CommandChannel {
return answer;
}
public void write(Command command) throws IOException {
write(command, targetAddress, targetPort);
public void write(Command command, SocketAddress address, Map requestMap, Future future) throws IOException {
InetSocketAddress ia = (InetSocketAddress) address;
write(command, ia.getAddress(), ia.getPort(), requestMap, future);
}
public void write(Command command, SocketAddress address) throws IOException {
if (address instanceof InetSocketAddress) {
InetSocketAddress ia = (InetSocketAddress) address;
write(command, ia.getAddress(), ia.getPort());
}
else {
write(command);
}
}
public void write(Command command, InetAddress address, int port) throws IOException {
public void write(Command command, InetAddress address, int port, Map requestMap, Future future) throws IOException {
synchronized (writeLock) {
command.setCommandId(transport.getNextCommandId());
ByteArrayOutputStream writeBuffer = createByteArrayOutputStream();
DataOutputStream dataOut = new DataOutputStream(writeBuffer);
headerMarshaller.writeHeader(command, dataOut);
@ -134,6 +133,9 @@ public class CommandDatagramSocket implements CommandChannel {
wireFormat.marshal(command, dataOut);
if (remaining(writeBuffer) >= 0) {
if (command.isResponseRequired()) {
requestMap.put(new Integer(command.getCommandId()), future);
}
sendWriteBuffer(address, port, writeBuffer);
}
else {
@ -189,7 +191,11 @@ public class CommandDatagramSocket implements CommandChannel {
bs.marshal(dataOut);
}
dataOut.writeInt(command.getCommandId());
int commandId = command.getCommandId();
if (fragment > 0) {
commandId = transport.getNextCommandId();
}
dataOut.writeInt(commandId);
if (bs == null) {
dataOut.write((byte) 1);
}
@ -205,12 +211,16 @@ public class CommandDatagramSocket implements CommandChannel {
}
// now lets write the last partial command
command = new LastPartialCommand(command);
command = new LastPartialCommand(command.isResponseRequired());
command.setCommandId(transport.getNextCommandId());
writeBuffer.reset();
headerMarshaller.writeHeader(command, dataOut);
wireFormat.marshal(command, dataOut);
if (command.isResponseRequired()) {
requestMap.put(new Integer(command.getCommandId()), future);
}
sendWriteBuffer(address, port, writeBuffer);
}
}
@ -238,6 +248,22 @@ public class CommandDatagramSocket implements CommandChannel {
this.headerMarshaller = headerMarshaller;
}
public SocketAddress getTargetAddress() {
return new InetSocketAddress(targetAddress, targetPort);
}
public void setTargetAddress(SocketAddress address) {
if (address instanceof InetSocketAddress) {
InetSocketAddress ia = (InetSocketAddress) address;
targetAddress = ia.getAddress();
targetPort = ia.getPort();
}
else {
throw new IllegalArgumentException("Address must be instance of InetSocketAddress");
}
}
// Implementation methods
// -------------------------------------------------------------------------
protected void sendWriteBuffer(InetAddress address, int port, ByteArrayOutputStream writeBuffer) throws IOException {

1
activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramEndpoint.java
View File

@ -38,4 +38,5 @@ public class DatagramEndpoint extends BaseEndpoint {
return address;
}
}

118
activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
View File

@ -16,11 +16,17 @@
*/
package org.apache.activemq.transport.udp;
import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
import org.apache.activemq.Service;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.Endpoint;
import org.apache.activemq.command.Response;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.transport.FutureResponse;
import org.apache.activemq.transport.ResponseCorrelator;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFilter;
import org.apache.activemq.transport.TransportThreadSupport;
import org.apache.activemq.transport.reliable.ExceptionIfDroppedReplayStrategy;
import org.apache.activemq.transport.reliable.ReplayStrategy;
@ -28,6 +34,7 @@ import org.apache.activemq.util.ServiceStopper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.io.EOFException;
import java.io.IOException;
import java.net.DatagramSocket;
import java.net.InetAddress;
@ -62,6 +69,10 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
private int port;
private int minmumWireFormatVersion;
private String description = null;
private final ConcurrentHashMap requestMap = new ConcurrentHashMap();
private int lastCommandId = 0;
private Runnable runnable;
protected UdpTransport(OpenWireFormat wireFormat) throws IOException {
this.wireFormat = wireFormat;
@ -89,6 +100,28 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
this.description = getProtocolName() + "Server@";
}
public TransportFilter createFilter(Transport transport) {
return new TransportFilter(transport) {
public void onCommand(Command command) {
boolean debug = log.isDebugEnabled();
if (command.isResponse()) {
Response response = (Response) command;
FutureResponse future = (FutureResponse) requestMap.remove(new Integer(response.getCorrelationId()));
if (future != null) {
future.set(response);
}
else {
if (debug)
log.debug("Received unexpected response for command id: " + response.getCorrelationId());
}
}
else {
super.onCommand(command);
}
}
};
}
/**
* A one way asynchronous send
*/
@ -96,17 +129,53 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
oneway(command, targetAddress);
}
/**
* A one way asynchronous send
*/
public void oneway(Command command, FutureResponse future) throws IOException {
oneway(command, targetAddress, future);
}
protected void oneway(Command command, SocketAddress address, FutureResponse future) throws IOException {
if (log.isDebugEnabled()) {
log.debug("Sending oneway from: " + this + " to target: " + targetAddress + " command: " + command);
}
checkStarted(command);
commandChannel.write(command, address, requestMap, future);
}
/**
* A one way asynchronous send to a given address
*/
public void oneway(Command command, SocketAddress address) throws IOException {
if (log.isDebugEnabled()) {
log.debug("Sending oneway from: " + this + " to target: " + targetAddress);
}
checkStarted(command);
commandChannel.write(command, address);
oneway(command, address, null);
}
public FutureResponse asyncRequest(Command command) throws IOException {
if (command.getCommandId() == 0) {
command.setCommandId(getNextCommandId());
}
command.setResponseRequired(true);
FutureResponse future = new FutureResponse();
oneway(command, future);
return future;
}
public Response request(Command command) throws IOException {
FutureResponse response = asyncRequest(command);
return response.getResult();
}
public Response request(Command command, int timeout) throws IOException {
FutureResponse response = asyncRequest(command);
return response.getResult(timeout);
}
public void setStartupRunnable(Runnable runnable) {
this.runnable = runnable;
}
/**
* @return pretty print of 'this'
*/
@ -124,6 +193,9 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
*/
public void run() {
log.trace("Consumer thread starting for: " + toString());
if (runnable != null) {
runnable.run();
}
while (!isStopped()) {
try {
Command command = commandChannel.read();
@ -135,7 +207,7 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
stop();
}
catch (Exception e2) {
log.warn("Caught while closing: " + e2 + ". Now Closed", e2);
log.warn("Caught in: " + this + " while closing: " + e2 + ". Now Closed", e2);
}
}
catch (SocketException e) {
@ -145,23 +217,32 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
stop();
}
catch (Exception e2) {
log.warn("Caught while closing: " + e2 + ". Now Closed", e2);
log.warn("Caught in: " + this + " while closing: " + e2 + ". Now Closed", e2);
}
}
catch (Exception e) {
System.out.println("Caught exception of type: " + e.getClass());
e.printStackTrace();
catch (EOFException e) {
// DataInputStream closed
log.debug("Socket closed: " + e, e);
try {
stop();
}
catch (Exception e2) {
log.warn("Caught while closing: " + e2 + ". Now Closed", e2);
log.warn("Caught in: " + this + " while closing: " + e2 + ". Now Closed", e2);
}
}
catch (Exception e) {
try {
stop();
}
catch (Exception e2) {
log.warn("Caught in: " + this + " while closing: " + e2 + ". Now Closed", e2);
}
if (e instanceof IOException) {
onException((IOException) e);
}
else {
log.error(e);
log.error("Caught: " + e, e);
e.printStackTrace();
}
}
}
@ -182,6 +263,7 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
originalTargetAddress = targetAddress;
}
targetAddress = address;
commandChannel.setTargetAddress(address);
}
}
}
@ -324,12 +406,12 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
if (bufferPool == null) {
bufferPool = new DefaultBufferPool();
}
return new CommandDatagramChannel(toString(), channel, wireFormat, bufferPool, datagramSize, targetAddress, createDatagramHeaderMarshaller());
return new CommandDatagramChannel(this, channel, wireFormat, bufferPool, datagramSize, targetAddress, createDatagramHeaderMarshaller());
}
protected void bind(DatagramSocket socket, SocketAddress localAddress) throws IOException {
channel.configureBlocking(true);
if (log.isDebugEnabled()) {
log.debug("Binding to address: " + localAddress);
}
@ -340,7 +422,7 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
// TODO
// connect to default target address to avoid security checks each time
// channel = channel.connect(targetAddress);
return channel;
}
@ -370,7 +452,7 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
return targetAddress;
}
public void setCommandChannel(CommandChannel commandChannel) {
this.commandChannel = commandChannel;
protected synchronized int getNextCommandId() {
return ++lastCommandId;
}
}

40
activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java
View File

@ -17,6 +17,8 @@
package org.apache.activemq.transport.udp;
import org.activeio.command.WireFormat;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.Endpoint;
import org.apache.activemq.command.WireFormatInfo;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.transport.CommandJoiner;
@ -24,9 +26,13 @@ import org.apache.activemq.transport.InactivityMonitor;
import org.apache.activemq.transport.ResponseCorrelator;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.transport.TransportFilter;
import org.apache.activemq.transport.TransportLogger;
import org.apache.activemq.transport.TransportServer;
import org.apache.activemq.transport.WireFormatNegotiator;
import org.apache.activemq.transport.reliable.ExceptionIfDroppedReplayStrategy;
import org.apache.activemq.transport.reliable.ReliableTransport;
import org.apache.activemq.transport.reliable.ReplayStrategy;
import org.apache.activemq.util.IOExceptionSupport;
import org.apache.activemq.util.IntrospectionSupport;
import org.apache.activemq.util.URISupport;
@ -85,6 +91,9 @@ public class UdpTransportFactory extends TransportFactory {
if (udpTransport.getMaxInactivityDuration() > 0) {
transport = new InactivityMonitor(transport, udpTransport.getMaxInactivityDuration());
}
// TODO should we have this?
//transport = udpTransport.createFilter(transport);
return transport;
}
@ -96,6 +105,7 @@ public class UdpTransportFactory extends TransportFactory {
protected Transport configure(Transport transport, WireFormat format, Map options, boolean server) {
IntrospectionSupport.setProperties(transport, options);
UdpTransport udpTransport = (UdpTransport) transport;
OpenWireFormat openWireFormat = asOpenWireFormat(format);
if (udpTransport.isTrace()) {
transport = new TransportLogger(transport);
@ -109,24 +119,46 @@ public class UdpTransportFactory extends TransportFactory {
transport = new InactivityMonitor(transport, udpTransport.getMaxInactivityDuration());
}
transport = new ResponseCorrelator(transport);
// add reliabilty
//transport = new ReliableTransport(transport, createReplayStrategy());
// deal with fragmentation
transport = new CommandJoiner(transport, asOpenWireFormat(format));
transport = new CommandJoiner(transport, openWireFormat);
transport = udpTransport.createFilter(transport);
return transport;
}
protected ReplayStrategy createReplayStrategy() {
return new ExceptionIfDroppedReplayStrategy(1);
}
protected Transport configureClientSideNegotiator(Transport transport, WireFormat format, final UdpTransport udpTransport) {
return new TransportFilter(transport) {
public void onCommand(Command command) {
// redirect to the endpoint that the last response came from
Endpoint from = command.getFrom();
udpTransport.setTargetEndpoint(from);
super.onCommand(command);
}
};
/*
transport = new WireFormatNegotiator(transport, asOpenWireFormat(format), udpTransport.getMinmumWireFormatVersion()) {
protected void onWireFormatNegotiated(WireFormatInfo info) {
// lets switch to the target endpoint
// based on the last packet that was received
// so that all future requests go to the newly created UDP channel
udpTransport.setTargetEndpoint(info.getFrom());
Endpoint from = info.getFrom();
System.out.println("####Êsetting the client side target to: " + from);
udpTransport.setTargetEndpoint(from);
}
};
return transport;
*/
}
protected OpenWireFormat asOpenWireFormat(WireFormat wf) {

70
activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportServer.java
View File

@ -24,6 +24,7 @@ import org.apache.activemq.transport.CommandJoiner;
import org.apache.activemq.transport.InactivityMonitor;
import org.apache.activemq.transport.ResponseCorrelator;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFilter;
import org.apache.activemq.transport.TransportListener;
import org.apache.activemq.transport.TransportServer;
import org.apache.activemq.transport.TransportServerSupport;
@ -49,13 +50,13 @@ public class UdpTransportServer extends TransportServerSupport {
private UdpTransport serverTransport;
private Transport configuredTransport;
private boolean usingWireFormatNegotiation;
private Map transports = new HashMap();
public UdpTransportServer(URI connectURI, UdpTransport serverTransport, Transport configuredTransport) {
super(connectURI);
this.serverTransport = serverTransport;
this.configuredTransport = configuredTransport;
// lets disable the incremental checking of the sequence numbers
@ -86,6 +87,7 @@ public class UdpTransportServer extends TransportServerSupport {
}
public void onException(IOException error) {
log.error("Caught: " + error, error);
}
public void transportInterupted() {
@ -110,16 +112,22 @@ public class UdpTransportServer extends TransportServerSupport {
synchronized (transports) {
transport = (Transport) transports.get(endpoint);
if (transport == null) {
if (log.isDebugEnabled()) {
log.debug("Creating a new UDP server connection");
if (usingWireFormatNegotiation && !command.isWireFormatInfo()) {
log.error("Received inbound server communication from: " + command.getFrom() + " expecting WireFormatInfo but was command: " + command);
}
try {
transport = createTransport(command, endpoint);
transport = configureTransport(transport);
transports.put(endpoint, transport);
}
catch (IOException e) {
getAcceptListener().onAcceptError(e);
else {
if (log.isDebugEnabled()) {
log.debug("Creating a new UDP server connection");
}
try {
transport = createTransport(command, endpoint);
transport = configureTransport(transport);
transports.put(endpoint, transport);
}
catch (IOException e) {
log.error("Caught: " + e, e);
getAcceptListener().onAcceptError(e);
}
}
}
else {
@ -129,7 +137,7 @@ public class UdpTransportServer extends TransportServerSupport {
}
protected Transport configureTransport(Transport transport) {
transport = new ResponseCorrelator(transport);
// transport = new ResponseCorrelator(transport);
if (serverTransport.getMaxInactivityDuration() > 0) {
transport = new InactivityMonitor(transport, serverTransport.getMaxInactivityDuration());
@ -140,26 +148,54 @@ public class UdpTransportServer extends TransportServerSupport {
}
protected Transport createTransport(final Command command, DatagramEndpoint endpoint) throws IOException {
if (endpoint == null) {
//log.error("No endpoint available for command: " + command);
throw new IOException("No endpoint available for command: " + command);
}
final SocketAddress address = endpoint.getAddress();
final OpenWireFormat connectionWireFormat = serverTransport.getWireFormat().copy();
final UdpTransport transport = new UdpTransport(connectionWireFormat, address);
Transport configuredTransport = new CommandJoiner(transport, connectionWireFormat);
return new WireFormatNegotiator(configuredTransport, transport.getWireFormat(), serverTransport.getMinmumWireFormatVersion()) {
// lets pass in the received transport
return new TransportFilter(configuredTransport) {
public void start() throws Exception {
super.start();
onCommand(command);
}
};
/**
// return configuredTransport;
// configuredTransport = transport.createFilter(configuredTransport);
final WireFormatNegotiator wireFormatNegotiator = new WireFormatNegotiator(configuredTransport, transport.getWireFormat(), serverTransport
.getMinmumWireFormatVersion()) {
public void start() throws Exception {
super.start();
// process the inbound wireformat
System.out.println("Starting a new server transport: " + this + " with command: " + command);
onCommand(command);
}
// lets use the specific addressing of wire format
protected void sendWireFormat(WireFormatInfo info) throws IOException {
System.out.println("#### we have negotiated the wireformat; sending a wireformat to: " + address);
transport.oneway(info, address);
}
};
return wireFormatNegotiator;
*/
/*
* transport.setStartupRunnable(new Runnable() {
*
* public void run() { System.out.println("Passing the incoming
* WireFormat into into: " + this);
* // process the inbound wireformat
* wireFormatNegotiator.onCommand(command); }});
*/
}
}

2
activemq-core/src/test/java/org/apache/activemq/test/JmsSendReceiveTestSupport.java
View File

@ -165,7 +165,7 @@ public class JmsSendReceiveTestSupport extends TestSupport implements MessageLis
* Waits for the messages to be delivered or when the wait time has been reached.
*/
protected void waitForMessagesToBeDelivered() {
long maxWaitTime = 30000;
long maxWaitTime = 60000;
long waitTime = maxWaitTime;
long start = (maxWaitTime <= 0) ? 0 : System.currentTimeMillis();

4
activemq-core/src/test/java/org/apache/activemq/test/JmsTopicSendReceiveWithTwoConnectionsTest.java
View File

@ -44,10 +44,14 @@ public class JmsTopicSendReceiveWithTwoConnectionsTest extends JmsSendReceiveTes
connectionFactory = createConnectionFactory();
log.info("Creating send connection");
sendConnection = createSendConnection();
log.info("Starting send connection");
sendConnection.start();
log.info("Creating receive connection");
receiveConnection = createReceiveConnection();
log.info("Starting receive connection");
receiveConnection.start();
log.info("Created sendConnection: " + sendConnection);

31
activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpSendReceiveWithTwoConnectionsAndLargeMessagesTest.java Normal file
View File

@ -0,0 +1,31 @@
/**
*
* Copyright 2005-2006 The Apache Software Foundation
*
* 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.apache.activemq.transport.udp;
/**
*
* @version $Revision$
*/
public class UdpSendReceiveWithTwoConnectionsAndLargeMessagesTest extends UdpSendReceiveWithTwoConnectionsTest {
protected void setUp() throws Exception {
largeMessages = true;
//messageCount = 1;
super.setUp();
}
}

1
activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpSendReceiveWithTwoConnectionsTest.java
View File

@ -29,7 +29,6 @@ public class UdpSendReceiveWithTwoConnectionsTest extends JmsTopicSendReceiveWit
protected BrokerService broker;
protected void setUp() throws Exception {
largeMessages = true;
broker = createBroker();
broker.start();

19
activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTestSupport.java
View File

@ -21,6 +21,7 @@ import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.command.WireFormatInfo;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportAcceptListener;
@ -189,6 +190,11 @@ public abstract class UdpTestSupport extends TestCase implements TransportListen
System.out.println("Got WireFormatInfo: " + command);
}
else {
if (command.isResponseRequired()) {
// lets send a response back...
sendResponse(command);
}
if (large) {
System.out.println("### Received command: " + command.getClass() + " with id: " + command.getCommandId());
}
@ -203,6 +209,19 @@ public abstract class UdpTestSupport extends TestCase implements TransportListen
}
}
protected void sendResponse(Command command) {
Response response = new Response();
response.setCorrelationId(command.getCommandId());
try {
consumer.oneway(response);
}
catch (IOException e) {
System.out.println("Caught: " + e);
e.printStackTrace();
throw new RuntimeException(e);
}
}
public void onException(IOException error) {
System.out.println("### Received error: " + error);
}

16
activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTransportUsingServerTest.java
View File

@ -16,6 +16,8 @@
*/
package org.apache.activemq.transport.udp;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.transport.TransportServer;
@ -32,6 +34,18 @@ public class UdpTransportUsingServerTest extends UdpTestSupport {
protected String producerURI = "udp://localhost:" + consumerPort;
protected String serverURI = producerURI;
public void testRequestResponse() throws Exception {
ConsumerInfo expected = new ConsumerInfo();
expected.setSelector("Edam");
expected.setResponseRequired(true);
System.out.println("About to send: " + expected);
Response response = producer.request(expected, 2000);
System.out.println("Received: " + response);
assertNotNull("Received a response", response);
assertTrue("Should not be an exception", !response.isException());
}
protected Transport createProducer() throws Exception {
System.out.println("Producer using URI: " + producerURI);
URI uri = new URI(producerURI);
@ -41,7 +55,7 @@ public class UdpTransportUsingServerTest extends UdpTestSupport {
protected TransportServer createServer() throws Exception {
return TransportFactory.bind("byBroker", new URI(serverURI));
}
protected Transport createConsumer() throws Exception {
return null;
}