CAMEL-4327 added more complex rejectOld feature unit test and removed…

… rejected messages from the sequence

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

camel-core/src/main/java/org/apache/camel/processor/resequencer/ResequencerEngine.java

@@ -223,6 +223,7 @@ public synchronized void insert(E o) {
         } else if (sequence.predecessor(element) != null) {
             // nothing to schedule
         } else if (rejectOld != null && rejectOld.booleanValue() && beforeLastDelivered(element)) {
+            sequence.remove(element);
             throw new MessageRejectedException("rejecting message [" + element.getObject()
                     + "], it should have been sent before the last delivered message [" + lastDelivered.getObject() + "]");
         } else {

camel-core/src/test/java/org/apache/camel/processor/ResequenceStreamRejectOldExchangesTest.java

@@ -51,28 +51,30 @@ public void testDuplicateAfterTimeout() throws Exception {
         assertMockEndpointsSatisfied();
     }
 
-    public void testOutOfSequenceAfterTimeout() throws Exception {
-        getMockEndpoint("mock:result").expectedBodiesReceived("A", "C", "D");
-        getMockEndpoint("mock:error").expectedBodiesReceived("B");
+    public void testOutOfSequenceAfterTimeoutSimple() throws Exception {
+        getMockEndpoint("mock:result").expectedBodiesReceived("B", "C", "D");
+        getMockEndpoint("mock:error").expectedBodiesReceived("A");
 
         template.sendBodyAndHeader("direct:start", "D", "seqno", 4);
         template.sendBodyAndHeader("direct:start", "C", "seqno", 3);
-        template.sendBodyAndHeader("direct:start", "A", "seqno", 1);
-        Thread.sleep(1100);
         template.sendBodyAndHeader("direct:start", "B", "seqno", 2);
+        Thread.sleep(1100);
+        template.sendBodyAndHeader("direct:start", "A", "seqno", 1);
 
         assertMockEndpointsSatisfied();
     }
 
-    public void testOutOfSequenceAfterTimeout2() throws Exception {
-        getMockEndpoint("mock:result").expectedBodiesReceived("B", "C", "D");
-        getMockEndpoint("mock:error").expectedBodiesReceived("A");
+    public void testOutOfSequenceAfterTimeoutComplex() throws Exception {
+        getMockEndpoint("mock:result").expectedBodiesReceived("A", "D", "E", "F");
+        getMockEndpoint("mock:error").expectedBodiesReceived("B", "C");
 
         template.sendBodyAndHeader("direct:start", "D", "seqno", 4);
-        template.sendBodyAndHeader("direct:start", "C", "seqno", 3);
-        template.sendBodyAndHeader("direct:start", "B", "seqno", 2);
-        Thread.sleep(1100);
         template.sendBodyAndHeader("direct:start", "A", "seqno", 1);
+        Thread.sleep(1100);
+        template.sendBodyAndHeader("direct:start", "B", "seqno", 2);
+        template.sendBodyAndHeader("direct:start", "E", "seqno", 5);
+        template.sendBodyAndHeader("direct:start", "C", "seqno", 3);
+        template.sendBodyAndHeader("direct:start", "F", "seqno", 6);
 
         assertMockEndpointsSatisfied();
     }