You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2012/03/31 16:32:45 UTC
svn commit: r1307815 [3/3] - in /camel/trunk: apache-camel/
apache-camel/src/main/descriptors/ components/ components/camel-leveldb/
components/camel-leveldb/src/ components/camel-leveldb/src/main/
components/camel-leveldb/src/main/java/ components/cam...
Added: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryTest.java?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryTest.java (added)
+++ camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryTest.java Sat Mar 31 14:32:43 2012
@@ -0,0 +1,94 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.camel.component.leveldb;
+
+import java.io.File;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultExchange;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+public class LevelDBAggregationRepositoryTest extends CamelTestSupport {
+
+ private LevelDBFile levelDBFile;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ deleteDirectory("target/data");
+ File file = new File("target/data/leveldb.dat");
+ levelDBFile = new LevelDBFile();
+ levelDBFile.setFile(file);
+ levelDBFile.start();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ levelDBFile.stop();
+ super.tearDown();
+ }
+
+ @Test
+ public void testOperations() {
+ LevelDBAggregationRepository repo = new LevelDBAggregationRepository();
+ repo.setLevelDBFile(levelDBFile);
+ repo.setRepositoryName("repo1");
+ repo.setReturnOldExchange(true);
+
+ // Can't get something we have not put in...
+ Exchange actual = repo.get(context, "missing");
+ assertEquals(null, actual);
+
+ // Store it..
+ Exchange exchange1 = new DefaultExchange(context);
+ exchange1.getIn().setBody("counter:1");
+ actual = repo.add(context, "foo", exchange1);
+ assertEquals(null, actual);
+
+ // Get it back..
+ actual = repo.get(context, "foo");
+ assertEquals("counter:1", actual.getIn().getBody());
+
+ // Change it..
+ Exchange exchange2 = new DefaultExchange(context);
+ exchange2.getIn().setBody("counter:2");
+ actual = repo.add(context, "foo", exchange2);
+ // the old one
+ assertEquals("counter:1", actual.getIn().getBody());
+
+ // Get it back..
+ actual = repo.get(context, "foo");
+ assertEquals("counter:2", actual.getIn().getBody());
+
+ // now remove it
+ repo.remove(context, "foo", actual);
+ actual = repo.get(context, "foo");
+ assertEquals(null, actual);
+
+ // add it again
+ exchange1 = new DefaultExchange(context);
+ exchange1.getIn().setBody("counter:3");
+ actual = repo.add(context, "foo", exchange1);
+ assertEquals(null, actual);
+
+ // Get it back..
+ actual = repo.get(context, "foo");
+ assertEquals("counter:3", actual.getIn().getBody());
+ }
+
+}
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBBigPayloadTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBBigPayloadTest.java?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBBigPayloadTest.java (added)
+++ camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBBigPayloadTest.java Sat Mar 31 14:32:43 2012
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.camel.component.leveldb;
+
+import java.io.File;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.aggregate.UseLatestAggregationStrategy;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+/**
+ * Test issue with leveldb file store growing to large
+ */
+@Ignore("Run this test manually")
+public class LevelDBBigPayloadTest extends CamelTestSupport {
+
+ private static final long TIME = 60 * 1000;
+ private static final AtomicLong NUMBER = new AtomicLong();
+ private LevelDBAggregationRepository repo;
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ deleteDirectory("target/data");
+ repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
+ super.setUp();
+ }
+
+ @Test
+ public void testBigPayload() throws Exception {
+
+ log.info("Running test for " + TIME + " millis.");
+ Thread.sleep(60 * 1000);
+
+ // assert the file size of the repo is not big < 32mb
+ File file = new File("target/data/leveldb.dat");
+ assertTrue(file + " should exists", file.exists());
+ long size = file.length();
+ log.info(file + " size is " + size);
+ // should be about 32mb, so we say 34 just in case
+ assertTrue(file + " should not be so big in size, was: " + size, size < 34 * 1024 * 1024);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:foo")
+ .bean(BigPayload.class)
+ .aggregate(bean(LevelDBBigPayloadTest.class, "number"), new UseLatestAggregationStrategy())
+ .aggregationRepository(repo)
+ .completionSize(2).completionTimeout(5000)
+ .log("Aggregated key ${header.CamelAggregatedCorrelationKey}");
+ }
+ };
+ }
+
+ public static long number() {
+ // return 123; (will not cause leveldb to grow in size)
+ return NUMBER.incrementAndGet();
+ }
+
+}
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBBigPayloadTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBBigPayloadTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBExchangeSerializationTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBExchangeSerializationTest.java?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBExchangeSerializationTest.java (added)
+++ camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBExchangeSerializationTest.java Sat Mar 31 14:32:43 2012
@@ -0,0 +1,91 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.camel.component.leveldb;
+
+import java.io.File;
+import java.util.Date;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultExchange;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+public class LevelDBExchangeSerializationTest extends CamelTestSupport {
+
+ private LevelDBFile levelDBFile;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ deleteDirectory("target/data");
+ File file = new File("target/data/leveldb.dat");
+ levelDBFile = new LevelDBFile();
+ levelDBFile.setFile(file);
+ levelDBFile.start();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ levelDBFile.stop();
+ super.tearDown();
+ }
+
+ @Test
+ public void testExchangeSerialization() {
+ LevelDBAggregationRepository repo = new LevelDBAggregationRepository();
+ repo.setLevelDBFile(levelDBFile);
+ repo.setRepositoryName("repo1");
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody("Hello World");
+ exchange.getIn().setHeader("name", "Claus");
+ exchange.getIn().setHeader("number", 123);
+ exchange.setProperty("quote", "Camel rocks");
+
+ Date now = new Date();
+ exchange.getIn().setHeader("date", now);
+
+ repo.add(context, "foo", exchange);
+
+ Exchange actual = repo.get(context, "foo");
+ assertEquals("Hello World", actual.getIn().getBody());
+ assertEquals("Claus", actual.getIn().getHeader("name"));
+ assertEquals(123, actual.getIn().getHeader("number"));
+ Date date = actual.getIn().getHeader("date", Date.class);
+ assertNotNull(date);
+ assertEquals(now.getTime(), date.getTime());
+ // we do not serialize properties to avoid storing all kind of not needed information
+ assertNull(actual.getProperty("quote"));
+ assertSame(context, actual.getContext());
+
+ // change something
+ exchange.getIn().setBody("Bye World");
+ exchange.getIn().setHeader("name", "Hiram");
+ exchange.getIn().removeHeader("date");
+
+ repo.add(context, "foo", exchange);
+
+ actual = repo.get(context, "foo");
+ assertEquals("Bye World", actual.getIn().getBody());
+ assertEquals("Hiram", actual.getIn().getHeader("name"));
+ assertEquals(123, actual.getIn().getHeader("number"));
+ date = actual.getIn().getHeader("date", Date.class);
+ assertNull(date);
+ assertSame(context, actual.getContext());
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBExchangeSerializationTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBExchangeSerializationTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBGetNotFoundTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBGetNotFoundTest.java?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBGetNotFoundTest.java (added)
+++ camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBGetNotFoundTest.java Sat Mar 31 14:32:43 2012
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.camel.component.leveldb;
+
+import java.io.File;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultExchange;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+public class LevelDBGetNotFoundTest extends CamelTestSupport {
+
+ private LevelDBFile levelDBFile;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ deleteDirectory("target/data");
+ File file = new File("target/data/leveldb.dat");
+ levelDBFile = new LevelDBFile();
+ levelDBFile.setFile(file);
+ levelDBFile.start();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ levelDBFile.stop();
+ super.tearDown();
+ }
+
+ @Test
+ public void testGetNotFound() {
+ LevelDBAggregationRepository repo = new LevelDBAggregationRepository();
+ repo.setLevelDBFile(levelDBFile);
+ repo.setRepositoryName("repo1");
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody("Hello World");
+
+ Exchange out = repo.get(context, exchange.getExchangeId());
+ assertNull("Should not find exchange", out);
+ }
+
+ @Test
+ public void testPutAndGetNotFound() {
+ LevelDBAggregationRepository repo = new LevelDBAggregationRepository();
+ repo.setLevelDBFile(levelDBFile);
+ repo.setRepositoryName("repo1");
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody("Hello World");
+ log.info("Created " + exchange.getExchangeId());
+
+ repo.add(context, exchange.getExchangeId(), exchange);
+ Exchange out = repo.get(context, exchange.getExchangeId());
+ assertNotNull("Should find exchange", out);
+
+ Exchange exchange2 = new DefaultExchange(context);
+ exchange2.getIn().setBody("Bye World");
+ log.info("Created " + exchange2.getExchangeId());
+
+ Exchange out2 = repo.get(context, exchange2.getExchangeId());
+ assertNull("Should not find exchange", out2);
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBGetNotFoundTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBGetNotFoundTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.java?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.java (added)
+++ camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.java Sat Mar 31 14:32:43 2012
@@ -0,0 +1,90 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.camel.component.leveldb;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.processor.aggregate.AggregationStrategy;
+import org.apache.camel.test.junit4.CamelSpringTestSupport;
+import org.junit.Test;
+import org.springframework.context.support.AbstractXmlApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+public class LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest extends CamelSpringTestSupport {
+
+ private static AtomicInteger counter = new AtomicInteger(0);
+
+ @Override
+ protected AbstractXmlApplicationContext createApplicationContext() {
+ return new ClassPathXmlApplicationContext("org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml");
+ }
+
+ @Override
+ public void setUp() throws Exception {
+ deleteDirectory("target/data");
+ super.setUp();
+ }
+
+ @Test
+ public void testLevelDBAggregateRecover() throws Exception {
+ getMockEndpoint("mock:aggregated").setResultWaitTime(20000);
+ getMockEndpoint("mock:result").setResultWaitTime(20000);
+
+ // should fail the first 3 times and then recover
+ getMockEndpoint("mock:aggregated").expectedMessageCount(4);
+ getMockEndpoint("mock:result").expectedBodiesReceived("ABCDE");
+ // should be marked as redelivered
+ getMockEndpoint("mock:result").message(0).header(Exchange.REDELIVERED).isEqualTo(Boolean.TRUE);
+ // on the 2nd redelivery attempt we success
+ getMockEndpoint("mock:result").message(0).header(Exchange.REDELIVERY_COUNTER).isEqualTo(3);
+
+ template.sendBodyAndHeader("direct:start", "A", "id", 123);
+ template.sendBodyAndHeader("direct:start", "B", "id", 123);
+ template.sendBodyAndHeader("direct:start", "C", "id", 123);
+ template.sendBodyAndHeader("direct:start", "D", "id", 123);
+ template.sendBodyAndHeader("direct:start", "E", "id", 123);
+
+ assertMockEndpointsSatisfied(30, TimeUnit.SECONDS);
+ }
+
+ public static class MyFailProcessor implements Processor {
+
+ public void process(Exchange exchange) throws Exception {
+ int count = counter.incrementAndGet();
+ if (count <= 3) {
+ throw new IllegalArgumentException("Damn");
+ }
+ }
+ }
+
+ public static class MyAggregationStrategy implements AggregationStrategy {
+
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ if (oldExchange == null) {
+ return newExchange;
+ }
+ String body1 = oldExchange.getIn().getBody(String.class);
+ String body2 = newExchange.getIn().getBody(String.class);
+
+ oldExchange.getIn().setBody(body1 + body2);
+ return oldExchange;
+ }
+ }
+}
\ No newline at end of file
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.java?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.java (added)
+++ camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.java Sat Mar 31 14:32:43 2012
@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.camel.component.leveldb;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.aggregate.AggregationStrategy;
+import org.apache.camel.test.junit4.CamelSpringTestSupport;
+import org.junit.Test;
+import org.springframework.context.support.AbstractXmlApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+public class LevelDBSpringAggregateTest extends CamelSpringTestSupport {
+
+ @Override
+ public void setUp() throws Exception {
+ deleteDirectory("target/data");
+ super.setUp();
+ }
+
+ @Override
+ protected AbstractXmlApplicationContext createApplicationContext() {
+ return new ClassPathXmlApplicationContext("org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml");
+ }
+
+ @Test
+ public void testLevelDBAggregate() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:aggregated");
+ mock.expectedBodiesReceived("ABCDE");
+
+ template.sendBodyAndHeader("direct:start", "A", "id", 123);
+ template.sendBodyAndHeader("direct:start", "B", "id", 123);
+ template.sendBodyAndHeader("direct:start", "C", "id", 123);
+ template.sendBodyAndHeader("direct:start", "D", "id", 123);
+ template.sendBodyAndHeader("direct:start", "E", "id", 123);
+
+ assertMockEndpointsSatisfied(30, TimeUnit.SECONDS);
+ }
+
+ public static class MyAggregationStrategy implements AggregationStrategy {
+
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ if (oldExchange == null) {
+ return newExchange;
+ }
+ String body1 = oldExchange.getIn().getBody(String.class);
+ String body2 = newExchange.getIn().getBody(String.class);
+
+ oldExchange.getIn().setBody(body1 + body2);
+ return oldExchange;
+ }
+ }
+
+}
\ No newline at end of file
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: camel/trunk/components/camel-leveldb/src/test/resources/jndi.properties
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/resources/jndi.properties?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/resources/jndi.properties (added)
+++ camel/trunk/components/camel-leveldb/src/test/resources/jndi.properties Sat Mar 31 14:32:43 2012
@@ -0,0 +1,18 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You 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.
+## ---------------------------------------------------------------------------
+
+java.naming.factory.initial = org.apache.camel.util.jndi.CamelInitialContextFactory
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/jndi.properties
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/jndi.properties
------------------------------------------------------------------------------
svn:keywords = Rev Date
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/jndi.properties
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: camel/trunk/components/camel-leveldb/src/test/resources/log4j.properties
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/resources/log4j.properties?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/resources/log4j.properties (added)
+++ camel/trunk/components/camel-leveldb/src/test/resources/log4j.properties Sat Mar 31 14:32:43 2012
@@ -0,0 +1,38 @@
+## ------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You 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.
+## ------------------------------------------------------------------------
+
+#
+# The logging properties used for eclipse testing, We want to see debug output on the console.
+#
+log4j.rootLogger=INFO, file
+
+# uncomment the following to enable camel debugging
+#log4j.logger.org.apache.camel=DEBUG
+#log4j.logger.org.apache.camel.component.leveldbb=DEBUG
+#log4j.logger.org.apache.camel.component.leveldbb.LevelDBFile=TRACE
+
+# CONSOLE appender not used by default
+log4j.appender.out=org.apache.log4j.ConsoleAppender
+log4j.appender.out.layout=org.apache.log4j.PatternLayout
+#log4j.appender.out.layout.ConversionPattern=[%30.30t] %-30.30c{1} %-5p %m%n
+log4j.appender.out.layout.ConversionPattern=%d [%-50.50t] %-5p %-30.30c{1} - %m%n
+
+# File appender
+log4j.appender.file=org.apache.log4j.FileAppender
+log4j.appender.file.layout=org.apache.log4j.PatternLayout
+log4j.appender.file.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.file.file=target/camel-leveldb-test.log
\ No newline at end of file
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/log4j.properties
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/log4j.properties
------------------------------------------------------------------------------
svn:keywords = Rev Date
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/log4j.properties
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml (added)
+++ camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml Sat Mar 31 14:32:43 2012
@@ -0,0 +1,66 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You 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.
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="
+ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
+ ">
+
+ <!-- START SNIPPET: e1 -->
+
+ <!-- a persistent aggregation repository using camel-leveldb -->
+ <bean id="repo" class="org.apache.camel.component.leveldb.LevelDBAggregationRepository">
+ <!-- store the repo in the leveldb.dat file -->
+ <property name="persistentFileName" value="target/data/leveldb.dat"/>
+ <!-- and use repo2 as the repository name -->
+ <property name="repositoryName" value="repo2"/>
+ <!-- scan 2 times per second -->
+ <property name="recoveryInterval" value="500"/>
+ <!-- enable recovery -->
+ <property name="useRecovery" value="true"/>
+ </bean>
+
+ <!-- aggregate the messages using this strategy -->
+ <bean id="myAggregatorStrategy" class="org.apache.camel.component.leveldb.LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest$MyAggregationStrategy"/>
+
+ <!-- and use this processor to simulate errors -->
+ <bean id="myFailProcessor" class="org.apache.camel.component.leveldb.LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest$MyFailProcessor"/>
+
+ <!-- this is the camel routes -->
+ <camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
+
+ <route>
+ <from uri="direct:start"/>
+ <!-- aggregate using our strategy and leveldb repo, and complete when we have 5 messages aggregated -->
+ <aggregate strategyRef="myAggregatorStrategy" aggregationRepositoryRef="repo" completionSize="5">
+ <!-- correlate by header with the key id -->
+ <correlationExpression><header>id</header></correlationExpression>
+ <!-- send aggregated messages out here -->
+ <to uri="mock:aggregated"/>
+ <!-- this processor will fail the first 2 attempts -->
+ <process ref="myFailProcessor"/>
+ <!-- at the 3rd attempt we should be able to send to this mock result endpoint -->
+ <to uri="mock:result"/>
+ </aggregate>
+ </route>
+
+ </camelContext>
+ <!-- END SNIPPET: e1 -->
+
+</beans>
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml
------------------------------------------------------------------------------
svn:keywords = Rev Date
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml
------------------------------------------------------------------------------
svn:mime-type = text/xml
Added: camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml?rev=1307815&view=auto
==============================================================================
--- camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml (added)
+++ camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml Sat Mar 31 14:32:43 2012
@@ -0,0 +1,55 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You 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.
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="
+ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
+ ">
+
+ <!-- START SNIPPET: e1 -->
+
+ <!-- a persistent aggregation repository using camel-leveldb -->
+ <bean id="repo" class="org.apache.camel.component.leveldb.LevelDBAggregationRepository">
+ <!-- store the repo in the leveldb.dat file -->
+ <property name="persistentFileName" value="target/data/leveldb.dat"/>
+ <!-- and use repo2 as the repository name -->
+ <property name="repositoryName" value="repo2"/>
+ </bean>
+
+ <!-- aggregate the messages using this strategy -->
+ <bean id="myAggregatorStrategy" class="org.apache.camel.component.leveldb.LevelDBSpringAggregateTest$MyAggregationStrategy"/>
+
+ <!-- this is the camel routes -->
+ <camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
+
+ <route>
+ <from uri="direct:start"/>
+ <!-- aggregate using our strategy and leveldb repo, and complete when we have 5 messages aggregated -->
+ <aggregate strategyRef="myAggregatorStrategy" aggregationRepositoryRef="repo" completionSize="5">
+ <!-- correlate by header with the key id -->
+ <correlationExpression><header>id</header></correlationExpression>
+ <!-- send aggregated messages to the mock endpoint -->
+ <to uri="mock:aggregated"/>
+ </aggregate>
+ </route>
+
+ </camelContext>
+ <!-- END SNIPPET: e1 -->
+
+</beans>
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml
------------------------------------------------------------------------------
svn:keywords = Rev Date
Propchange: camel/trunk/components/camel-leveldb/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml
------------------------------------------------------------------------------
svn:mime-type = text/xml
Modified: camel/trunk/components/pom.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/pom.xml?rev=1307815&r1=1307814&r2=1307815&view=diff
==============================================================================
--- camel/trunk/components/pom.xml (original)
+++ camel/trunk/components/pom.xml Sat Mar 31 14:32:43 2012
@@ -107,6 +107,7 @@
<module>camel-kestrel</module>
<module>camel-krati</module>
<module>camel-ldap</module>
+ <module>camel-leveldb</module>
<module>camel-lucene</module>
<module>camel-mail</module>
<module>camel-mina</module>
Modified: camel/trunk/parent/pom.xml
URL: http://svn.apache.org/viewvc/camel/trunk/parent/pom.xml?rev=1307815&r1=1307814&r2=1307815&view=diff
==============================================================================
--- camel/trunk/parent/pom.xml (original)
+++ camel/trunk/parent/pom.xml Sat Mar 31 14:32:43 2012
@@ -121,6 +121,7 @@
<jt400-version>6.0</jt400-version>
<karaf-version>2.2.5</karaf-version>
<krati-version>0.4.4</krati-version>
+ <leveldbjni-version>1.2</leveldbjni-version>
<log4j-version>1.2.16</log4j-version>
<lucene-version>3.5.0</lucene-version>
<mina-version>1.1.7</mina-version>
@@ -579,6 +580,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-leveldb</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-mail</artifactId>
<version>${project.version}</version>
</dependency>