refactor of the UDP transport to enable the ReliableTransport by default

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@385826 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java

@@ -328,31 +328,37 @@ public class BrokerService implements Service {
             return;
         }
         
-        processHelperProperties();
+        try {
+            processHelperProperties();
 
-        BrokerRegistry.getInstance().bind(getBrokerName(), this);
+            BrokerRegistry.getInstance().bind(getBrokerName(), this);
 
-        addShutdownHook();
+            addShutdownHook();
-        if (deleteAllMessagesOnStartup) {
+            if (deleteAllMessagesOnStartup) {
-            deleteAllMessages();
+                deleteAllMessages();
-        }
-
-        if (isUseJmx()) {
-            getManagementContext().start();
-        }
-
-        getBroker().start();
-        if (masterConnectorURI!=null){
-            initializeMasterConnector(new URI(masterConnectorURI));
-            if (masterConnector!=null){
-                masterConnector.start();
             }
-        }
-        
-        startAllConnectors();
-        
 
-        log.info("ActiveMQ JMS Message Broker (" + getBrokerName() + ") started");
+            if (isUseJmx()) {
+                getManagementContext().start();
+            }
+
+            getBroker().start();
+            if (masterConnectorURI!=null){
+                initializeMasterConnector(new URI(masterConnectorURI));
+                if (masterConnector!=null){
+                    masterConnector.start();
+                }
+            }
+            
+            startAllConnectors();
+            
+
+            log.info("ActiveMQ JMS Message Broker (" + getBrokerName() + ") started");
+        }
+        catch (Exception e) {
+            log.error("Failed to start ActiveMQ JMS Message Broker. Reason: " + e, e);
+            throw e;
+        }
     }
 
     public void stop() throws Exception {

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

@@ -50,10 +50,6 @@ public class LastPartialCommand extends BaseCommand {
      * @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());
 

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

@@ -20,6 +20,7 @@ import java.io.IOException;
 
 import org.apache.activemq.command.Command;
 import org.apache.activemq.command.Response;
+import org.apache.activemq.util.IntSequenceGenerator;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -37,30 +38,25 @@ public class ResponseCorrelator extends TransportFilter {
     private static final Log log = LogFactory.getLog(ResponseCorrelator.class);
     
     private final ConcurrentHashMap requestMap = new ConcurrentHashMap();
-    private int lastCommandId = 0;
+    private IntSequenceGenerator sequenceGenerator;
-
-    public synchronized int getNextCommandId() {
-        return ++lastCommandId;
-    }
     
     public ResponseCorrelator(Transport next) {
-        super(next);
+        this(next, new IntSequenceGenerator());
     }
     
+    public ResponseCorrelator(Transport next, IntSequenceGenerator sequenceGenerator) {
+        super(next);
+        this.sequenceGenerator = sequenceGenerator;
+    }
+
     public void oneway(Command command) throws IOException {
-        // a parent transport could have set the ID
+        command.setCommandId(sequenceGenerator.getNextSequenceId());
-        if (command.getCommandId() == 0) {
-            command.setCommandId(getNextCommandId());
-        }
         command.setResponseRequired(false);
         next.oneway(command);
     }
 
     public FutureResponse asyncRequest(Command command) throws IOException {
-        // a parent transport could have set the ID
+        command.setCommandId(sequenceGenerator.getNextSequenceId());
-        if (command.getCommandId() == 0) {
-            command.setCommandId(getNextCommandId());
-        }
         command.setResponseRequired(true);
         FutureResponse future = new FutureResponse();
         requestMap.put(new Integer(command.getCommandId()), future);
@@ -93,6 +89,10 @@ public class ResponseCorrelator extends TransportFilter {
         }
     }
     
+    public IntSequenceGenerator getSequenceGenerator() {
+        return sequenceGenerator;
+    }
+
     public String toString() {
         return next.toString();
     }

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

@@ -22,6 +22,7 @@ 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.util.IntSequenceGenerator;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -48,6 +49,12 @@ public class ReliableTransport extends ResponseCorrelator {
         this.replayStrategy = replayStrategy;
     }
 
+    public ReliableTransport(Transport next, IntSequenceGenerator sequenceGenerator, ReplayStrategy replayStrategy) {
+        super(next, sequenceGenerator);
+        this.replayStrategy = replayStrategy;
+    }
+
+
     public Response request(Command command) throws IOException {
         FutureResponse response = asyncRequest(command);
         while (true) {
@@ -155,14 +162,31 @@ public class ReliableTransport extends ResponseCorrelator {
         this.expectedCounter = expectedCounter;
     }
 
-    public String toString() {
+    
-        return next.toString();
+    public int getRequestTimeout() {
+        return requestTimeout;
+    }
+
+    /**
+     * Sets the default timeout of requests before starting to request commands are replayed
+     */
+    public void setRequestTimeout(int requestTimeout) {
+        this.requestTimeout = requestTimeout;
     }
 
 
+    public ReplayStrategy getReplayStrategy() {
+        return replayStrategy;
+    }
+
+
+    public String toString() {
+        return next.toString();
+    }
     /**
      * Lets attempt to replay the request as a command may have disappeared
      */
     protected void replayRequest(Command command, FutureResponse response) {
+        log.debug("Still waiting for response on: " + this + " to command: " + command);
     }
 }

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

@@ -16,14 +16,11 @@
  */
 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;
 
 /**
  *
@@ -33,7 +30,7 @@ public interface CommandChannel extends Service {
 
     public abstract Command read() throws IOException;
 
-    public abstract void write(Command command, SocketAddress address, Map requestMap, Future future) throws IOException;
+    public abstract void write(Command command, SocketAddress address) throws IOException;
 
     public abstract int getDatagramSize();
 

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

@@ -16,8 +16,6 @@
  */
 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.command.Command;
@@ -26,6 +24,7 @@ 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.util.IntSequenceGenerator;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -35,7 +34,6 @@ 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.
@@ -46,8 +44,8 @@ public class CommandDatagramChannel implements CommandChannel {
 
     private static final Log log = LogFactory.getLog(CommandDatagramChannel.class);
 
-    private final UdpTransport transport;
     private final String name;
+    private final IntSequenceGenerator sequenceGenerator;
     private DatagramChannel channel;
     private OpenWireFormat wireFormat;
     private ByteBufferPool bufferPool;
@@ -65,9 +63,9 @@ public class CommandDatagramChannel implements CommandChannel {
     private int defaultMarshalBufferSize = 64 * 1024;
 
 
+
     public CommandDatagramChannel(UdpTransport transport, DatagramChannel channel, OpenWireFormat wireFormat, ByteBufferPool bufferPool, int datagramSize,
             SocketAddress targetAddress, DatagramHeaderMarshaller headerMarshaller) {
-        this.transport = transport;
         this.channel = channel;
         this.wireFormat = wireFormat;
         this.bufferPool = bufferPool;
@@ -75,6 +73,10 @@ public class CommandDatagramChannel implements CommandChannel {
         this.targetAddress = targetAddress;
         this.headerMarshaller = headerMarshaller;
         this.name = transport.toString();
+        this.sequenceGenerator = transport.getSequenceGenerator();
+        if (sequenceGenerator == null) {
+            throw new IllegalArgumentException("No sequenceGenerator on the given transport: " + transport);
+        }
     }
 
     public String toString() {
@@ -130,12 +132,9 @@ public class CommandDatagramChannel implements CommandChannel {
         return answer;
     }
 
-    public void write(Command command, SocketAddress address, Map requestMap, Future future) throws IOException {
+    public void write(Command command, SocketAddress address) 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();
@@ -202,7 +201,7 @@ public class CommandDatagramChannel implements CommandChannel {
 
                     int commandId = command.getCommandId();
                     if (fragment > 0) {
-                        commandId = transport.getNextCommandId();
+                        commandId = sequenceGenerator.getNextSequenceId();
                     }
                     writeBuffer.putInt(commandId);
                     if (bs == null) {
@@ -216,12 +215,12 @@ public class CommandDatagramChannel implements CommandChannel {
                     writeBuffer.put(data, offset, chunkSize);
 
                     offset += chunkSize;
-                    sendWriteBuffer(address);
+                    sendWriteBuffer(address, commandId);
                 }
 
                 // now lets write the last partial command
                 command = new LastPartialCommand(command.isResponseRequired());
-                command.setCommandId(transport.getNextCommandId());
+                command.setCommandId(sequenceGenerator.getNextSequenceId());
                 
                 largeBuffer = new ByteArrayOutputStream(defaultMarshalBufferSize);
                 wireFormat.marshal(command, new DataOutputStream(largeBuffer));
@@ -232,11 +231,7 @@ public class CommandDatagramChannel implements CommandChannel {
             }
 
             writeBuffer.put(data);
-
+            sendWriteBuffer(address, command.getCommandId());
-            if (command.isResponseRequired()) {
-                requestMap.put(new Integer(command.getCommandId()), future);
-            }
-            sendWriteBuffer(address);
         }
     }
 
@@ -281,11 +276,11 @@ public class CommandDatagramChannel implements CommandChannel {
 
     // Implementation methods
     // -------------------------------------------------------------------------
-    protected void sendWriteBuffer(SocketAddress address) throws IOException {
+    protected void sendWriteBuffer(SocketAddress address, int commandId) throws IOException {
         writeBuffer.flip();
 
         if (log.isDebugEnabled()) {
-            log.debug("Channel: " + name + " sending datagram to: " + address);
+            log.debug("Channel: " + name + " sending datagram: " + commandId + " to: " + address);
         }
         channel.send(writeBuffer, address);
     }

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

@@ -16,9 +16,6 @@
  */
 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;
@@ -27,8 +24,7 @@ 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.util.IntSequenceGenerator;
-import org.apache.activemq.transport.ResponseCorrelator;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -40,7 +36,6 @@ 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.
@@ -51,7 +46,6 @@ 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 +53,12 @@ public class CommandDatagramSocket implements CommandChannel {
     private OpenWireFormat wireFormat;
     private int datagramSize = 4 * 1024;
     private DatagramHeaderMarshaller headerMarshaller;
-
+    private IntSequenceGenerator sequenceGenerator;
-    // reading
     private Object readLock = new Object();
-
-    // writing
     private Object writeLock = new Object();
 
     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;
@@ -76,6 +66,10 @@ public class CommandDatagramSocket implements CommandChannel {
         this.targetPort = targetPort;
         this.headerMarshaller = headerMarshaller;
         this.name = transport.toString();
+        this.sequenceGenerator = transport.getSequenceGenerator();
+        if (sequenceGenerator == null) {
+            throw new IllegalArgumentException("No sequenceGenerator on the given transport: " + transport);
+        }
     }
 
     public String toString() {
@@ -115,15 +109,14 @@ public class CommandDatagramSocket implements CommandChannel {
         return answer;
     }
 
-    public void write(Command command, SocketAddress address, Map requestMap, Future future) throws IOException {
+    public void write(Command command, SocketAddress address) throws IOException {
         InetSocketAddress ia = (InetSocketAddress) address;
-        write(command, ia.getAddress(), ia.getPort(), requestMap, future);
+        write(command, ia.getAddress(), ia.getPort());
     }
 
-    public void write(Command command, InetAddress address, int port, Map requestMap, Future future) throws IOException {
+    public void write(Command command, InetAddress address, int port) throws IOException {
         synchronized (writeLock) {
 
-            command.setCommandId(transport.getNextCommandId());
             ByteArrayOutputStream writeBuffer = createByteArrayOutputStream();
             DataOutputStream dataOut = new DataOutputStream(writeBuffer);
             headerMarshaller.writeHeader(command, dataOut);
@@ -133,10 +126,7 @@ public class CommandDatagramSocket implements CommandChannel {
             wireFormat.marshal(command, dataOut);
 
             if (remaining(writeBuffer) >= 0) {
-                if (command.isResponseRequired()) {
+                sendWriteBuffer(address, port, writeBuffer, command.getCommandId());
-                    requestMap.put(new Integer(command.getCommandId()), future);
-                }
-                sendWriteBuffer(address, port, writeBuffer);
             }
             else {
                 // lets split the command up into chunks
@@ -193,7 +183,7 @@ public class CommandDatagramSocket implements CommandChannel {
 
                     int commandId = command.getCommandId();
                     if (fragment > 0) {
-                        commandId = transport.getNextCommandId();
+                        commandId = sequenceGenerator.getNextSequenceId();
                     }
                     dataOut.writeInt(commandId);
                     if (bs == null) {
@@ -207,21 +197,18 @@ public class CommandDatagramSocket implements CommandChannel {
                     dataOut.write(data, offset, chunkSize);
 
                     offset += chunkSize;
-                    sendWriteBuffer(address, port, writeBuffer);
+                    sendWriteBuffer(address, port, writeBuffer, commandId);
                 }
 
                 // now lets write the last partial command
                 command = new LastPartialCommand(command.isResponseRequired());
-                command.setCommandId(transport.getNextCommandId());
+                command.setCommandId(sequenceGenerator.getNextSequenceId());
 
                 writeBuffer.reset();
                 headerMarshaller.writeHeader(command, dataOut);
                 wireFormat.marshal(command, dataOut);
 
-                if (command.isResponseRequired()) {
+                sendWriteBuffer(address, port, writeBuffer, command.getCommandId());
-                    requestMap.put(new Integer(command.getCommandId()), future);
-                }
-                sendWriteBuffer(address, port, writeBuffer);
             }
         }
     }
@@ -266,9 +253,9 @@ public class CommandDatagramSocket implements CommandChannel {
 
     // Implementation methods
     // -------------------------------------------------------------------------
-    protected void sendWriteBuffer(InetAddress address, int port, ByteArrayOutputStream writeBuffer) throws IOException {
+    protected void sendWriteBuffer(InetAddress address, int port, ByteArrayOutputStream writeBuffer, int commandId) throws IOException {
         if (log.isDebugEnabled()) {
-            log.debug("Channel: " + name + " sending datagram to: " + address);
+            log.debug("Channel: " + name + " sending datagram: " + commandId + " to: " + address);
         }
         byte[] data = writeBuffer.toByteArray();
         DatagramPacket packet = new DatagramPacket(data, 0, data.length, address, port);

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

@@ -18,7 +18,6 @@ package org.apache.activemq.transport.udp;
 
 import org.apache.activemq.command.BaseEndpoint;
 
-import java.net.InetAddress;
 import java.net.SocketAddress;
 
 /**

activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramHeaderMarshaller.java

@@ -17,14 +17,12 @@
 package org.apache.activemq.transport.udp;
 
 
-import org.activeio.ByteArrayOutputStream;
 import org.apache.activemq.command.Command;
 import org.apache.activemq.command.Endpoint;
 
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.net.DatagramPacket;
-import java.net.InetAddress;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 

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

@@ -16,20 +16,15 @@
  */
 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;
+import org.apache.activemq.util.IntSequenceGenerator;
 import org.apache.activemq.util.ServiceStopper;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -69,10 +64,8 @@ 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;
+    private IntSequenceGenerator sequenceGenerator;
 
     protected UdpTransport(OpenWireFormat wireFormat) throws IOException {
         this.wireFormat = wireFormat;
@@ -100,28 +93,6 @@ 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
      */
@@ -130,48 +101,16 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
     }
 
     /**
-     * A one way asynchronous send
+     * A one way asynchronous send to a given address
      */
-    public void oneway(Command command, FutureResponse future) throws IOException {
+    public void oneway(Command command, SocketAddress address) 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);
+        commandChannel.write(command, address);
     }
 
-    /**
-     * A one way asynchronous send to a given address
-     */
-    public void oneway(Command command, SocketAddress address) throws IOException {
-        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;
     }
@@ -363,6 +302,15 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
         this.checkSequenceNumbers = checkSequenceNumbers;
     }
 
+
+    public IntSequenceGenerator getSequenceGenerator() {
+        return sequenceGenerator;
+    }
+
+    public void setSequenceGenerator(IntSequenceGenerator sequenceGenerator) {
+        this.sequenceGenerator = sequenceGenerator;
+    }
+    
     // Implementation methods
     // -------------------------------------------------------------------------
 
@@ -451,8 +399,4 @@ public class UdpTransport extends TransportThreadSupport implements Transport, S
     protected SocketAddress getTargetAddress() {
         return targetAddress;
     }
-
-    protected synchronized int getNextCommandId() {
-        return ++lastCommandId;
-    }
 }

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

@@ -19,21 +19,19 @@ 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;
 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.IntSequenceGenerator;
 import org.apache.activemq.util.IntrospectionSupport;
 import org.apache.activemq.util.URISupport;
 
@@ -58,7 +56,15 @@ public class UdpTransportFactory extends TransportFactory {
             UdpTransport transport = new UdpTransport(openWireFormat, port);
 
             Transport configuredTransport = configure(transport, wf, options, true);
-            UdpTransportServer server = new UdpTransportServer(location, transport, configuredTransport);
+            ReplayStrategy replayStrategy = null;
+            if (configuredTransport instanceof ReliableTransport) {
+                ReliableTransport rt = (ReliableTransport) configuredTransport;
+                replayStrategy = rt.getReplayStrategy();
+            }
+            if (replayStrategy == null) {
+                replayStrategy = createReplayStrategy();
+            }
+            UdpTransportServer server = new UdpTransportServer(location, transport, configuredTransport, replayStrategy);
             return server;
         }
         catch (URISyntaxException e) {
@@ -76,7 +82,7 @@ public class UdpTransportFactory extends TransportFactory {
     public Transport compositeConfigure(Transport transport, WireFormat format, Map options) {
         IntrospectionSupport.setProperties(transport, options);
         final UdpTransport udpTransport = (UdpTransport) transport;
-        
+
         // deal with fragmentation
         transport = new CommandJoiner(transport, asOpenWireFormat(format));
 
@@ -91,9 +97,7 @@ 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;
     }
 
@@ -119,15 +123,23 @@ public class UdpTransportFactory extends TransportFactory {
             transport = new InactivityMonitor(transport, udpTransport.getMaxInactivityDuration());
         }
 
-        // add reliabilty
-        //transport = new ReliableTransport(transport, createReplayStrategy());
-        
         // deal with fragmentation
-        transport = new CommandJoiner(transport, openWireFormat);
+
-        
+        if (server) {
-        transport = udpTransport.createFilter(transport);
+            // we don't want to do reliable checks on this transport as we
-        
+            // delegate to one that does
-        return transport;
+            transport = new CommandJoiner(transport, openWireFormat);
+            udpTransport.setSequenceGenerator(new IntSequenceGenerator());
+            return transport;
+        }
+        else {
+            ReliableTransport reliableTransport = new ReliableTransport(transport, createReplayStrategy());
+            udpTransport.setSequenceGenerator(reliableTransport.getSequenceGenerator());
+
+            // Joiner must be on outside as the inbound messages must be
+            // processed by the reliable transport first
+            return new CommandJoiner(reliableTransport, openWireFormat);
+        }
     }
 
     protected ReplayStrategy createReplayStrategy() {
@@ -141,24 +153,21 @@ public class UdpTransportFactory extends TransportFactory {
                 // 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()) {
+         * transport = new WireFormatNegotiator(transport,
-            protected void onWireFormatNegotiated(WireFormatInfo info) {
+         * asOpenWireFormat(format), udpTransport.getMinmumWireFormatVersion()) {
-                // lets switch to the target endpoint
+         * protected void onWireFormatNegotiated(WireFormatInfo info) { // lets
-                // based on the last packet that was received
+         * switch to the target endpoint // based on the last packet that was
-                // so that all future requests go to the newly created UDP channel
+         * received // so that all future requests go to the newly created UDP
-                Endpoint from = info.getFrom();
+         * channel Endpoint from = info.getFrom();
-                System.out.println("####Êsetting the client side target to: " + from);
+         * System.out.println("####Êsetting the client side target to: " +
-                udpTransport.setTargetEndpoint(from);
+         * from); udpTransport.setTargetEndpoint(from); } }; return transport;
-            }
+         */
-        };
-        return transport;
-        */
     }
 
     protected OpenWireFormat asOpenWireFormat(WireFormat wf) {

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

@@ -18,17 +18,15 @@ package org.apache.activemq.transport.udp;
 
 import org.apache.activemq.command.BrokerInfo;
 import org.apache.activemq.command.Command;
-import org.apache.activemq.command.WireFormatInfo;
 import org.apache.activemq.openwire.OpenWireFormat;
 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;
-import org.apache.activemq.transport.WireFormatNegotiator;
+import org.apache.activemq.transport.reliable.ReliableTransport;
+import org.apache.activemq.transport.reliable.ReplayStrategy;
 import org.apache.activemq.util.ServiceStopper;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -49,15 +47,17 @@ public class UdpTransportServer extends TransportServerSupport {
     private static final Log log = LogFactory.getLog(UdpTransportServer.class);
 
     private UdpTransport serverTransport;
+    private ReplayStrategy replayStrategy;
     private Transport configuredTransport;
     private boolean usingWireFormatNegotiation;
     private Map transports = new HashMap();
 
-    public UdpTransportServer(URI connectURI, UdpTransport serverTransport, Transport configuredTransport) {
+
+    public UdpTransportServer(URI connectURI, UdpTransport serverTransport, Transport configuredTransport, ReplayStrategy replayStrategy) {
         super(connectURI);
         this.serverTransport = serverTransport;
-
         this.configuredTransport = configuredTransport;
+        this.replayStrategy = replayStrategy;
 
         // lets disable the incremental checking of the sequence numbers
         // as we are getting messages from many different clients
@@ -137,8 +137,6 @@ public class UdpTransportServer extends TransportServerSupport {
     }
 
     protected Transport configureTransport(Transport transport) {
-        // transport = new ResponseCorrelator(transport);
-
         if (serverTransport.getMaxInactivityDuration() > 0) {
             transport = new InactivityMonitor(transport, serverTransport.getMaxInactivityDuration());
         }
@@ -149,28 +147,26 @@ 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);
+        final ReliableTransport reliableTransport = new ReliableTransport(transport, replayStrategy);
-
+        transport.setSequenceGenerator(reliableTransport.getSequenceGenerator());
-        // lets pass in the received transport
+        
-        return new TransportFilter(configuredTransport) {
+        // Joiner must be on outside as the inbound messages must be processed by the reliable transport first
+        return new CommandJoiner(reliableTransport, connectionWireFormat) {
             public void start() throws Exception {
                 super.start();
-                onCommand(command);
+                reliableTransport.onCommand(command);
             }
         };
 
+
+        
         /**
-        // return configuredTransport;
-
-        // configuredTransport = transport.createFilter(configuredTransport);
-
         final WireFormatNegotiator wireFormatNegotiator = new WireFormatNegotiator(configuredTransport, transport.getWireFormat(), serverTransport
                 .getMinmumWireFormatVersion()) {
 
@@ -188,14 +184,5 @@ public class UdpTransportServer extends TransportServerSupport {
         };
         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); }});
-         */
     }
 }

activemq-core/src/main/java/org/apache/activemq/util/IntSequenceGenerator.java

@@ -0,0 +1,34 @@
+/**
+ *
+ * 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.util;
+
+public class IntSequenceGenerator {
+
+    private int lastSequenceId;
+
+    public synchronized int getNextSequenceId() {
+        return ++lastSequenceId;
+    }
+
+    public synchronized int getLastSequenceId() {
+        return lastSequenceId;
+    }
+
+    public synchronized void setLastSequenceId(int l) {
+        lastSequenceId = l;
+    }
+}

activemq-core/src/test/java/org/apache/activemq/transport/multicast/MulticastTransportTest.java

@@ -21,6 +21,7 @@ import org.apache.activemq.transport.CommandJoiner;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.udp.UdpTransport;
 import org.apache.activemq.transport.udp.UdpTransportTest;
+import org.apache.activemq.util.IntSequenceGenerator;
 
 import java.net.URI;
 
@@ -42,12 +43,14 @@ public class MulticastTransportTest extends UdpTransportTest {
         
         OpenWireFormat wireFormat = createWireFormat();
         MulticastTransport transport = new MulticastTransport(wireFormat, new URI(multicastURI));
+        transport.setSequenceGenerator(new IntSequenceGenerator());
         return new CommandJoiner(transport, wireFormat);
     }
 
     protected Transport createConsumer() throws Exception {
         OpenWireFormat wireFormat = createWireFormat();
         MulticastTransport transport = new MulticastTransport(wireFormat, new URI(multicastURI));
+        transport.setSequenceGenerator(new IntSequenceGenerator());
         return new CommandJoiner(transport, wireFormat);
     }
 

activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpSendReceiveWithTwoConnectionsAndLargeMessagesTest.java

@@ -24,7 +24,7 @@ public class UdpSendReceiveWithTwoConnectionsAndLargeMessagesTest extends UdpSen
 
     protected void setUp() throws Exception {
         largeMessages = true;
-        //messageCount = 1;
+        //messageCount = 2;
         super.setUp();
     }
 

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

@@ -45,6 +45,7 @@ public class UdpSendReceiveWithTwoConnectionsTest extends JmsTopicSendReceiveWit
     protected BrokerService createBroker() throws Exception {
         BrokerService answer = new BrokerService();
         answer.setPersistent(false);
+        answer.setUseJmx(false);
         answer.addConnector(brokerURI);
         return answer;
     }

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

@@ -54,7 +54,6 @@ public abstract class UdpTestSupport extends TestCase implements TransportListen
         ConsumerInfo expected = new ConsumerInfo();
         expected.setSelector("Cheese");
         expected.setExclusive(true);
-        expected.setCommandId((short) 12);
         expected.setExclusive(true);
         expected.setPrefetchSize(3456);
 
@@ -67,7 +66,6 @@ public abstract class UdpTestSupport extends TestCase implements TransportListen
             ConsumerInfo actual = (ConsumerInfo) received;
             assertEquals("Selector", expected.getSelector(), actual.getSelector());
             assertEquals("isExclusive", expected.isExclusive(), actual.isExclusive());
-            assertEquals("getCommandId", expected.getCommandId(), actual.getCommandId());
             assertEquals("getPrefetchSize", expected.getPrefetchSize(), actual.getPrefetchSize());
         }
         catch (Exception e) {

activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTransportTest.java

@@ -19,6 +19,7 @@ package org.apache.activemq.transport.udp;
 import org.apache.activemq.openwire.OpenWireFormat;
 import org.apache.activemq.transport.CommandJoiner;
 import org.apache.activemq.transport.Transport;
+import org.apache.activemq.util.IntSequenceGenerator;
 
 import java.net.URI;
 
@@ -40,6 +41,7 @@ public class UdpTransportTest extends UdpTestSupport {
         
         OpenWireFormat wireFormat = createWireFormat();
         UdpTransport transport = new UdpTransport(wireFormat, new URI(producerURI));
+        transport.setSequenceGenerator(new IntSequenceGenerator());
         return new CommandJoiner(transport, wireFormat);
     }
 
@@ -47,6 +49,7 @@ public class UdpTransportTest extends UdpTestSupport {
         System.out.println("Consumer on port: " + consumerPort);
         OpenWireFormat wireFormat = createWireFormat();
         UdpTransport transport = new UdpTransport(wireFormat, consumerPort);
+        transport.setSequenceGenerator(new IntSequenceGenerator());
         return new CommandJoiner(transport, wireFormat);
     }