refactored the reliable transport into the reliable package and added a test case demonstrating the transport handling duplicate packets and dealing with reordering OK

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

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

@@ -14,25 +14,40 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.transport.replay;
+package org.apache.activemq.transport.reliable;
-
-import org.apache.activemq.transport.ReliableTransport;
 
 import java.io.IOException;
 
 /**
- * Throws an exception if packets are dropped causing the transport to be closed.
+ * Throws an exception if packets are dropped causing the transport to be
+ * closed.
  * 
  * @version $Revision$
  */
 public class ExceptionIfDroppedReplayStrategy implements ReplayStrategy {
 
-    public void onDroppedPackets(ReliableTransport transport, long expectedCounter, long actualCounter) throws IOException {
+    private int maximumDifference = 5;
-        long count = actualCounter - expectedCounter;
+
+    public void onDroppedPackets(ReliableTransport transport, int expectedCounter, int actualCounter) throws IOException {
+        long count = Math.abs(actualCounter - expectedCounter);
+        if (count > maximumDifference) {
             throw new IOException("Packets dropped on: " + transport + " count: " + count + " expected: " + expectedCounter + " but was: " + actualCounter);
         }
+    }
 
     public void onReceivedPacket(ReliableTransport transport, long expectedCounter) {
     }
 
+    public int getMaximumDifference() {
+        return maximumDifference;
+    }
+
+    /**
+     * Sets the maximum allowed difference between an expected packet and an
+     * actual packet before an error occurs
+     */
+    public void setMaximumDifference(int maximumDifference) {
+        this.maximumDifference = maximumDifference;
+    }
+
 }

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

@@ -14,13 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.transport;
+package org.apache.activemq.transport.reliable;
 
 import org.apache.activemq.command.Command;
 import org.apache.activemq.openwire.CommandIdComparator;
-import org.apache.activemq.transport.replay.ReplayStrategy;
+import org.apache.activemq.transport.Transport;
-import org.apache.commons.logging.Log;
+import org.apache.activemq.transport.TransportFilter;
-import org.apache.commons.logging.LogFactory;
 
 import java.io.IOException;
 import java.util.SortedSet;
@@ -33,10 +32,8 @@ import java.util.TreeSet;
  * @version $Revision$
  */
 public class ReliableTransport extends TransportFilter {
-    private static final Log log = LogFactory.getLog(ReliableTransport.class);
-
     private ReplayStrategy replayStrategy;
-    private SortedSet headers = new TreeSet(new CommandIdComparator());
+    private SortedSet commands = new TreeSet(new CommandIdComparator());
     private int expectedCounter = 1;
 
     public ReliableTransport(Transport next, ReplayStrategy replayStrategy) {
@@ -49,8 +46,9 @@ public class ReliableTransport extends TransportFilter {
         boolean valid = expectedCounter == actualCounter;
 
         if (!valid) {
+            synchronized (commands) {
                 // lets add it to the list for later on
-            headers.add(command);
+                commands.add(command);
 
                 try {
                     replayStrategy.onDroppedPackets(this, expectedCounter, actualCounter);
@@ -59,18 +57,37 @@ public class ReliableTransport extends TransportFilter {
                     getTransportListener().onException(e);
                 }
 
-            if (!headers.isEmpty()) {
+                if (!commands.isEmpty()) {
-                // lets see if the first item in the set is the next header
+                    // lets see if the first item in the set is the next
-                command = (Command) headers.first();
+                    // expected
+                    command = (Command) commands.first();
                     valid = expectedCounter == command.getCommandId();
+                    if (valid) {
+                        commands.remove(command);
+                    }
+                }
             }
         }
 
-        if (valid) {
+        while (valid) {
             // we've got a valid header so increment counter
             replayStrategy.onReceivedPacket(this, expectedCounter);
             expectedCounter++;
             getTransportListener().onCommand(command);
+
+            synchronized (commands) {
+                // we could have more commands left
+                valid = !commands.isEmpty();
+                if (valid) {
+                    // lets see if the first item in the set is the next
+                    // expected
+                    command = (Command) commands.first();
+                    valid = expectedCounter == command.getCommandId();
+                    if (valid) {
+                        commands.remove(command);
+                    }
+                }
+            }
         }
     }
 

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

@@ -14,9 +14,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.transport.replay;
+package org.apache.activemq.transport.reliable;
 
-import org.apache.activemq.transport.ReliableTransport;
 
 import java.io.IOException;
 
@@ -27,7 +26,7 @@ import java.io.IOException;
  */
 public interface ReplayStrategy {
 
-    void onDroppedPackets(ReliableTransport transport, long expectedCounter, long actualCounter) throws IOException;
+    void onDroppedPackets(ReliableTransport transport, int expectedCounter, int actualCounter) throws IOException;
 
     void onReceivedPacket(ReliableTransport transport, long expectedCounter);
 

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

@@ -22,8 +22,8 @@ import org.apache.activemq.command.Endpoint;
 import org.apache.activemq.openwire.OpenWireFormat;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportThreadSupport;
-import org.apache.activemq.transport.replay.ExceptionIfDroppedReplayStrategy;
+import org.apache.activemq.transport.reliable.ExceptionIfDroppedReplayStrategy;
-import org.apache.activemq.transport.replay.ReplayStrategy;
+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;

activemq-core/src/test/java/org/apache/activemq/transport/ReliableTransportTest.java

@@ -19,8 +19,9 @@ package org.apache.activemq.transport;
 import edu.emory.mathcs.backport.java.util.Queue;
 
 import org.apache.activemq.command.ConsumerInfo;
-import org.apache.activemq.transport.replay.ExceptionIfDroppedReplayStrategy;
+import org.apache.activemq.transport.reliable.ExceptionIfDroppedReplayStrategy;
-import org.apache.activemq.transport.replay.ReplayStrategy;
+import org.apache.activemq.transport.reliable.ReliableTransport;
+import org.apache.activemq.transport.reliable.ReplayStrategy;
 
 import junit.framework.TestCase;
 
@@ -40,13 +41,29 @@ public class ReliableTransportTest extends TestCase {
         sendStreamOfCommands(sequenceNumbers, true);
     }
     
-    public void testInvalidSequenceOfPackets() throws Exception {
+    public void testDuplicatePacketsDropped() throws Exception {
-        int[] sequenceNumbers = { 1, 2, /* 3, */  4, 5, 6, 7 };
+        int[] sequenceNumbers = { 1, 2, 2, 3, 4, 5, 6, 7 };
+        
+        sendStreamOfCommands(sequenceNumbers, true, 7);
+    }
+    
+    public void testWrongOrderOfPackets() throws Exception {
+        int[] sequenceNumbers = { 4, 3, 1, 5, 2, 7, 6, 8, 10, 9 };
+
+        sendStreamOfCommands(sequenceNumbers, true);
+    }
+
+    public void testMissingPacketsFails() throws Exception {
+        int[] sequenceNumbers = { 1, 2, /* 3, */  4, 5, 6, 7, 8, 9, 10 };
 
         sendStreamOfCommands(sequenceNumbers, false);
     }
 
     protected void sendStreamOfCommands(int[] sequenceNumbers, boolean expected) {
+        sendStreamOfCommands(sequenceNumbers, expected, sequenceNumbers.length);
+    }
+    
+    protected void sendStreamOfCommands(int[] sequenceNumbers, boolean expected, int expectedCount) {
         for (int i = 0; i < sequenceNumbers.length; i++) {
             int commandId = sequenceNumbers[i];
             
@@ -65,7 +82,7 @@ public class ReliableTransportTest extends TestCase {
                 e.printStackTrace();
                 fail("Caught exception: " + e);
             }
-            assertEquals("number of messages received", sequenceNumbers.length, commands.size());
+            assertEquals("number of messages received", expectedCount, commands.size());
         }
         else {
             assertTrue("Should have received an exception!", exceptions.size() > 0);