activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jstrac...@apache.org
Subject svn commit: r385478 - in /incubator/activemq/trunk/activemq-core/src: main/java/org/apache/activemq/transport/ main/java/org/apache/activemq/transport/reliable/ main/java/org/apache/activemq/transport/replay/ main/java/org/apache/activemq/transport/udp...
Date Mon, 13 Mar 2006 09:17:48 GMT
Author: jstrachan
Date: Mon Mar 13 01:17:43 2006
New Revision: 385478

URL: http://svn.apache.org/viewcvs?rev=385478&view=rev
Log:
refactored the reliable transport into the reliable package and added a test case demonstrating
the transport handling duplicate packets and dealing with reordering OK

Added:
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ExceptionIfDroppedReplayStrategy.java
  (contents, props changed)
      - copied, changed from r385456, incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ExceptionIfDroppedReplayStrategy.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReliableTransport.java
  (contents, props changed)
      - copied, changed from r385456, incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ReliableTransport.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReplayStrategy.java
  (contents, props changed)
      - copied, changed from r385456, incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ReplayStrategy.java
Removed:
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ReliableTransport.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ExceptionIfDroppedReplayStrategy.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ReplayStrategy.java
Modified:
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
    incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/ReliableTransportTest.java

Copied: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ExceptionIfDroppedReplayStrategy.java
(from r385456, incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ExceptionIfDroppedReplayStrategy.java)
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ExceptionIfDroppedReplayStrategy.java?p2=incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ExceptionIfDroppedReplayStrategy.java&p1=incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ExceptionIfDroppedReplayStrategy.java&r1=385456&r2=385478&rev=385478&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ExceptionIfDroppedReplayStrategy.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ExceptionIfDroppedReplayStrategy.java
Mon Mar 13 01:17:43 2006
@@ -14,25 +14,40 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.transport.replay;
-
-import org.apache.activemq.transport.ReliableTransport;
+package org.apache.activemq.transport.reliable;
 
 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 {
-        long count = actualCounter - expectedCounter;
-        throw new IOException("Packets dropped on: " + transport + " count: " + count + 
" expected: " + expectedCounter + " but was: " + actualCounter);
+    private int maximumDifference = 5;
+
+    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;
     }
 
 }

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ExceptionIfDroppedReplayStrategy.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ExceptionIfDroppedReplayStrategy.java
------------------------------------------------------------------------------
    svn:keywords = Date Author Id Revision HeadURL

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ExceptionIfDroppedReplayStrategy.java
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Copied: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReliableTransport.java
(from r385456, incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ReliableTransport.java)
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReliableTransport.java?p2=incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReliableTransport.java&p1=incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ReliableTransport.java&r1=385456&r2=385478&rev=385478&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ReliableTransport.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReliableTransport.java
Mon Mar 13 01:17:43 2006
@@ -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.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.TransportFilter;
 
 import java.io.IOException;
 import java.util.SortedSet;
@@ -33,10 +32,8 @@
  * @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,28 +46,48 @@
         boolean valid = expectedCounter == actualCounter;
 
         if (!valid) {
-            // lets add it to the list for later on
-            headers.add(command);
-
-            try {
-                replayStrategy.onDroppedPackets(this, expectedCounter, actualCounter);
-            }
-            catch (IOException e) {
-                getTransportListener().onException(e);
-            }
-
-            if (!headers.isEmpty()) {
-                // lets see if the first item in the set is the next header
-                command = (Command) headers.first();
-                valid = expectedCounter == command.getCommandId();
+            synchronized (commands) {
+                // lets add it to the list for later on
+                commands.add(command);
+
+                try {
+                    replayStrategy.onDroppedPackets(this, expectedCounter, actualCounter);
+                }
+                catch (IOException e) {
+                    getTransportListener().onException(e);
+                }
+
+                if (!commands.isEmpty()) {
+                    // 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);
+                    }
+                }
             }
         }
 
-        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);
+                    }
+                }
+            }
         }
     }
 

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReliableTransport.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReliableTransport.java
------------------------------------------------------------------------------
    svn:keywords = Date Author Id Revision HeadURL

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReliableTransport.java
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Copied: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReplayStrategy.java
(from r385456, incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ReplayStrategy.java)
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReplayStrategy.java?p2=incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReplayStrategy.java&p1=incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ReplayStrategy.java&r1=385456&r2=385478&rev=385478&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/replay/ReplayStrategy.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReplayStrategy.java
Mon Mar 13 01:17:43 2006
@@ -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 @@
  */
 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);
 

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReplayStrategy.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReplayStrategy.java
------------------------------------------------------------------------------
    svn:keywords = Date Author Id Revision HeadURL

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/reliable/ReplayStrategy.java
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java?rev=385478&r1=385477&r2=385478&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
Mon Mar 13 01:17:43 2006
@@ -22,8 +22,8 @@
 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.replay.ReplayStrategy;
+import org.apache.activemq.transport.reliable.ExceptionIfDroppedReplayStrategy;
+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;

Modified: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/ReliableTransportTest.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/ReliableTransportTest.java?rev=385478&r1=385477&r2=385478&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/ReliableTransportTest.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/ReliableTransportTest.java
Mon Mar 13 01:17:43 2006
@@ -19,8 +19,9 @@
 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.replay.ReplayStrategy;
+import org.apache.activemq.transport.reliable.ExceptionIfDroppedReplayStrategy;
+import org.apache.activemq.transport.reliable.ReliableTransport;
+import org.apache.activemq.transport.reliable.ReplayStrategy;
 
 import junit.framework.TestCase;
 
@@ -36,17 +37,33 @@
 
     public void testValidSequenceOfPackets() throws Exception {
         int[] sequenceNumbers = { 1, 2, 3, 4, 5, 6, 7 };
+        
+        sendStreamOfCommands(sequenceNumbers, true);
+    }
+    
+    public void testDuplicatePacketsDropped() throws Exception {
+        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 testInvalidSequenceOfPackets() throws Exception {
-        int[] sequenceNumbers = { 1, 2, /* 3, */  4, 5, 6, 7 };
+    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 @@
                 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);



Mime
View raw message