camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject svn commit: r930647 - in /camel/trunk: camel-core/src/main/java/org/apache/camel/spi/ components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/ components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/
Date Sun, 04 Apr 2010 08:58:43 GMT
Author: davsclaus
Date: Sun Apr  4 08:58:43 2010
New Revision: 930647

URL: http://svn.apache.org/viewvc?rev=930647&view=rev
Log:
CAMEL-2568: added more tests.

Added:
    camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadAndRecoverTest.java
      - copied, changed from r930635, camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadTest.java
Modified:
    camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RecoverableAggregationRepository.java
    camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java
    camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadConcurrentTest.java

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RecoverableAggregationRepository.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RecoverableAggregationRepository.java?rev=930647&r1=930646&r2=930647&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RecoverableAggregationRepository.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RecoverableAggregationRepository.java
Sun Apr  4 08:58:43 2010
@@ -56,6 +56,13 @@ public interface RecoverableAggregationR
     void setCheckInterval(long interval, TimeUnit timeUnit);
 
     /**
+     * Sets the interval between scans
+     *
+     * @param interval  the interval in millis
+     */
+    void setCheckInterval(long interval);
+
+    /**
      * Gets the interval between scans in millis.
      *
      * @return the interval in millis

Modified: camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java?rev=930647&r1=930646&r2=930647&view=diff
==============================================================================
--- camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java
(original)
+++ camel/trunk/components/camel-hawtdb/src/main/java/org/apache/camel/component/hawtdb/HawtDBAggregationRepository.java
Sun Apr  4 08:58:43 2010
@@ -51,7 +51,7 @@ public class HawtDBAggregationRepository
     private boolean returnOldExchange;
     private HawtDBCamelMarshaller<K> marshaller = new HawtDBCamelMarshaller<K>();
     private long interval = 5000;
-    private boolean useRecovery = false;
+    private boolean useRecovery = true;
 
     /**
      * Creates an aggregation repository
@@ -348,6 +348,10 @@ public class HawtDBAggregationRepository
         this.interval = timeUnit.toMillis(interval);
     }
 
+    public void setCheckInterval(long interval) {
+        this.interval = interval;
+    }
+
     public long getCheckIntervalInMillis() {
         return interval;
     }

Copied: camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadAndRecoverTest.java
(from r930635, camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadAndRecoverTest.java?p2=camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadAndRecoverTest.java&p1=camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadTest.java&r1=930635&r2=930647&rev=930647&view=diff
==============================================================================
--- camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadTest.java
(original)
+++ camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadAndRecoverTest.java
Sun Apr  4 08:58:43 2010
@@ -16,7 +16,12 @@
  */
 package org.apache.camel.component.hawtdb;
 
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.component.mock.MockEndpoint;
 import org.apache.camel.processor.aggregate.AggregationStrategy;
@@ -24,9 +29,10 @@ import org.apache.camel.test.junit4.Came
 import org.junit.Before;
 import org.junit.Test;
 
-public class HawtDBAggregateLoadTest extends CamelTestSupport {
+public class HawtDBAggregateLoadAndRecoverTest extends CamelTestSupport {
 
-    private static final int SIZE = 5000;
+    private static final int SIZE = 1000;
+    private static final AtomicInteger counter = new AtomicInteger();
 
     @Before
     @Override
@@ -36,9 +42,9 @@ public class HawtDBAggregateLoadTest ext
     }
 
     @Test
-    public void testLoadTestHawtDBAggregate() throws Exception {
+    public void testLoadAndRecoverHawtDBAggregate() throws Exception {
         MockEndpoint mock = getMockEndpoint("mock:result");
-        mock.expectedMinimumMessageCount(1);
+        mock.expectedMessageCount(SIZE / 10);
         mock.setResultWaitTime(30 * 1000);
 
         System.out.println("Staring to send " + SIZE + " messages.");
@@ -46,12 +52,23 @@ public class HawtDBAggregateLoadTest ext
         for (int i = 0; i < SIZE; i++) {
             final int value = 1;
             char id = 'A';
-            template.sendBodyAndHeader("seda:start?size=" + SIZE, value, "id", "" + id);
+            Map headers = new HashMap();
+            headers.put("id", id);
+            headers.put("seq", i);
+            template.sendBodyAndHeaders("seda:start?size=" + SIZE, value, headers);
         }
 
         System.out.println("Sending all " + SIZE + " message done. Now waiting for aggregation
to complete.");
 
         assertMockEndpointsSatisfied();
+
+        int recovered = 0;
+        for (Exchange exchange : mock.getReceivedExchanges()) {
+            if (exchange.getIn().getHeader(Exchange.REDELIVERED) != null) {
+                recovered++;
+            }
+        }
+        assertEquals("There should be 5 recovered", 5, recovered);
     }
 
     @Override
@@ -60,13 +77,23 @@ public class HawtDBAggregateLoadTest ext
             @Override
             public void configure() throws Exception {
                 HawtDBAggregationRepository<String> repo = new HawtDBAggregationRepository<String>("repo1",
"target/data/hawtdb.dat");
+                repo.setUseRecovery(true);
 
                 from("seda:start?size=" + SIZE)
                     .to("log:input?groupSize=500")
                     .aggregate(header("id"), new MyAggregationStrategy())
                         .aggregationRepository(repo)
-                        .completionSize(SIZE)
+                        .completionSize(10)
                         .to("log:output?showHeaders=true")
+                        // have every 20th exchange fail which should then be recovered
+                        .process(new Processor() {
+                            public void process(Exchange exchange) throws Exception {
+                                int num = counter.incrementAndGet();
+                                if (num % 20 == 0) {
+                                    throw new IllegalStateException("Failed for num " + num);
+                                }
+                            }
+                        })
                         .to("mock:result")
                     .end();
             }

Modified: camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadConcurrentTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadConcurrentTest.java?rev=930647&r1=930646&r2=930647&view=diff
==============================================================================
--- camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadConcurrentTest.java
(original)
+++ camel/trunk/components/camel-hawtdb/src/test/java/org/apache/camel/component/hawtdb/HawtDBAggregateLoadConcurrentTest.java
Sun Apr  4 08:58:43 2010
@@ -43,7 +43,7 @@ public class HawtDBAggregateLoadConcurre
     @Test
     public void testLoadTestHawtDBAggregate() throws Exception {
         MockEndpoint mock = getMockEndpoint("mock:result");
-        mock.expectedMinimumMessageCount(10);
+        mock.expectedMinimumMessageCount(9);
         mock.setResultWaitTime(30 * 1000);
 
         ExecutorService executor = Executors.newFixedThreadPool(10);



Mime
View raw message