You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2022/04/01 17:32:40 UTC
[camel] branch main updated: CAMEL-17879: removed the deprecated camel-leveldb-legacy component
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 01f0b2f CAMEL-17879: removed the deprecated camel-leveldb-legacy component
01f0b2f is described below
commit 01f0b2fe7ca1e44711103969a0797655a6aab94d
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Fri Apr 1 17:33:21 2022 +0200
CAMEL-17879: removed the deprecated camel-leveldb-legacy component
---
bom/camel-bom/pom.xml | 5 -
catalog/camel-allcomponents/pom.xml | 4 -
.../org/apache/camel/catalog/others.properties | 1 -
.../camel/catalog/others/leveldb-legacy.json | 15 -
.../bonita/api/BonitaAuthFilterConnectionTest.java | 1 -
.../component/bonita/api/BonitaAuthFilterTest.java | 1 -
components/camel-leveldb-legacy/.gitignore | 1 -
components/camel-leveldb-legacy/pom.xml | 100 -----
.../services/org/apache/camel/other.properties | 7 -
.../src/generated/resources/leveldb-legacy.json | 15 -
.../src/main/docs/leveldb-legacy.adoc | 164 -------
.../leveldb/LevelDBAggregationRepository.java | 481 ---------------------
.../camel/component/leveldb/LevelDBCamelCodec.java | 95 ----
.../camel/component/leveldb/LevelDBFile.java | 218 ----------
.../apache/camel/component/leveldb/package.html | 26 --
.../apache/camel/component/leveldb/BigPayload.java | 34 --
...elDBAggregateConcurrentDifferentGroupsTest.java | 112 -----
.../LevelDBAggregateConcurrentSameGroupTest.java | 111 -----
.../LevelDBAggregateDiscardOnTimeoutTest.java | 106 -----
.../LevelDBAggregateForceCompletionHeaderTest.java | 125 ------
.../LevelDBAggregateLoadAndRecoverTest.java | 141 ------
.../LevelDBAggregateLoadConcurrentTest.java | 117 -----
.../leveldb/LevelDBAggregateLoadTest.java | 102 -----
...DBAggregateNotLostRemovedWhenConfirmedTest.java | 100 -----
.../leveldb/LevelDBAggregateNotLostTest.java | 117 -----
...ggregateRecoverDeadLetterChannelFailedTest.java | 121 ------
...velDBAggregateRecoverDeadLetterChannelTest.java | 116 -----
.../leveldb/LevelDBAggregateRecoverTest.java | 111 -----
...DBAggregateRecoverWithRedeliveryPolicyTest.java | 117 -----
.../LevelDBAggregateRecoverWithSedaTest.java | 116 -----
.../LevelDBAggregateSerializedHeadersTest.java | 110 -----
.../component/leveldb/LevelDBAggregateTest.java | 95 ----
...velDBAggregateTimeoutCompletionRestartTest.java | 97 -----
.../LevelDBAggregationRepositoryAlotDataTest.java | 114 -----
...velDBAggregationRepositoryLoadExistingTest.java | 93 ----
...velDBAggregationRepositoryMultipleRepoTest.java | 134 ------
...DBAggregationRepositoryRecoverExistingTest.java | 89 ----
.../leveldb/LevelDBAggregationRepositoryTest.java | 104 -----
.../component/leveldb/LevelDBBigPayloadTest.java | 88 ----
.../leveldb/LevelDBExchangeSerializationTest.java | 104 -----
.../component/leveldb/LevelDBGetNotFoundTest.java | 95 ----
.../camel/component/leveldb/LevelDBSetupTest.java | 69 ---
...ngAggregateRecoverWithRedeliveryPolicyTest.java | 100 -----
.../leveldb/LevelDBSpringAggregateTest.java | 78 ----
.../org/apache/camel/component/util/HeaderDto.java | 60 ---
.../src/test/resources/jndi.properties | 18 -
.../src/test/resources/log4j2.properties | 28 --
...ingAggregateRecoverWithRedeliveryPolicyTest.xml | 68 ---
.../leveldb/LevelDBSpringAggregateTest.xml | 57 ---
components/pom.xml | 1 -
.../others/examples/json/leveldb-legacy.json | 1 -
docs/components/modules/others/nav.adoc | 1 -
.../modules/others/pages/leveldb-legacy.adoc | 1 -
parent/pom.xml | 5 -
54 files changed, 4390 deletions(-)
diff --git a/bom/camel-bom/pom.xml b/bom/camel-bom/pom.xml
index acb5ef80..dc27938 100644
--- a/bom/camel-bom/pom.xml
+++ b/bom/camel-bom/pom.xml
@@ -1334,11 +1334,6 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
- <artifactId>camel-leveldb-legacy</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.camel</groupId>
<artifactId>camel-log</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/catalog/camel-allcomponents/pom.xml b/catalog/camel-allcomponents/pom.xml
index 9c118dd..8b12bd8 100644
--- a/catalog/camel-allcomponents/pom.xml
+++ b/catalog/camel-allcomponents/pom.xml
@@ -887,10 +887,6 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
- <artifactId>camel-leveldb-legacy</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.camel</groupId>
<artifactId>camel-leveldb</artifactId>
</dependency>
<dependency>
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties
index 3dda39f..0f296d8 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties
@@ -28,7 +28,6 @@ jta
kamelet-main
kotlin-dsl
leveldb
-leveldb-legacy
lra
main
management
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/leveldb-legacy.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/leveldb-legacy.json
deleted file mode 100644
index a863a4b..0000000
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/leveldb-legacy.json
+++ /dev/null
@@ -1,15 +0,0 @@
-{
- "other": {
- "kind": "other",
- "name": "leveldb-legacy",
- "title": "LevelDB-legacy",
- "description": "Using LevelDB as persistent EIP store",
- "deprecated": true,
- "firstVersion": "2.10.0",
- "label": "database",
- "supportLevel": "Stable",
- "groupId": "org.apache.camel",
- "artifactId": "camel-leveldb-legacy",
- "version": "3.17.0-SNAPSHOT"
- }
-}
diff --git a/components/camel-bonita/src/test/java/org/apache/camel/component/bonita/api/BonitaAuthFilterConnectionTest.java b/components/camel-bonita/src/test/java/org/apache/camel/component/bonita/api/BonitaAuthFilterConnectionTest.java
index c2b8296..d272c3e 100644
--- a/components/camel-bonita/src/test/java/org/apache/camel/component/bonita/api/BonitaAuthFilterConnectionTest.java
+++ b/components/camel-bonita/src/test/java/org/apache/camel/component/bonita/api/BonitaAuthFilterConnectionTest.java
@@ -16,7 +16,6 @@
*/
package org.apache.camel.component.bonita.api;
-import java.io.IOException;
import java.util.HashMap;
import javax.ws.rs.client.ClientRequestContext;
diff --git a/components/camel-bonita/src/test/java/org/apache/camel/component/bonita/api/BonitaAuthFilterTest.java b/components/camel-bonita/src/test/java/org/apache/camel/component/bonita/api/BonitaAuthFilterTest.java
index 599dc16..2df382c 100644
--- a/components/camel-bonita/src/test/java/org/apache/camel/component/bonita/api/BonitaAuthFilterTest.java
+++ b/components/camel-bonita/src/test/java/org/apache/camel/component/bonita/api/BonitaAuthFilterTest.java
@@ -16,7 +16,6 @@
*/
package org.apache.camel.component.bonita.api;
-import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
diff --git a/components/camel-leveldb-legacy/.gitignore b/components/camel-leveldb-legacy/.gitignore
deleted file mode 100644
index c52b279..0000000
--- a/components/camel-leveldb-legacy/.gitignore
+++ /dev/null
@@ -1 +0,0 @@
-leveldb.dat
diff --git a/components/camel-leveldb-legacy/pom.xml b/components/camel-leveldb-legacy/pom.xml
deleted file mode 100644
index 84a353b..0000000
--- a/components/camel-leveldb-legacy/pom.xml
+++ /dev/null
@@ -1,100 +0,0 @@
-<?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.
-
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.camel</groupId>
- <artifactId>components</artifactId>
- <version>3.17.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>camel-leveldb-legacy</artifactId>
- <packaging>jar</packaging>
- <name>Camel :: LevelDB Legacy (deprecated)</name>
- <description>Using LevelDB as persistent EIP store</description>
-
- <properties>
- <firstVersion>2.10.0</firstVersion>
- <label>database</label>
- <title>LevelDB-legacy</title>
-
- <camel.osgi.import.pkg>
- org.apache.camel*,
- org.fusesource.hawtbuf*,
- org.slf4j,
- org.fusesource.leveldbjni;resolution:=optional,
- org.iq80.leveldb;resolution:=optional,
- org.iq80.leveldb.impl;resolution:=optional
- </camel.osgi.import.pkg>
- </properties>
-
- <dependencies>
-
- <dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-support</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.fusesource.leveldbjni</groupId>
- <artifactId>leveldbjni-all</artifactId>
- <version>${leveldbjni-version}</version>
- </dependency>
- <dependency>
- <groupId>org.iq80.leveldb</groupId>
- <artifactId>leveldb</artifactId>
- <version>${leveldb-version}</version>
- <exclusions>
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${leveldb-guava-version}</version>
- </dependency>
- <dependency>
- <groupId>org.fusesource.hawtbuf</groupId>
- <artifactId>hawtbuf</artifactId>
- <version>${hawtbuf-version}</version>
- </dependency>
-
- <!-- testing -->
- <dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-test-spring-junit5</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-slf4j-impl</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.junit.jupiter</groupId>
- <artifactId>junit-jupiter</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
-</project>
diff --git a/components/camel-leveldb-legacy/src/generated/resources/META-INF/services/org/apache/camel/other.properties b/components/camel-leveldb-legacy/src/generated/resources/META-INF/services/org/apache/camel/other.properties
deleted file mode 100644
index 4a52d8c..0000000
--- a/components/camel-leveldb-legacy/src/generated/resources/META-INF/services/org/apache/camel/other.properties
+++ /dev/null
@@ -1,7 +0,0 @@
-# Generated by camel build tools - do NOT edit this file!
-name=leveldb-legacy
-groupId=org.apache.camel
-artifactId=camel-leveldb-legacy
-version=3.17.0-SNAPSHOT
-projectName=Camel :: LevelDB Legacy (deprecated)
-projectDescription=Using LevelDB as persistent EIP store
diff --git a/components/camel-leveldb-legacy/src/generated/resources/leveldb-legacy.json b/components/camel-leveldb-legacy/src/generated/resources/leveldb-legacy.json
deleted file mode 100644
index a863a4b..0000000
--- a/components/camel-leveldb-legacy/src/generated/resources/leveldb-legacy.json
+++ /dev/null
@@ -1,15 +0,0 @@
-{
- "other": {
- "kind": "other",
- "name": "leveldb-legacy",
- "title": "LevelDB-legacy",
- "description": "Using LevelDB as persistent EIP store",
- "deprecated": true,
- "firstVersion": "2.10.0",
- "label": "database",
- "supportLevel": "Stable",
- "groupId": "org.apache.camel",
- "artifactId": "camel-leveldb-legacy",
- "version": "3.17.0-SNAPSHOT"
- }
-}
diff --git a/components/camel-leveldb-legacy/src/main/docs/leveldb-legacy.adoc b/components/camel-leveldb-legacy/src/main/docs/leveldb-legacy.adoc
deleted file mode 100644
index 270ef44..0000000
--- a/components/camel-leveldb-legacy/src/main/docs/leveldb-legacy.adoc
+++ /dev/null
@@ -1,164 +0,0 @@
-= LevelDB-legacy Component (deprecated)
-:doctitle: LevelDB-legacy
-:shortname: leveldb-legacy
-:artifactid: camel-leveldb-legacy
-:description: Using LevelDB as persistent EIP store
-:since: 2.10
-:supportlevel: Stable-deprecated
-:deprecated: *deprecated*
-//Manually maintained attributes
-:camel-spring-boot-name: leveldb-legacy
-
-*Since Camel {since}*
-
-*Deprecated* use camel-leveDB instead.
-(This legacy library should be used only if db file with existing data is used. Newer version of camel-levelDB
-uses different serialization method, which is not back compatible.)
-
-https://code.google.com/p/leveldb/[Leveldb] is a very lightweight and
-embedable key value database. It allows together with Camel to provide
-persistent support for various Camel features such as
-Aggregator.
-
-Current features it provides:
-
-* LevelDBAggregationRepository
-
-[[LevelDB-UsingLevelDBAggregationRepository]]
-== Using LevelDBAggregationRepository
-
-`LevelDBAggregationRepository` is an `AggregationRepository` which on
-the fly persists the aggregated messages. This ensures that you will not
-loose messages, as the default aggregator will use an in memory only
-`AggregationRepository`.
-
-It has the following options:
-
-[width="100%",cols="10%,10%,80%",options="header",]
-|=======================================================================
-
-|Option |Type |Description
-
-|`repositoryName` |String |A mandatory repository name. Allows you to use a shared `LevelDBFile`
-for multiple repositories.
-
-|`persistentFileName` |String |Filename for the persistent storage. If no file exists on startup a new
-file is created.
-
-|`levelDBFile` |LevelDBFile |Use an existing configured
-`org.apache.camel.component.leveldb.LevelDBFile` instance.
-
-|`sync` |boolean |Whether or not the LevelDBFile should sync on write or
-not. Default is false. By sync on write ensures that its always waiting
-for all writes to be spooled to disk and thus will not loose updates.
-See http://leveldb.googlecode.com/svn/trunk/doc/index.html[LevelDB docs]
-for more details about async vs sync writes.
-
-|`returnOldExchange` |boolean |Whether the get operation should return the old existing Exchange if any
-existed. By default this option is `false` to optimize as we do not need
-the old exchange when aggregating.
-
-|`useRecovery` |boolean |Whether or not recovery is enabled. This option is by default `true`.
-When enabled the Camel Aggregator automatic
-recover failed aggregated exchange and have them resubmitted.
-
-|`recoveryInterval` |long |If recovery is enabled then a background task is run every x'th time to
-scan for failed exchanges to recover and resubmit. By default this
-interval is 5000 millis.
-
-|`maximumRedeliveries` |int |Allows you to limit the maximum number of redelivery attempts for a
-recovered exchange. If enabled then the Exchange will be moved to the
-dead letter channel if all redelivery attempts failed. By default this
-option is disabled. If this option is used then the `deadLetterUri`
-option must also be provided.
-
-|`deadLetterUri` |String |An endpoint uri for a Dead Letter Channel
-where exhausted recovered Exchanges will be moved. If this option is
-used then the `maximumRedeliveries` option must also be provided.
-|=======================================================================
-
-The `repositoryName` option must be provided. Then either the
-`persistentFileName` or the `levelDBFile` must be provided.
-
-[[LevelDB-Whatispreservedwhenpersisting]]
-=== What is preserved when persisting
-
-`LevelDBAggregationRepository` will only preserve any `Serializable`
-compatible message body data types. Message headers must be primitive /
-string / numbers / etc. If a data type is not such a type its dropped
-and a `WARN` is logged. And it only persists the `Message` body and the
-`Message` headers. The `Exchange` properties are *not* persisted.
-
-[[LevelDB-Recovery]]
-=== Recovery
-
-The `LevelDBAggregationRepository` will by default recover any failed
-Exchange. It does this by having a background tasks
-that scans for failed Exchanges in the persistent
-store. You can use the `checkInterval` option to set how often this task
-runs. The recovery works as transactional which ensures that Camel will
-try to recover and redeliver the failed Exchange.
-Any Exchange which was found to be recovered will be
-restored from the persistent store and resubmitted and send out again.
-
-The following headers is set when an Exchange is
-being recovered/redelivered:
-
-[width="100%",cols="10%,10%,80%",options="header",]
-|=======================================================================
-|Header |Type |Description
-
-|`Exchange.REDELIVERED` |Boolean |Is set to true to indicate the Exchange is being
-redelivered.
-
-|`Exchange.REDELIVERY_COUNTER` |Integer |The redelivery attempt, starting from 1.
-|=======================================================================
-
-Only when an Exchange has been successfully
-processed it will be marked as complete which happens when the `confirm`
-method is invoked on the `AggregationRepository`. This means if the same
-Exchange fails again it will be kept retried until
-it success.
-
-You can use option `maximumRedeliveries` to limit the maximum number of
-redelivery attempts for a given recovered Exchange.
-You must also set the `deadLetterUri` option so Camel knows where to
-send the Exchange when the `maximumRedeliveries` was
-hit.
-
-You can see some examples in the unit tests of camel-leveldb, for
-example
-https://svn.apache.org/repos/asf/camel/trunk/components/camel-leveldb/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverTest.java[this
-test].
-
-[[LevelDB-UsingLevelDBAggregationRepositoryinJavaDSL]]
-== Using LevelDBAggregationRepository in Java DSL
-
-In this example we want to persist aggregated messages in the
-`target/data/leveldb.dat` file.
-
-[[LevelDB-UsingLevelDBAggregationRepositoryinSpringXML]]
-== Using LevelDBAggregationRepository in Spring XML
-
-The same example but using Spring XML instead:
-
-[[LevelDB-Dependencies]]
-== Dependencies
-
-To use LevelDB in your camel routes you need to add
-the a dependency on *camel-leveldb*.
-
-If you use maven you could just add the following to your pom.xml,
-substituting the version number for the latest & greatest release (see
-the download page for the latest versions).
-
-[source,xml]
-----------------------------------------
-<dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-leveldb</artifactId>
- <version>x.y.z</version>
-</dependency>
-----------------------------------------
-
-include::spring-boot:partial$starter.adoc[]
diff --git a/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/LevelDBAggregationRepository.java b/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/LevelDBAggregationRepository.java
deleted file mode 100644
index fc60dbb..0000000
--- a/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/LevelDBAggregationRepository.java
+++ /dev/null
@@ -1,481 +0,0 @@
-/*
- * 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.IOException;
-import java.io.UnsupportedEncodingException;
-import java.util.Collections;
-import java.util.LinkedHashSet;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.camel.CamelContext;
-import org.apache.camel.Exchange;
-import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.spi.RecoverableAggregationRepository;
-import org.apache.camel.support.service.ServiceHelper;
-import org.apache.camel.support.service.ServiceSupport;
-import org.apache.camel.util.IOHelper;
-import org.apache.camel.util.ObjectHelper;
-import org.apache.camel.util.StringHelper;
-import org.fusesource.hawtbuf.Buffer;
-import org.iq80.leveldb.DBIterator;
-import org.iq80.leveldb.WriteBatch;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * An instance of {@link org.apache.camel.spi.AggregationRepository} which is backed by a {@link LevelDBFile}.
- *
- * @deprecated Use {@link org.apache.camel.component.leveldb.LevelDBAggregationRepository} from component camel-leveldb.
- */
-@Deprecated
-public class LevelDBAggregationRepository extends ServiceSupport implements RecoverableAggregationRepository {
-
- private static final Logger LOG = LoggerFactory.getLogger(LevelDBAggregationRepository.class);
-
- private LevelDBFile levelDBFile;
- private String persistentFileName;
- private String repositoryName;
- private boolean sync;
- private boolean returnOldExchange;
- private LevelDBCamelCodec codec = new LevelDBCamelCodec();
- private long recoveryInterval = 5000;
- private boolean useRecovery = true;
- private int maximumRedeliveries;
- private String deadLetterUri;
- private boolean allowSerializedHeaders;
-
- /**
- * Creates an aggregation repository
- */
- public LevelDBAggregationRepository() {
- }
-
- /**
- * Creates an aggregation repository
- *
- * @param repositoryName the repository name
- */
- public LevelDBAggregationRepository(String repositoryName) {
- StringHelper.notEmpty(repositoryName, "repositoryName");
- this.repositoryName = repositoryName;
- }
-
- /**
- * Creates an aggregation repository using a new {@link LevelDBFile} that persists using the provided file.
- *
- * @param repositoryName the repository name
- * @param persistentFileName the persistent store filename
- */
- public LevelDBAggregationRepository(String repositoryName, String persistentFileName) {
- StringHelper.notEmpty(repositoryName, "repositoryName");
- StringHelper.notEmpty(persistentFileName, "persistentFileName");
- this.repositoryName = repositoryName;
- this.persistentFileName = persistentFileName;
- }
-
- /**
- * Creates an aggregation repository using the provided {@link LevelDBFile}.
- *
- * @param repositoryName the repository name
- * @param levelDBFile the leveldb file to use as persistent store
- */
- public LevelDBAggregationRepository(String repositoryName, LevelDBFile levelDBFile) {
- StringHelper.notEmpty(repositoryName, "repositoryName");
- ObjectHelper.notNull(levelDBFile, "levelDBFile");
- this.levelDBFile = levelDBFile;
- this.repositoryName = repositoryName;
- }
-
- @Override
- public Exchange add(final CamelContext camelContext, final String key, final Exchange exchange) {
- LOG.debug("Adding key [{}] -> {}", key, exchange);
- try {
- byte[] lDbKey = keyBuilder(repositoryName, key);
- final Buffer exchangeBuffer = codec.marshallExchange(camelContext, exchange, allowSerializedHeaders);
-
- byte[] rc = null;
- if (isReturnOldExchange()) {
- rc = levelDBFile.getDb().get(lDbKey);
- }
-
- LOG.trace("Adding key index {} for repository {}", key, repositoryName);
- levelDBFile.getDb().put(lDbKey, exchangeBuffer.toByteArray(), levelDBFile.getWriteOptions());
- LOG.trace("Added key index {}", key);
-
- if (rc == null) {
- return null;
- }
-
- // only return old exchange if enabled
- if (isReturnOldExchange()) {
- return codec.unmarshallExchange(camelContext, new Buffer(rc));
- }
- } catch (IOException e) {
- throw new RuntimeCamelException("Error adding to repository " + repositoryName + " with key " + key, e);
- }
-
- return null;
- }
-
- @Override
- public Exchange get(final CamelContext camelContext, final String key) {
- Exchange answer = null;
-
- try {
- byte[] lDbKey = keyBuilder(repositoryName, key);
- LOG.trace("Getting key index {}", key);
- byte[] rc = levelDBFile.getDb().get(lDbKey);
-
- if (rc != null) {
- answer = codec.unmarshallExchange(camelContext, new Buffer(rc));
- }
- } catch (IOException e) {
- throw new RuntimeCamelException("Error getting key " + key + " from repository " + repositoryName, e);
- }
-
- LOG.debug("Getting key [{}] -> {}", key, answer);
- return answer;
- }
-
- @Override
- public void remove(final CamelContext camelContext, final String key, final Exchange exchange) {
- LOG.debug("Removing key [{}]", key);
-
- try {
- byte[] lDbKey = keyBuilder(repositoryName, key);
- final String exchangeId = exchange.getExchangeId();
- final Buffer exchangeBuffer = codec.marshallExchange(camelContext, exchange, allowSerializedHeaders);
-
- // remove the exchange
- byte[] rc = levelDBFile.getDb().get(lDbKey);
-
- if (rc != null) {
- WriteBatch batch = levelDBFile.getDb().createWriteBatch();
- try {
- batch.delete(lDbKey);
- LOG.trace("Removed key index {} -> {}", key, new Buffer(rc));
-
- // add exchange to confirmed index
- byte[] confirmedLDBKey = keyBuilder(getRepositoryNameCompleted(), exchangeId);
- batch.put(confirmedLDBKey, exchangeBuffer.toByteArray());
- LOG.trace("Added confirm index {} for repository {}", exchangeId, getRepositoryNameCompleted());
-
- levelDBFile.getDb().write(batch, levelDBFile.getWriteOptions());
- } finally {
- batch.close();
- }
- }
-
- } catch (IOException e) {
- throw new RuntimeCamelException("Error removing key " + key + " from repository " + repositoryName, e);
- }
- }
-
- @Override
- public void confirm(final CamelContext camelContext, final String exchangeId) {
- LOG.debug("Confirming exchangeId [{}]", exchangeId);
-
- byte[] confirmedLDBKey = keyBuilder(getRepositoryNameCompleted(), exchangeId);
-
- byte[] rc = levelDBFile.getDb().get(confirmedLDBKey);
-
- if (rc != null) {
- levelDBFile.getDb().delete(confirmedLDBKey);
- LOG.trace("Removed confirm index {} -> {}", exchangeId, new Buffer(rc));
- }
- }
-
- @Override
- public Set<String> getKeys() {
- final Set<String> keys = new LinkedHashSet<>();
-
- // interval task could potentially be running while we are shutting down so check for that
- if (!isRunAllowed()) {
- return null;
- }
-
- DBIterator it = levelDBFile.getDb().iterator();
-
- String keyBuffer;
- try {
- String prefix = repositoryName + '\0';
- for (it.seek(keyBuilder(repositoryName, "")); it.hasNext(); it.next()) {
- if (!isRunAllowed()) {
- break;
- }
- keyBuffer = asString(it.peekNext().getKey());
-
- if (!keyBuffer.startsWith(prefix)) {
- break;
- }
-
- String key = keyBuffer.substring(prefix.length());
-
- LOG.trace("getKey [{}]", key);
- keys.add(key);
- }
- } finally {
- // Make sure you close the iterator to avoid resource leaks.
- IOHelper.close(it);
- }
-
- return Collections.unmodifiableSet(keys);
- }
-
- @Override
- public Set<String> scan(CamelContext camelContext) {
- final Set<String> answer = new LinkedHashSet<>();
-
- if (!isRunAllowed()) {
- return null;
- }
-
- DBIterator it = levelDBFile.getDb().iterator();
-
- String keyBuffer;
- try {
- String prefix = getRepositoryNameCompleted() + '\0';
-
- for (it.seek(keyBuilder(getRepositoryNameCompleted(), "")); it.hasNext(); it.next()) {
- keyBuffer = asString(it.peekNext().getKey());
-
- if (!keyBuffer.startsWith(prefix)) {
- break;
- }
- String exchangeId = keyBuffer.substring(prefix.length());
-
- LOG.trace("Scan exchangeId [{}]", exchangeId);
- answer.add(exchangeId);
- }
- } finally {
- // Make sure you close the iterator to avoid resource leaks.
- IOHelper.close(it);
- }
-
- if (answer.isEmpty()) {
- LOG.trace("Scanned and found no exchange to recover.");
- } else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Scanned and found {} exchange(s) to recover (note some of them may already be in progress).",
- answer.size());
- }
- }
- return answer;
-
- }
-
- @Override
- public Exchange recover(CamelContext camelContext, final String exchangeId) {
- Exchange answer = null;
-
- try {
- byte[] completedLDBKey = keyBuilder(getRepositoryNameCompleted(), exchangeId);
-
- byte[] rc = levelDBFile.getDb().get(completedLDBKey);
-
- if (rc != null) {
- answer = codec.unmarshallExchange(camelContext, new Buffer(rc));
- }
- } catch (IOException e) {
- throw new RuntimeCamelException(
- "Error recovering exchangeId " + exchangeId + " from repository " + repositoryName, e);
- }
-
- LOG.debug("Recovering exchangeId [{}] -> {}", exchangeId, answer);
- return answer;
- }
-
- private int size(final String repositoryName) {
- DBIterator it = levelDBFile.getDb().iterator();
-
- String prefix = repositoryName + '\0';
- int count = 0;
- try {
- for (it.seek(keyBuilder(repositoryName, "")); it.hasNext(); it.next()) {
- if (!asString(it.peekNext().getKey()).startsWith(prefix)) {
- break;
- }
- count++;
- }
- } finally {
- // Make sure you close the iterator to avoid resource leaks.
- IOHelper.close(it);
- }
-
- LOG.debug("Size of repository [{}] -> {}", repositoryName, count);
- return count;
- }
-
- public LevelDBFile getLevelDBFile() {
- return levelDBFile;
- }
-
- public void setLevelDBFile(LevelDBFile levelDBFile) {
- this.levelDBFile = levelDBFile;
- }
-
- public String getRepositoryName() {
- return repositoryName;
- }
-
- private String getRepositoryNameCompleted() {
- return repositoryName + "-completed";
- }
-
- public void setRepositoryName(String repositoryName) {
- this.repositoryName = repositoryName;
- }
-
- public boolean isSync() {
- return sync;
- }
-
- public void setSync(boolean sync) {
- this.sync = sync;
- }
-
- public boolean isReturnOldExchange() {
- return returnOldExchange;
- }
-
- public void setReturnOldExchange(boolean returnOldExchange) {
- this.returnOldExchange = returnOldExchange;
- }
-
- @Override
- public void setRecoveryInterval(long interval, TimeUnit timeUnit) {
- this.recoveryInterval = timeUnit.toMillis(interval);
- }
-
- @Override
- public void setRecoveryInterval(long interval) {
- this.recoveryInterval = interval;
- }
-
- @Override
- public long getRecoveryIntervalInMillis() {
- return recoveryInterval;
- }
-
- @Override
- public boolean isUseRecovery() {
- return useRecovery;
- }
-
- @Override
- public void setUseRecovery(boolean useRecovery) {
- this.useRecovery = useRecovery;
- }
-
- @Override
- public int getMaximumRedeliveries() {
- return maximumRedeliveries;
- }
-
- @Override
- public void setMaximumRedeliveries(int maximumRedeliveries) {
- this.maximumRedeliveries = maximumRedeliveries;
- }
-
- @Override
- public String getDeadLetterUri() {
- return deadLetterUri;
- }
-
- @Override
- public void setDeadLetterUri(String deadLetterUri) {
- this.deadLetterUri = deadLetterUri;
- }
-
- public String getPersistentFileName() {
- return persistentFileName;
- }
-
- public void setPersistentFileName(String persistentFileName) {
- this.persistentFileName = persistentFileName;
- }
-
- public boolean isAllowSerializedHeaders() {
- return allowSerializedHeaders;
- }
-
- public void setAllowSerializedHeaders(boolean allowSerializedHeaders) {
- this.allowSerializedHeaders = allowSerializedHeaders;
- }
-
- @Override
- protected void doStart() throws Exception {
- // either we have a LevelDB configured or we use a provided fileName
- if (levelDBFile == null && persistentFileName != null) {
- levelDBFile = new LevelDBFile();
- levelDBFile.setSync(isSync());
- levelDBFile.setFileName(persistentFileName);
- }
-
- ObjectHelper.notNull(levelDBFile, "Either set a persistentFileName or a levelDBFile");
- ObjectHelper.notNull(repositoryName, "repositoryName");
-
- ServiceHelper.startService(levelDBFile);
-
- // log number of existing exchanges
- int current = size(getRepositoryName());
- int completed = size(getRepositoryNameCompleted());
-
- if (current > 0) {
- LOG.info("On startup there are {} aggregate exchanges (not completed) in repository: {}",
- current, getRepositoryName());
- } else {
- LOG.info("On startup there are no existing aggregate exchanges (not completed) in repository: {}",
- getRepositoryName());
- }
- if (completed > 0) {
- LOG.warn("On startup there are {} completed exchanges to be recovered in repository: {}",
- completed, getRepositoryNameCompleted());
- } else {
- LOG.info("On startup there are no completed exchanges to be recovered in repository: {}",
- getRepositoryNameCompleted());
- }
- }
-
- @Override
- protected void doStop() throws Exception {
- ServiceHelper.stopService(levelDBFile);
- }
-
- public static byte[] keyBuilder(String repo, String key) {
- try {
- return (repo + '\0' + key).getBytes("UTF-8");
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeCamelException(e);
- }
- }
-
- public static String asString(byte[] value) {
- if (value == null) {
- return null;
- } else {
- try {
- return new String(value, "UTF-8");
- } catch (UnsupportedEncodingException var2) {
- throw new RuntimeCamelException(var2);
- }
- }
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/LevelDBCamelCodec.java b/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/LevelDBCamelCodec.java
deleted file mode 100644
index d967c58..0000000
--- a/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/LevelDBCamelCodec.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.IOException;
-
-import org.apache.camel.CamelContext;
-import org.apache.camel.Endpoint;
-import org.apache.camel.Exchange;
-import org.apache.camel.ExchangePropertyKey;
-import org.apache.camel.ExtendedExchange;
-import org.apache.camel.support.DefaultExchange;
-import org.apache.camel.support.DefaultExchangeHolder;
-import org.fusesource.hawtbuf.Buffer;
-import org.fusesource.hawtbuf.DataByteArrayInputStream;
-import org.fusesource.hawtbuf.DataByteArrayOutputStream;
-import org.fusesource.hawtbuf.codec.Codec;
-import org.fusesource.hawtbuf.codec.ObjectCodec;
-import org.fusesource.hawtbuf.codec.StringCodec;
-
-public final class LevelDBCamelCodec {
-
- private Codec<String> keyCodec = new StringCodec();
- private Codec<DefaultExchangeHolder> exchangeCodec = new ObjectCodec<>();
-
- public Buffer marshallKey(String key) throws IOException {
- DataByteArrayOutputStream baos = new DataByteArrayOutputStream();
- keyCodec.encode(key, baos);
- return baos.toBuffer();
- }
-
- public String unmarshallKey(Buffer buffer) throws IOException {
- DataByteArrayInputStream bais = new DataByteArrayInputStream(buffer);
- String key = keyCodec.decode(bais);
- return key;
- }
-
- public Buffer marshallExchange(CamelContext camelContext, Exchange exchange, boolean allowSerializedHeaders)
- throws IOException {
- DataByteArrayOutputStream baos = new DataByteArrayOutputStream();
- // use DefaultExchangeHolder to marshal to a serialized object
- DefaultExchangeHolder pe = DefaultExchangeHolder.marshal(exchange, false, allowSerializedHeaders);
- // add the aggregated size and timeout property as the only properties we want to retain
- DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_SIZE,
- exchange.getProperty(ExchangePropertyKey.AGGREGATED_SIZE, Integer.class));
- DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_TIMEOUT,
- exchange.getProperty(ExchangePropertyKey.AGGREGATED_TIMEOUT, Long.class));
- // add the aggregated completed by property to retain
- DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_COMPLETED_BY,
- exchange.getProperty(ExchangePropertyKey.AGGREGATED_COMPLETED_BY, String.class));
- // add the aggregated correlation key property to retain
- DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_CORRELATION_KEY,
- exchange.getProperty(ExchangePropertyKey.AGGREGATED_CORRELATION_KEY, String.class));
- // and a guard property if using the flexible toolbox aggregator
- DefaultExchangeHolder.addProperty(pe, Exchange.AGGREGATED_COLLECTION_GUARD,
- exchange.getProperty(Exchange.AGGREGATED_COLLECTION_GUARD, String.class));
- // persist the from endpoint as well
- if (exchange.getFromEndpoint() != null) {
- DefaultExchangeHolder.addProperty(pe, "CamelAggregatedFromEndpoint", exchange.getFromEndpoint().getEndpointUri());
- }
- exchangeCodec.encode(pe, baos);
- return baos.toBuffer();
- }
-
- public Exchange unmarshallExchange(CamelContext camelContext, Buffer buffer) throws IOException {
- DataByteArrayInputStream bais = new DataByteArrayInputStream(buffer);
- DefaultExchangeHolder pe = exchangeCodec.decode(bais);
- Exchange answer = new DefaultExchange(camelContext);
- DefaultExchangeHolder.unmarshal(answer, pe);
- // restore the from endpoint
- String fromEndpointUri = (String) answer.removeProperty("CamelAggregatedFromEndpoint");
- if (fromEndpointUri != null) {
- Endpoint fromEndpoint = camelContext.hasEndpoint(fromEndpointUri);
- if (fromEndpoint != null) {
- answer.adapt(ExtendedExchange.class).setFromEndpoint(fromEndpoint);
- }
- }
- return answer;
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/LevelDBFile.java b/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/LevelDBFile.java
deleted file mode 100644
index 5545611..0000000
--- a/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/LevelDBFile.java
+++ /dev/null
@@ -1,218 +0,0 @@
-/*
- * 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.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-
-import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.Service;
-import org.apache.camel.util.IOHelper;
-import org.apache.camel.util.ObjectHelper;
-import org.iq80.leveldb.CompressionType;
-import org.iq80.leveldb.DB;
-import org.iq80.leveldb.DBFactory;
-import org.iq80.leveldb.Options;
-import org.iq80.leveldb.WriteOptions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Manages access to a shared <a href="https://github.com/fusesource/leveldbjni/">LevelDB</a> file.
- * <p/>
- * Will by default not sync writes which allows it to be faster. You can force syncing by setting the sync option to
- * <tt>true</tt>.
- */
-public class LevelDBFile implements Service {
-
- private static final Logger LOG = LoggerFactory.getLogger(LevelDBFile.class);
-
- private DB db;
- private File file;
- private int writeBufferSize = 4 << 20;
- private int maxOpenFiles = 1000;
- private int blockRestartInterval = 16;
- private int blockSize = 4 * 1024;
- private String compressionType;
- private boolean verifyChecksums = true;
- private boolean paranoidChecks;
- private long cacheSize = 32 << 20;
- private boolean sync;
-
- public DB getDb() {
- return db;
- }
-
- public File getFile() {
- return file;
- }
-
- public void setFile(File file) {
- this.file = file;
- }
-
- public String getFileName() throws IOException {
- return file.getCanonicalPath();
- }
-
- public void setFileName(String fileName) {
- this.file = new File(fileName);
- }
-
- public int getWriteBufferSize() {
- return writeBufferSize;
- }
-
- public void setWriteBufferSize(int writeBufferSize) {
- this.writeBufferSize = writeBufferSize;
- }
-
- public int getMaxOpenFiles() {
- return maxOpenFiles;
- }
-
- public void setMaxOpenFiles(int maxOpenFiles) {
- this.maxOpenFiles = maxOpenFiles;
- }
-
- public int getBlockRestartInterval() {
- return blockRestartInterval;
- }
-
- public void setBlockRestartInterval(int blockRestartInterval) {
- this.blockRestartInterval = blockRestartInterval;
- }
-
- public int getBlockSize() {
- return blockSize;
- }
-
- public void setBlockSize(int blockSize) {
- this.blockSize = blockSize;
- }
-
- public String getCompressionType() {
- return compressionType;
- }
-
- public void setCompressionType(String compressionType) {
- this.compressionType = compressionType;
- }
-
- public boolean isVerifyChecksums() {
- return verifyChecksums;
- }
-
- public void setVerifyChecksums(boolean verifyChecksums) {
- this.verifyChecksums = verifyChecksums;
- }
-
- public boolean isParanoidChecks() {
- return paranoidChecks;
- }
-
- public void setParanoidChecks(boolean paranoidChecks) {
- this.paranoidChecks = paranoidChecks;
- }
-
- public long getCacheSize() {
- return cacheSize;
- }
-
- public void setCacheSize(long cacheSize) {
- this.cacheSize = cacheSize;
- }
-
- public boolean isSync() {
- return sync;
- }
-
- public void setSync(boolean sync) {
- this.sync = sync;
- }
-
- public WriteOptions getWriteOptions() {
- WriteOptions options = new WriteOptions();
- options.sync(sync);
- return options;
- }
-
- @Override
- public void start() {
- if (getFile() == null) {
- throw new IllegalArgumentException("A file must be configured");
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Starting LevelDB using file: {}", getFile());
- }
-
- Options options = new Options().writeBufferSize(writeBufferSize).maxOpenFiles(maxOpenFiles)
- .blockRestartInterval(blockRestartInterval).blockSize(blockSize).verifyChecksums(verifyChecksums)
- .paranoidChecks(paranoidChecks).cacheSize(cacheSize);
-
- if ("snappy".equals(compressionType)) {
- options.compressionType(CompressionType.SNAPPY);
- } else {
- options.compressionType(CompressionType.NONE);
- }
-
- options.createIfMissing(true);
- try {
- final Path dbFile = Paths.get(this.getFileName());
- Files.createDirectories(dbFile.getParent());
- DBFactory factory = getFactory();
- db = factory.open(getFile(), options);
- } catch (IOException ioe) {
- throw new RuntimeCamelException("Error opening LevelDB with file " + getFile(), ioe);
- }
- }
-
- private DBFactory getFactory() {
- String[] classNames = new String[] {
- "org.fusesource.leveldbjni.JniDBFactory",
- "org.iq80.leveldb.impl.Iq80DBFactory"
- };
- for (String cn : classNames) {
- try {
- Class<?> clz = ObjectHelper.loadClass(cn, getClass().getClassLoader());
- DBFactory factory = (DBFactory) clz.newInstance();
- if (LOG.isDebugEnabled()) {
- LOG.debug("Using {} implementation of org.iq80.leveldb.DBFactory", factory.getClass().getName());
- }
- return factory;
- } catch (Throwable ignored) {
- // must be throwable
- }
- }
- throw new IllegalStateException("Can't find implementation of org.iq80.leveldb.DBFactory");
- }
-
- @Override
- public void stop() {
- File file = getFile();
-
- LOG.debug("Stopping LevelDB using file: {}", file);
- if (db != null) {
- IOHelper.close(db);
- db = null;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/package.html b/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/package.html
deleted file mode 100644
index 7e78f99..0000000
--- a/components/camel-leveldb-legacy/src/main/java/org/apache/camel/component/leveldb/package.html
+++ /dev/null
@@ -1,26 +0,0 @@
-<!--
-
- 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.
-
--->
-<html>
-<head>
-</head>
-<body>
-Camel <a href="https://github.com/fusesource/leveldbjni">LevelDB</a> support
-
-</body>
-</html>
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/BigPayload.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/BigPayload.java
deleted file mode 100644
index aa1269d..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/BigPayload.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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 org.apache.camel.Exchange;
-import org.apache.camel.Handler;
-
-/**
- * a 3mb payload
- */
-public class BigPayload {
-
- private int bodySize = 3 * 1024 * 1024;
-
- @Handler
- public void process(Exchange exchange) {
- exchange.getIn().setBody(new byte[bodySize]);
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateConcurrentDifferentGroupsTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateConcurrentDifferentGroupsTest.java
deleted file mode 100644
index 81e90cf..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateConcurrentDifferentGroupsTest.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * 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.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-import org.apache.camel.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateConcurrentDifferentGroupsTest extends CamelTestSupport {
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- super.setUp();
- }
-
- @Test
- public void testNoConcurrentProducers() throws Exception {
- doSendMessages(2, 1);
- }
-
- @Test
- public void testConcurrentProducers() throws Exception {
- doSendMessages(10, 5);
- }
-
- @Test
- public void testMoreConcurrentProducers() throws Exception {
- doSendMessages(50, 10);
- }
-
- private void doSendMessages(int files, int poolSize) throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:aggregated");
- mock.expectedMessageCount(2);
- mock.setResultWaitTime(30 * 1000L);
-
- ExecutorService executor = Executors.newFixedThreadPool(poolSize);
- for (int i = 0; i < files; i++) {
- final int index = i;
- executor.submit(new Callable<Object>() {
- public Object call() throws Exception {
- String id = index % 2 == 0 ? "A" : "B";
- template.sendBodyAndHeader("direct:start", index, "id", id);
- // simulate a little delay
- Thread.sleep(3);
- return null;
- }
- });
- }
-
- assertMockEndpointsSatisfied();
- executor.shutdownNow();
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- public void configure() throws Exception {
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
-
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionTimeout(1000).aggregationRepository(repo)
- .to("mock:aggregated");
- }
- };
- }
-
- private static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateConcurrentSameGroupTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateConcurrentSameGroupTest.java
deleted file mode 100644
index da79df3..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateConcurrentSameGroupTest.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * 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.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-import org.apache.camel.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateConcurrentSameGroupTest extends CamelTestSupport {
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- super.setUp();
- }
-
- @Test
- public void testNoConcurrentProducers() throws Exception {
- doSendMessages(1, 1);
- }
-
- @Test
- public void testConcurrentProducers() throws Exception {
- doSendMessages(10, 5);
- }
-
- @Test
- public void testMoreConcurrentProducers() throws Exception {
- doSendMessages(50, 10);
- }
-
- private void doSendMessages(int files, int poolSize) throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:aggregated");
- mock.setResultWaitTime(30 * 1000L);
- mock.expectedMessageCount(1);
-
- ExecutorService executor = Executors.newFixedThreadPool(poolSize);
- for (int i = 0; i < files; i++) {
- final int index = i;
- executor.submit(new Callable<Object>() {
- public Object call() throws Exception {
- template.sendBodyAndHeader("direct:start", index, "id", 123);
- // simulate a little delay
- Thread.sleep(3);
- return null;
- }
- });
- }
-
- assertMockEndpointsSatisfied();
- executor.shutdownNow();
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- public void configure() throws Exception {
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
-
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionTimeout(1000).aggregationRepository(repo)
- .to("mock:aggregated");
- }
- };
- }
-
- private static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateDiscardOnTimeoutTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateDiscardOnTimeoutTest.java
deleted file mode 100644
index de732d4..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateDiscardOnTimeoutTest.java
+++ /dev/null
@@ -1,106 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateDiscardOnTimeoutTest extends CamelTestSupport {
-
- private LevelDBAggregationRepository repo;
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- // enable recovery
- repo.setUseRecovery(true);
- // check faster
- repo.setRecoveryInterval(500, TimeUnit.MILLISECONDS);
- super.setUp();
- }
-
- @Test
- public void testAggregateDiscardOnTimeout() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:aggregated");
- mock.expectedMessageCount(0);
-
- template.sendBodyAndHeader("direct:start", "A", "id", 123);
- template.sendBodyAndHeader("direct:start", "B", "id", 123);
-
- // wait 3 seconds
- Thread.sleep(3000);
-
- mock.assertIsSatisfied();
-
- // now send 3 which does not timeout
- mock.reset();
- mock.expectedBodiesReceived("C+D+E");
-
- template.sendBodyAndHeader("direct:start", "A", "id", 123);
- template.sendBodyAndHeader("direct:start", "B", "id", 123);
- template.sendBodyAndHeader("direct:start", "C", "id", 123);
-
- // should complete before timeout
- mock.await(1500, TimeUnit.MILLISECONDS);
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionSize(3).aggregationRepository(repo)
- // use a 3 second timeout
- .completionTimeout(2000)
- // and if timeout occurred then just discard the aggregated message
- .discardOnCompletionTimeout()
- .to("mock:aggregated");
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateForceCompletionHeaderTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateForceCompletionHeaderTest.java
deleted file mode 100644
index ff7bf8e..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateForceCompletionHeaderTest.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/*
- * 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.HashMap;
-import java.util.Map;
-
-import org.apache.camel.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-/**
- * To test CAMEL-4118 support for completing all aggregation groups with a signal message
- */
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateForceCompletionHeaderTest extends CamelTestSupport {
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- super.setUp();
- }
-
- @Test
- public void testForceCompletionTrueExclusive() throws Exception {
-
- getMockEndpoint("mock:aggregated").expectedMessageCount(0);
-
- template.sendBodyAndHeader("direct:start", "test1", "id", "1");
- template.sendBodyAndHeader("direct:start", "test2", "id", "2");
- template.sendBodyAndHeader("direct:start", "test3", "id", "1");
- template.sendBodyAndHeader("direct:start", "test4", "id", "2");
-
- assertMockEndpointsSatisfied();
-
- getMockEndpoint("mock:aggregated").expectedMessageCount(2);
- getMockEndpoint("mock:aggregated").expectedBodiesReceivedInAnyOrder("test1test3", "test2test4");
- getMockEndpoint("mock:aggregated").expectedPropertyReceived(Exchange.AGGREGATED_COMPLETED_BY, "force");
-
- //now send the signal message to trigger completion of all groups, message should NOT be aggregated
- template.sendBodyAndProperty("direct:start", "test5", Exchange.AGGREGATION_COMPLETE_ALL_GROUPS, true);
-
- assertMockEndpointsSatisfied();
- }
-
- @Test
- public void testForceCompletionTrueInclusive() throws Exception {
-
- getMockEndpoint("mock:aggregated").expectedMessageCount(0);
-
- template.sendBodyAndHeader("direct:start", "test1", "id", "1");
- template.sendBodyAndHeader("direct:start", "test2", "id", "2");
- template.sendBodyAndHeader("direct:start", "test3", "id", "1");
- template.sendBodyAndHeader("direct:start", "test4", "id", "2");
-
- assertMockEndpointsSatisfied();
-
- getMockEndpoint("mock:aggregated").expectedMessageCount(3);
- getMockEndpoint("mock:aggregated").expectedBodiesReceivedInAnyOrder("test1test3", "test2test4", "test5");
- getMockEndpoint("mock:aggregated").expectedPropertyReceived(Exchange.AGGREGATED_COMPLETED_BY, "force");
-
- //now send a message to trigger completion of all groups, message should be aggregated
- Map<String, Object> headers = new HashMap<>();
- headers.put("id", "3");
- headers.put(Exchange.AGGREGATION_COMPLETE_ALL_GROUPS_INCLUSIVE, true);
- template.sendBodyAndHeaders("direct:start", "test5", headers);
-
- assertMockEndpointsSatisfied();
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- // create the leveldb repo
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
-
- // here is the Camel route where we aggregate
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- // use our created leveldb repo as aggregation repository
- .completionSize(10).aggregationRepository(repo)
- .to("mock:aggregated");
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateLoadAndRecoverTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateLoadAndRecoverTest.java
deleted file mode 100644
index 1eab3d5..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateLoadAndRecoverTest.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- * 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.HashMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.camel.AggregationStrategy;
-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.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateLoadAndRecoverTest extends CamelTestSupport {
-
- private static final Logger LOG = LoggerFactory.getLogger(LevelDBAggregateLoadAndRecoverTest.class);
- private static final int SIZE = 200;
- private static AtomicInteger counter = new AtomicInteger();
-
- @BeforeEach
- @Override
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- super.setUp();
- }
-
- @Test
- public void testLoadAndRecoverLevelDBAggregate() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result");
- mock.expectedMessageCount(SIZE / 10);
- mock.setResultWaitTime(50 * 1000);
-
- LOG.info("Staring to send " + SIZE + " messages.");
-
- for (int i = 0; i < SIZE; i++) {
- final int value = 1;
- char id = 'A';
- Map<String, Object> headers = new HashMap<>();
- headers.put("id", id);
- headers.put("seq", i);
- LOG.debug("Sending {} with id {}", value, id);
- template.sendBodyAndHeaders("seda:start", value, headers);
- // simulate a little delay
- Thread.sleep(5);
- }
-
- LOG.info("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++;
- }
- }
- int expected = SIZE / 10 / 10;
- int delta = Math.abs(expected - recovered);
- if (delta == 0) {
- assertEquals(expected, recovered, "There should be " + expected + " recovered");
- } else {
- assertTrue(delta < 3, "We expected " + expected + " recovered but the delta is within accepted range " + delta);
- }
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- repo.setUseRecovery(true);
- // for faster unit testing
- repo.setRecoveryInterval(500);
-
- from("seda:start?size=" + SIZE)
- .to("log:input?groupSize=500")
- .aggregate(header("id"), new MyAggregationStrategy())
- .aggregationRepository(repo)
- .completionSize(10)
- .to("log:output?showHeaders=true")
- // have every 10th exchange fail which should then be recovered
- .process(new Processor() {
- public void process(Exchange exchange) throws Exception {
- int num = counter.incrementAndGet();
- if (num % 10 == 0) {
- throw new IllegalStateException("Failed for num " + num);
- }
- }
- })
- .to("mock:result")
- .end();
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
- if (oldExchange == null) {
- return newExchange;
- }
-
- Integer body1 = oldExchange.getIn().getBody(Integer.class);
- Integer body2 = newExchange.getIn().getBody(Integer.class);
- int sum = body1 + body2;
-
- oldExchange.getIn().setBody(sum);
- return oldExchange;
- }
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateLoadConcurrentTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateLoadConcurrentTest.java
deleted file mode 100644
index 0090716..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateLoadConcurrentTest.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-import org.apache.camel.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateLoadConcurrentTest extends CamelTestSupport {
-
- private static final Logger LOG = LoggerFactory.getLogger(LevelDBAggregateLoadConcurrentTest.class);
- private static final char[] KEYS = new char[] { 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J' };
- private static final int SIZE = 500;
-
- @BeforeEach
- @Override
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- super.setUp();
- }
-
- @Test
- public void testLoadTestLevelDBAggregate() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result");
- mock.expectedMessageCount(10);
- mock.setResultWaitTime(50 * 1000);
-
- ExecutorService executor = Executors.newFixedThreadPool(10);
-
- LOG.info("Staring to send " + SIZE + " messages.");
-
- for (int i = 0; i < SIZE; i++) {
- final int value = 1;
- final int key = i % 10;
- executor.submit(new Callable<Object>() {
- public Object call() throws Exception {
- char id = KEYS[key];
- LOG.debug("Sending {} with id {}", value, id);
- template.sendBodyAndHeader("direct:start", value, "id", "" + id);
- // simulate a little delay
- Thread.sleep(3);
- return null;
- }
- });
- }
-
- LOG.info("Sending all " + SIZE + " message done. Now waiting for aggregation to complete.");
-
- assertMockEndpointsSatisfied();
- executor.shutdownNow();
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
-
- from("direct:start")
- .to("log:input?groupSize=500")
- .aggregate(header("id"), new MyAggregationStrategy())
- .aggregationRepository(repo)
- .completionSize(SIZE / 10)
- .to("log:output?showHeaders=true")
- .to("mock:result")
- .end();
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
- if (oldExchange == null) {
- return newExchange;
- }
-
- Integer body1 = oldExchange.getIn().getBody(Integer.class);
- Integer body2 = newExchange.getIn().getBody(Integer.class);
- int sum = body1 + body2;
-
- oldExchange.getIn().setBody(sum);
- return oldExchange;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateLoadTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateLoadTest.java
deleted file mode 100644
index d736d3c..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateLoadTest.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * 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 org.apache.camel.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateLoadTest extends CamelTestSupport {
-
- private static final Logger LOG = LoggerFactory.getLogger(LevelDBAggregateLoadTest.class);
- private static final int SIZE = 500;
- private LevelDBAggregationRepository repo;
-
- @BeforeEach
- @Override
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- super.setUp();
- }
-
- @Test
- public void testLoadTestLevelDBAggregate() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result");
- mock.expectedMinimumMessageCount(1);
- mock.setResultWaitTime(50 * 1000);
-
- LOG.info("Staring to send " + SIZE + " messages.");
-
- for (int i = 0; i < SIZE; i++) {
- final int value = 1;
- char id = 'A';
- LOG.debug("Sending {} with id {}", value, id);
- template.sendBodyAndHeader("seda:start?size=" + SIZE, value, "id", "" + id);
- }
-
- LOG.info("Sending all " + SIZE + " message done. Now waiting for aggregation to complete.");
-
- assertMockEndpointsSatisfied();
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("seda:start?size=" + SIZE)
- .to("log:input?groupSize=500")
- .aggregate(header("id"), new MyAggregationStrategy())
- .aggregationRepository(repo)
- .completionSize(SIZE)
- .to("log:output?showHeaders=true")
- .to("mock:result")
- .end();
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
- if (oldExchange == null) {
- return newExchange;
- }
-
- Integer body1 = oldExchange.getIn().getBody(Integer.class);
- Integer body2 = newExchange.getIn().getBody(Integer.class);
- int sum = body1 + body2;
-
- oldExchange.getIn().setBody(sum);
- return oldExchange;
- }
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateNotLostRemovedWhenConfirmedTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateNotLostRemovedWhenConfirmedTest.java
deleted file mode 100644
index e60d77d..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateNotLostRemovedWhenConfirmedTest.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.component.leveldb.LevelDBAggregationRepository.keyBuilder;
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertNull;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateNotLostRemovedWhenConfirmedTest extends CamelTestSupport {
-
- private LevelDBAggregationRepository repo;
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- super.setUp();
- }
-
- @Test
- public void testLevelDBAggregateNotLostRemovedWhenConfirmed() throws Exception {
- getMockEndpoint("mock:result").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);
-
- Thread.sleep(1000);
-
- String exchangeId = getMockEndpoint("mock:result").getReceivedExchanges().get(0).getExchangeId();
-
- // the exchange should NOT be in the completed repo as it was confirmed
- final LevelDBFile levelDBFile = repo.getLevelDBFile();
- byte[] bf = levelDBFile.getDb().get(keyBuilder("repo1-completed", exchangeId));
-
- // assert the exchange was deleted
- assertNull(bf);
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionSize(5).aggregationRepository(repo)
- .log("aggregated exchange id ${exchangeId} with ${body}")
- .to("mock:result")
- .end();
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateNotLostTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateNotLostTest.java
deleted file mode 100644
index 2fb0e73..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateNotLostTest.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.fusesource.hawtbuf.Buffer;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.component.leveldb.LevelDBAggregationRepository.keyBuilder;
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateNotLostTest extends CamelTestSupport {
-
- private LevelDBAggregationRepository repo;
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- super.setUp();
- }
-
- @Test
- public void testLevelDBAggregateNotLost() throws Exception {
- getMockEndpoint("mock:aggregated").expectedBodiesReceived("ABCDE");
- getMockEndpoint("mock:result").expectedMessageCount(0);
-
- 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);
-
- Thread.sleep(1000);
-
- String exchangeId = getMockEndpoint("mock:aggregated").getReceivedExchanges().get(0).getExchangeId();
-
- // the exchange should be in the completed repo where we should be able to find it
- final LevelDBFile levelDBFile = repo.getLevelDBFile();
- final LevelDBCamelCodec codec = new LevelDBCamelCodec();
- byte[] bf = levelDBFile.getDb().get(keyBuilder("repo1-completed", exchangeId));
-
- // assert the exchange was not lost and we got all the information still
- assertNotNull(bf);
- Exchange completed = codec.unmarshallExchange(context, new Buffer(bf));
- assertNotNull(completed);
- // should retain the exchange id
- assertEquals(exchangeId, completed.getExchangeId());
- assertEquals("ABCDE", completed.getIn().getBody());
- assertEquals(123, completed.getIn().getHeader("id"));
- assertEquals("size", completed.getProperty(Exchange.AGGREGATED_COMPLETED_BY));
- assertEquals(5, completed.getProperty(Exchange.AGGREGATED_SIZE));
- // will store correlation keys as String
- assertEquals("123", completed.getProperty(Exchange.AGGREGATED_CORRELATION_KEY));
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionSize(5).aggregationRepository(repo)
- .log("aggregated exchange id ${exchangeId} with ${body}")
- .to("mock:aggregated")
- // throw an exception to fail, which we then will loose this message
- .throwException(new IllegalArgumentException("Damn"))
- .to("mock:result")
- .end();
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverDeadLetterChannelFailedTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverDeadLetterChannelFailedTest.java
deleted file mode 100644
index 0344f7a..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverDeadLetterChannelFailedTest.java
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateRecoverDeadLetterChannelFailedTest extends CamelTestSupport {
-
- private LevelDBAggregationRepository repo;
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- // enable recovery
- repo.setUseRecovery(true);
- // exhaust after at most 2 attempts
- repo.setMaximumRedeliveries(2);
- // and move to this dead letter channel
- repo.setDeadLetterUri("direct:dead");
- // check faster
- repo.setRecoveryInterval(1000, TimeUnit.MILLISECONDS);
-
- super.setUp();
- }
-
- @Test
- public void testLevelDBAggregateRecoverDeadLetterChannelFailed() throws Exception {
- // should fail all times
- getMockEndpoint("mock:result").expectedMessageCount(0);
- getMockEndpoint("mock:aggregated").expectedMessageCount(3);
- // it should keep sending to DLC if it failed, so test for min 2 attempts
- getMockEndpoint("mock:dead").expectedMinimumMessageCount(2);
- // all the details should be the same about redelivered and redelivered 2 times
- getMockEndpoint("mock:dead").message(0).header(Exchange.REDELIVERED).isEqualTo(Boolean.TRUE);
- getMockEndpoint("mock:dead").message(0).header(Exchange.REDELIVERY_COUNTER).isEqualTo(2);
- getMockEndpoint("mock:dead").message(1).header(Exchange.REDELIVERY_COUNTER).isEqualTo(2);
- getMockEndpoint("mock:dead").message(1).header(Exchange.REDELIVERED).isEqualTo(Boolean.TRUE);
-
- 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);
-
- // all the details should be the same about redelivered and redelivered 2 times
- Exchange first = getMockEndpoint("mock:dead").getReceivedExchanges().get(0);
- assertEquals(true, first.getIn().getHeader(Exchange.REDELIVERED));
- assertEquals(2, first.getIn().getHeader(Exchange.REDELIVERY_COUNTER));
-
- Exchange second = getMockEndpoint("mock:dead").getReceivedExchanges().get(1);
- assertEquals(true, second.getIn().getHeader(Exchange.REDELIVERED));
- assertEquals(2, first.getIn().getHeader(Exchange.REDELIVERY_COUNTER));
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionSize(5).aggregationRepository(repo)
- .log("aggregated exchange id ${exchangeId} with ${body}")
- .to("mock:aggregated")
- .throwException(new IllegalArgumentException("Damn"))
- .to("mock:result")
- .end();
-
- from("direct:dead")
- .to("mock:dead")
- .throwException(new IllegalArgumentException("We are dead"));
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverDeadLetterChannelTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverDeadLetterChannelTest.java
deleted file mode 100644
index 1b70dca..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverDeadLetterChannelTest.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateRecoverDeadLetterChannelTest extends CamelTestSupport {
-
- private LevelDBAggregationRepository repo;
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- // enable recovery
- repo.setUseRecovery(true);
- // exhaust after at most 3 attempts
- repo.setMaximumRedeliveries(3);
- // and move to this dead letter channel
- repo.setDeadLetterUri("mock:dead");
- // check faster
- repo.setRecoveryInterval(500, TimeUnit.MILLISECONDS);
-
- super.setUp();
- }
-
- @Test
- public void testLevelDBAggregateRecoverDeadLetterChannel() throws Exception {
- // should fail all times
- getMockEndpoint("mock:result").expectedMessageCount(0);
-
- getMockEndpoint("mock:aggregated").expectedMessageCount(4);
- getMockEndpoint("mock:aggregated").message(0).header(Exchange.REDELIVERED).isNull();
- getMockEndpoint("mock:aggregated").message(1).header(Exchange.REDELIVERED).isEqualTo(Boolean.TRUE);
- getMockEndpoint("mock:aggregated").message(1).header(Exchange.REDELIVERY_COUNTER).isEqualTo(1);
- getMockEndpoint("mock:aggregated").message(1).header(Exchange.REDELIVERY_MAX_COUNTER).isEqualTo(3);
- getMockEndpoint("mock:aggregated").message(2).header(Exchange.REDELIVERED).isEqualTo(Boolean.TRUE);
- getMockEndpoint("mock:aggregated").message(2).header(Exchange.REDELIVERY_COUNTER).isEqualTo(2);
- getMockEndpoint("mock:aggregated").message(2).header(Exchange.REDELIVERY_MAX_COUNTER).isEqualTo(3);
- getMockEndpoint("mock:aggregated").message(3).header(Exchange.REDELIVERED).isEqualTo(Boolean.TRUE);
- getMockEndpoint("mock:aggregated").message(3).header(Exchange.REDELIVERY_COUNTER).isEqualTo(3);
- getMockEndpoint("mock:aggregated").message(3).header(Exchange.REDELIVERY_MAX_COUNTER).isEqualTo(3);
-
- getMockEndpoint("mock:dead").expectedBodiesReceived("ABCDE");
- getMockEndpoint("mock:dead").message(0).header(Exchange.REDELIVERED).isEqualTo(Boolean.TRUE);
- getMockEndpoint("mock:dead").message(0).header(Exchange.REDELIVERY_COUNTER).isEqualTo(3);
- getMockEndpoint("mock:dead").message(0).header(Exchange.REDELIVERY_MAX_COUNTER).isNull();
-
- 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();
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionSize(5).aggregationRepository(repo)
- .log("aggregated exchange id ${exchangeId} with ${body}")
- .to("mock:aggregated")
- .throwException(new IllegalArgumentException("Damn"))
- .to("mock:result")
- .end();
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverTest.java
deleted file mode 100644
index aca0470..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverTest.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateRecoverTest extends CamelTestSupport {
-
- private static AtomicInteger counter = new AtomicInteger();
- private LevelDBAggregationRepository repo;
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- // enable recovery
- repo.setUseRecovery(true);
- // check faster
- repo.setRecoveryInterval(500, TimeUnit.MILLISECONDS);
- super.setUp();
- }
-
- @Test
- public void testLevelDBAggregateRecover() throws Exception {
- // should fail the first 2 times and then recover
- getMockEndpoint("mock:aggregated").expectedMessageCount(3);
- 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(2);
-
- 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);
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionSize(5).aggregationRepository(repo)
- .log("aggregated exchange id ${exchangeId} with ${body}")
- .to("mock:aggregated")
- .delay(1000)
- // simulate errors the first two times
- .process(new Processor() {
- public void process(Exchange exchange) throws Exception {
- int count = counter.incrementAndGet();
- if (count <= 2) {
- throw new IllegalArgumentException("Damn");
- }
- }
- })
- .to("mock:result")
- .end();
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverWithRedeliveryPolicyTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverWithRedeliveryPolicyTest.java
deleted file mode 100644
index 0963a40..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverWithRedeliveryPolicyTest.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateRecoverWithRedeliveryPolicyTest extends CamelTestSupport {
-
- private static AtomicInteger counter = new AtomicInteger();
- private LevelDBAggregationRepository repo;
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- // enable recovery
- repo.setUseRecovery(true);
- // check faster
- repo.setRecoveryInterval(500, TimeUnit.MILLISECONDS);
- 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);
- getMockEndpoint("mock:result").message(0).header(Exchange.REDELIVERY_MAX_COUNTER).isNull();
-
- 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);
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- // CHECKSTYLE:OFF
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionSize(5).aggregationRepository(repo)
- // this is the output from the aggregator
- .log("aggregated exchange id ${exchangeId} with ${body}")
- .to("mock:aggregated")
- // simulate errors the first three times
- .process(new Processor() {
- public void process(Exchange exchange) throws Exception {
- int count = counter.incrementAndGet();
- if (count <= 3) {
- throw new IllegalArgumentException("Damn");
- }
- }
- })
- .to("mock:result")
- .end();
- // CHECKSTYLE:ON
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverWithSedaTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverWithSedaTest.java
deleted file mode 100644
index f401073..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateRecoverWithSedaTest.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateRecoverWithSedaTest extends CamelTestSupport {
-
- private static AtomicInteger counter = new AtomicInteger();
- private LevelDBAggregationRepository repo;
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- // enable recovery
- repo.setUseRecovery(true);
- // check faster
- repo.setRecoveryInterval(500, TimeUnit.MILLISECONDS);
- super.setUp();
- }
-
- @Test
- public void testLevelDBAggregateRecoverWithSeda() throws Exception {
- // should fail the first 2 times and then recover
- getMockEndpoint("mock:aggregated").expectedMessageCount(3);
- 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(2);
-
- 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);
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- .completionSize(5).aggregationRepository(repo)
- .log("aggregated exchange id ${exchangeId} with ${body}")
- .to("mock:aggregated")
- .to("seda:foo")
- .end();
-
- // should be able to recover when we send over SEDA as its a OnCompletion
- // which confirms the exchange when its complete.
- from("seda:foo")
- .delay(1000)
- // simulate errors the first two times
- .process(new Processor() {
- public void process(Exchange exchange) throws Exception {
- int count = counter.incrementAndGet();
- if (count <= 2) {
- throw new IllegalArgumentException("Damn");
- }
- }
- })
- .to("mock:result");
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateSerializedHeadersTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateSerializedHeadersTest.java
deleted file mode 100644
index 8bb06d4..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateSerializedHeadersTest.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * 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.HashMap;
-import java.util.Map;
-
-import org.apache.camel.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.component.util.HeaderDto;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateSerializedHeadersTest extends CamelTestSupport {
-
- private static final Logger LOG = LoggerFactory.getLogger(LevelDBAggregateSerializedHeadersTest.class);
- private static final int SIZE = 500;
- private LevelDBAggregationRepository repo;
-
- @BeforeEach
- @Override
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
- repo.setAllowSerializedHeaders(true);
- super.setUp();
- }
-
- @Test
- public void testLoadTestLevelDBAggregate() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result");
- mock.expectedMinimumMessageCount(1);
- mock.setResultWaitTime(50 * 1000);
-
- LOG.info("Staring to send " + SIZE + " messages.");
-
- for (int i = 0; i < SIZE; i++) {
- final int value = 1;
- HeaderDto headerDto = new HeaderDto("test", "company", 1);
- char id = 'A';
- LOG.debug("Sending {} with id {}", value, id);
- Map<String, Object> headers = new HashMap<>();
- headers.put("id", headerDto);
- template.sendBodyAndHeaders("seda:start?size=" + SIZE, value, headers);
- }
-
- LOG.info("Sending all " + SIZE + " message done. Now waiting for aggregation to complete.");
-
- assertMockEndpointsSatisfied();
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("seda:start?size=" + SIZE)
- .to("log:input?groupSize=500")
- .aggregate(header("id"), new MyAggregationStrategy())
- .aggregationRepository(repo)
- .completionSize(SIZE)
- .to("log:output?showHeaders=true")
- .to("mock:result")
- .end();
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
- if (oldExchange == null) {
- return newExchange;
- }
-
- Integer body1 = oldExchange.getIn().getBody(Integer.class);
- Integer body2 = newExchange.getIn().getBody(Integer.class);
- int sum = body1 + body2;
-
- oldExchange.getIn().setBody(sum);
- return oldExchange;
- }
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateTest.java
deleted file mode 100644
index b3b604a..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateTest.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateTest extends CamelTestSupport {
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- super.setUp();
- }
-
- @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);
-
- // from endpoint should be preserved
- assertEquals("direct://start", mock.getReceivedExchanges().get(0).getFromEndpoint().getEndpointUri());
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- // START SNIPPET: e1
- public void configure() throws Exception {
- // create the leveldb repo
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
-
- // here is the Camel route where we aggregate
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- // use our created leveldb repo as aggregation repository
- .completionSize(5).aggregationRepository(repo)
- .to("mock:aggregated");
- }
- // END SNIPPET: e1
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateTimeoutCompletionRestartTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateTimeoutCompletionRestartTest.java
deleted file mode 100644
index 775b644..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregateTimeoutCompletionRestartTest.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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 org.apache.camel.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregateTimeoutCompletionRestartTest extends CamelTestSupport {
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- deleteDirectory("target/data");
- super.setUp();
- }
-
- @Test
- public void testLevelDBAggregateTimeoutCompletionRestart() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:aggregated");
- mock.expectedMessageCount(0);
-
- template.sendBodyAndHeader("direct:start", "A", "id", 123);
- template.sendBodyAndHeader("direct:start", "B", "id", 123);
- template.sendBodyAndHeader("direct:start", "C", "id", 123);
-
- // stop Camel
- context.stop();
- assertEquals(0, mock.getReceivedCounter());
-
- // start Camel again, and the timeout should trigger a completion
- context.start();
-
- mock = getMockEndpoint("mock:aggregated");
- mock.expectedBodiesReceived("ABC");
-
- assertMockEndpointsSatisfied();
- assertEquals(1, mock.getReceivedCounter());
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- // create the leveldb repo
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository("repo1", "target/data/leveldb.dat");
-
- // here is the Camel route where we aggregate
- from("direct:start")
- .aggregate(header("id"), new MyAggregationStrategy())
- // use our created leveldb repo as aggregation repository
- .completionTimeout(3000).aggregationRepository(repo)
- .to("mock:aggregated");
- }
- };
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryAlotDataTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryAlotDataTest.java
deleted file mode 100644
index 87e23ca..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryAlotDataTest.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * 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.support.DefaultExchange;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregationRepositoryAlotDataTest extends CamelTestSupport {
-
- private LevelDBFile levelDBFile;
-
- @Override
- @BeforeEach
- 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
- @AfterEach
- public void tearDown() throws Exception {
- levelDBFile.stop();
- super.tearDown();
- }
-
- @Test
- public void testWithAlotOfDataSameKey() {
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository();
- repo.setLevelDBFile(levelDBFile);
- repo.setRepositoryName("repo1");
-
- for (int i = 0; i < 100; i++) {
- Exchange exchange1 = new DefaultExchange(context);
- exchange1.getIn().setBody("counter:" + i);
- repo.add(context, "foo", exchange1);
- }
-
- // Get it back..
- Exchange actual = repo.get(context, "foo");
- assertEquals("counter:99", actual.getIn().getBody());
- }
-
- @Test
- public void testWithAlotOfDataTwoKesy() {
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository();
- repo.setLevelDBFile(levelDBFile);
- repo.setRepositoryName("repo1");
-
- for (int i = 0; i < 100; i++) {
- Exchange exchange1 = new DefaultExchange(context);
- exchange1.getIn().setBody("counter:" + i);
- String key = i % 2 == 0 ? "foo" : "bar";
- repo.add(context, key, exchange1);
- }
-
- // Get it back..
- Exchange actual = repo.get(context, "foo");
- assertEquals("counter:98", actual.getIn().getBody());
-
- actual = repo.get(context, "bar");
- assertEquals("counter:99", actual.getIn().getBody());
- }
-
- @Test
- public void testWithAlotOfDataWithDifferentKesy() {
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository();
- repo.setLevelDBFile(levelDBFile);
- repo.setRepositoryName("repo1");
-
- for (int i = 0; i < 100; i++) {
- Exchange exchange1 = new DefaultExchange(context);
- exchange1.getIn().setBody("counter:" + i);
- String key = "key" + i;
- repo.add(context, key, exchange1);
- }
-
- // Get it back..
- for (int i = 0; i < 100; i++) {
- Exchange actual = repo.get(context, "key" + i);
- assertEquals("counter:" + i, actual.getIn().getBody());
- }
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryLoadExistingTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryLoadExistingTest.java
deleted file mode 100644
index 04bc1c5..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryLoadExistingTest.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * 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.support.DefaultExchange;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregationRepositoryLoadExistingTest extends CamelTestSupport {
-
- private LevelDBFile levelDBFile;
-
- @Override
- @BeforeEach
- 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
- @AfterEach
- public void tearDown() throws Exception {
- levelDBFile.stop();
- super.tearDown();
- }
-
- @Test
- public void testExisting() throws Exception {
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository();
- repo.setLevelDBFile(levelDBFile);
- repo.setRepositoryName("repo1");
- repo.setReturnOldExchange(true);
-
- // Store it..
- Exchange exchange1 = new DefaultExchange(context);
- exchange1.getIn().setBody("counter:1");
- Exchange actual = repo.add(context, "foo", exchange1);
- assertEquals(null, actual);
-
- // stop the repo
- levelDBFile.stop();
-
- Thread.sleep(1000);
-
- // load the repo again
- levelDBFile.start();
-
- // 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());
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryMultipleRepoTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryMultipleRepoTest.java
deleted file mode 100644
index 2c41f1b..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryMultipleRepoTest.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.support.DefaultExchange;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregationRepositoryMultipleRepoTest extends CamelTestSupport {
-
- private LevelDBFile levelDBFile;
-
- @Override
- @BeforeEach
- 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
- @AfterEach
- public void tearDown() throws Exception {
- levelDBFile.stop();
- super.tearDown();
- }
-
- @Test
- public void testMultipeRepo() {
- LevelDBAggregationRepository repo1 = new LevelDBAggregationRepository();
- repo1.setLevelDBFile(levelDBFile);
- repo1.setRepositoryName("repo1");
- repo1.setReturnOldExchange(true);
-
- LevelDBAggregationRepository repo2 = new LevelDBAggregationRepository();
- repo2.setLevelDBFile(levelDBFile);
- repo2.setRepositoryName("repo2");
- repo2.setReturnOldExchange(true);
-
- // Can't get something we have not put in...
- Exchange actual = repo1.get(context, "missing");
- assertEquals(null, actual);
-
- actual = repo2.get(context, "missing");
- assertEquals(null, actual);
-
- // Store it..
- Exchange exchange1 = new DefaultExchange(context);
- exchange1.getIn().setBody("counter:1");
- actual = repo1.add(context, "foo", exchange1);
- assertEquals(null, actual);
-
- // Get it back..
- actual = repo1.get(context, "foo");
- assertEquals("counter:1", actual.getIn().getBody());
- assertEquals(null, repo2.get(context, "foo"));
-
- // Change it..
- Exchange exchange2 = new DefaultExchange(context);
- exchange2.getIn().setBody("counter:2");
- actual = repo1.add(context, "foo", exchange2);
- // the old one
- assertEquals("counter:1", actual.getIn().getBody());
-
- // add to repo2
- Exchange exchange3 = new DefaultExchange(context);
- exchange3.getIn().setBody("Hello World");
- actual = repo2.add(context, "bar", exchange3);
- assertEquals(null, actual);
- assertEquals(null, repo1.get(context, "bar"));
-
- // Get it back..
- actual = repo1.get(context, "foo");
- assertEquals("counter:2", actual.getIn().getBody());
- assertEquals(null, repo2.get(context, "foo"));
-
- actual = repo2.get(context, "bar");
- assertEquals("Hello World", actual.getIn().getBody());
- assertEquals(null, repo1.get(context, "bar"));
- }
-
- @Test
- public void testMultipeRepoSameKeyDifferentContent() {
- LevelDBAggregationRepository repo1 = new LevelDBAggregationRepository();
- repo1.setLevelDBFile(levelDBFile);
- repo1.setRepositoryName("repo1");
-
- LevelDBAggregationRepository repo2 = new LevelDBAggregationRepository();
- repo2.setLevelDBFile(levelDBFile);
- repo2.setRepositoryName("repo2");
-
- Exchange exchange1 = new DefaultExchange(context);
- exchange1.getIn().setBody("Hello World");
- repo1.add(context, "foo", exchange1);
-
- Exchange exchange2 = new DefaultExchange(context);
- exchange2.getIn().setBody("Bye World");
- repo2.add(context, "foo", exchange2);
-
- Exchange actual = repo1.get(context, "foo");
- assertEquals("Hello World", actual.getIn().getBody());
- actual = repo2.get(context, "foo");
- assertEquals("Bye World", actual.getIn().getBody());
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryRecoverExistingTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryRecoverExistingTest.java
deleted file mode 100644
index 0c0c9f4..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryRecoverExistingTest.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * 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.support.DefaultExchange;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertNull;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregationRepositoryRecoverExistingTest extends CamelTestSupport {
-
- private LevelDBFile levelDBFile;
-
- @Override
- @BeforeEach
- public void setUp() throws Exception {
- super.setUp();
- deleteDirectory("target/data");
- File file = new File("target/data/leveldb.dat");
- levelDBFile = new LevelDBFile();
- levelDBFile.setFile(file);
- }
-
- @Test
- public void testExisting() throws Exception {
- LevelDBAggregationRepository repo = new LevelDBAggregationRepository();
- repo.setLevelDBFile(levelDBFile);
- repo.setRepositoryName("repo1");
- repo.setReturnOldExchange(true);
- repo.setUseRecovery(true);
- repo.start();
-
- // Store it..
- Exchange exchange1 = new DefaultExchange(context);
- exchange1.getIn().setBody("counter:1");
- Exchange actual = repo.add(context, "foo", exchange1);
- assertEquals(null, actual);
-
- // Remove it, which makes it in the pre confirm stage
- repo.remove(context, "foo", exchange1);
-
- String id = exchange1.getExchangeId();
-
- // stop the repo
- repo.stop();
-
- Thread.sleep(1000);
-
- // load the repo again
- repo.start();
-
- // Get it back..
- actual = repo.get(context, "foo");
- assertNull(actual);
-
- // Recover it
- actual = repo.recover(context, id);
- assertNotNull(actual);
- assertEquals("counter:1", actual.getIn().getBody());
-
- repo.stop();
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryTest.java
deleted file mode 100644
index d50f938..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBAggregationRepositoryTest.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * 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.support.DefaultExchange;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBAggregationRepositoryTest extends CamelTestSupport {
-
- private LevelDBFile levelDBFile;
-
- @Override
- @BeforeEach
- 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
- @AfterEach
- 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());
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBBigPayloadTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBBigPayloadTest.java
deleted file mode 100644
index eddc762..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBBigPayloadTest.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-/**
- * Test issue with leveldb file store growing to large
- */
-@Disabled("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;
- private Logger log = LoggerFactory.getLogger(getClass());
-
- @BeforeEach
- @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.exists(), file + " should exists");
- long size = file.length();
- log.info(file + " size is " + size);
- // should be about 32mb, so we say 34 just in case
- assertTrue(size < 34 * 1024 * 1024, file + " should not be so big in size, was: " + size);
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("timer:foo")
- .bean(BigPayload.class)
- .aggregate(method(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();
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBExchangeSerializationTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBExchangeSerializationTest.java
deleted file mode 100644
index 61c33ac..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBExchangeSerializationTest.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * 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.support.DefaultExchange;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertSame;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBExchangeSerializationTest extends CamelTestSupport {
-
- private LevelDBFile levelDBFile;
-
- @Override
- @BeforeEach
- 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
- @AfterEach
- 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());
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBGetNotFoundTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBGetNotFoundTest.java
deleted file mode 100644
index 59aa723..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBGetNotFoundTest.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.support.DefaultExchange;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertNull;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBGetNotFoundTest extends CamelTestSupport {
-
- private LevelDBFile levelDBFile;
- private Logger log = LoggerFactory.getLogger(getClass());
-
- @Override
- @BeforeEach
- 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
- @AfterEach
- 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(out, "Should not find exchange");
- }
-
- @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(out, "Should find exchange");
-
- Exchange exchange2 = new DefaultExchange(context);
- exchange2.getIn().setBody("Bye World");
- log.info("Created " + exchange2.getExchangeId());
-
- Exchange out2 = repo.get(context, exchange2.getExchangeId());
- assertNull(out2, "Should not find exchange");
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBSetupTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBSetupTest.java
deleted file mode 100644
index 45399d9..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBSetupTest.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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 org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBSetupTest extends CamelTestSupport {
- /**
- * The Level db file.
- */
- private LevelDBFile levelDBFile;
-
- @Override
- public void setUp() throws Exception {
- super.setUp();
- }
-
- @Override
- public void tearDown() throws Exception {
- deleteDirectory("leveldb.dat");
- super.tearDown();
- }
-
- /**
- * Test level db start with no path.
- */
- @Test
- public void testLevelDBStartWithNoPath() {
- Assertions.assertDoesNotThrow(() -> runTest("leveldb.dat"));
- }
-
- private void runTest(String fileName) {
- levelDBFile = new LevelDBFile();
- levelDBFile.setFileName(fileName);
- levelDBFile.start();
- levelDBFile.stop();
- }
-
- /**
- * Test level db start with path.
- */
- @Test
- public void testLevelDBStartWithPath() {
- deleteDirectory("target/data");
- Assertions.assertDoesNotThrow(() -> runTest("target/data/leveldb.dat"));
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.java
deleted file mode 100644
index b1cc768..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.test.spring.junit5.CamelSpringTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-import org.springframework.context.support.AbstractXmlApplicationContext;
-import org.springframework.context.support.ClassPathXmlApplicationContext;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest extends CamelSpringTestSupport {
-
- private static AtomicInteger counter = new AtomicInteger();
-
- @Override
- protected AbstractXmlApplicationContext createApplicationContext() {
- return new ClassPathXmlApplicationContext(
- "org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml");
- }
-
- @Override
- @BeforeEach
- 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 {
-
- @Override
- public void process(Exchange exchange) throws Exception {
- int count = counter.incrementAndGet();
- if (count <= 3) {
- throw new IllegalArgumentException("Damn");
- }
- }
- }
-
- public static class MyAggregationStrategy implements AggregationStrategy {
-
- @Override
- 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;
- }
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.java
deleted file mode 100644
index 51b8abe..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.AggregationStrategy;
-import org.apache.camel.Exchange;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.spring.junit5.CamelSpringTestSupport;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
-import org.junit.jupiter.api.condition.OS;
-import org.springframework.context.support.AbstractXmlApplicationContext;
-import org.springframework.context.support.ClassPathXmlApplicationContext;
-
-import static org.apache.camel.test.junit5.TestSupport.deleteDirectory;
-
-@DisabledOnOs({ OS.AIX, OS.OTHER })
-public class LevelDBSpringAggregateTest extends CamelSpringTestSupport {
-
- @Override
- @BeforeEach
- 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 {
-
- @Override
- 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;
- }
- }
-
-}
diff --git a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/util/HeaderDto.java b/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/util/HeaderDto.java
deleted file mode 100644
index e3de676..0000000
--- a/components/camel-leveldb-legacy/src/test/java/org/apache/camel/component/util/HeaderDto.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.util;
-
-import java.io.Serializable;
-
-public class HeaderDto implements Cloneable, Serializable {
- private String org;
- private String type;
- private int key;
-
- public HeaderDto(String org, String type, int key) {
- this.org = org;
- this.type = type;
- this.key = key;
- }
-
- public int getKey() {
- return key;
- }
-
- public void setKey(int key) {
- this.key = key;
- }
-
- public String getOrg() {
- return org;
- }
-
- public void setOrg(String org) {
- this.org = org;
- }
-
- public String getType() {
- return type;
- }
-
- public void setType(String type) {
- this.type = type;
- }
-
- @Override
- public String toString() {
- return "HeaderDto [org=" + org + ", type=" + type + ", key=" + key + "]";
- }
-}
diff --git a/components/camel-leveldb-legacy/src/test/resources/jndi.properties b/components/camel-leveldb-legacy/src/test/resources/jndi.properties
deleted file mode 100644
index 7675797..0000000
--- a/components/camel-leveldb-legacy/src/test/resources/jndi.properties
+++ /dev/null
@@ -1,18 +0,0 @@
-## ---------------------------------------------------------------------------
-## 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.support.jndi.CamelInitialContextFactory
diff --git a/components/camel-leveldb-legacy/src/test/resources/log4j2.properties b/components/camel-leveldb-legacy/src/test/resources/log4j2.properties
deleted file mode 100644
index 6f40432..0000000
--- a/components/camel-leveldb-legacy/src/test/resources/log4j2.properties
+++ /dev/null
@@ -1,28 +0,0 @@
-## ---------------------------------------------------------------------------
-## 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.
-## ---------------------------------------------------------------------------
-
-appender.file.type = File
-appender.file.name = file
-appender.file.fileName = target/camel-leveldb-test.log
-appender.file.layout.type = PatternLayout
-appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
-appender.out.type = Console
-appender.out.name = out
-appender.out.layout.type = PatternLayout
-appender.out.layout.pattern = %d [%-50.50t] %-5p %-30.30c{1} - %m%n
-rootLogger.level = INFO
-rootLogger.appenderRef.file.ref = file
diff --git a/components/camel-leveldb-legacy/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml b/components/camel-leveldb-legacy/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml
deleted file mode 100644
index 8fabcd3..0000000
--- a/components/camel-leveldb-legacy/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateRecoverWithRedeliveryPolicyTest.xml
+++ /dev/null
@@ -1,68 +0,0 @@
-<?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 aggregationStrategy="myAggregatorStrategy" aggregationRepository="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>
diff --git a/components/camel-leveldb-legacy/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml b/components/camel-leveldb-legacy/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml
deleted file mode 100644
index 6d1ac57..0000000
--- a/components/camel-leveldb-legacy/src/test/resources/org/apache/camel/component/leveldb/LevelDBSpringAggregateTest.xml
+++ /dev/null
@@ -1,57 +0,0 @@
-<?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 aggregationStrategy="myAggregatorStrategy" aggregationRepository="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>
diff --git a/components/pom.xml b/components/pom.xml
index 9cbd1db..793afae 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -218,7 +218,6 @@
<module>camel-kudu</module>
<module>camel-ldap</module>
<module>camel-ldif</module>
- <module>camel-leveldb-legacy</module>
<module>camel-leveldb</module>
<module>camel-lra</module>
<module>camel-lucene</module>
diff --git a/docs/components/modules/others/examples/json/leveldb-legacy.json b/docs/components/modules/others/examples/json/leveldb-legacy.json
deleted file mode 120000
index af72acc0..0000000
--- a/docs/components/modules/others/examples/json/leveldb-legacy.json
+++ /dev/null
@@ -1 +0,0 @@
-../../../../../../components/camel-leveldb-legacy/src/generated/resources/leveldb-legacy.json
\ No newline at end of file
diff --git a/docs/components/modules/others/nav.adoc b/docs/components/modules/others/nav.adoc
index 4399163..6d46753 100644
--- a/docs/components/modules/others/nav.adoc
+++ b/docs/components/modules/others/nav.adoc
@@ -32,7 +32,6 @@
** xref:jta.adoc[JTA]
** xref:knative-http.adoc[Knative Http]
** xref:leveldb.adoc[LevelDB]
-** xref:leveldb-legacy.adoc[LevelDB-legacy]
** xref:lra.adoc[LRA]
** xref:main.adoc[Main]
** xref:microprofile-config.adoc[Microprofile Config]
diff --git a/docs/components/modules/others/pages/leveldb-legacy.adoc b/docs/components/modules/others/pages/leveldb-legacy.adoc
deleted file mode 120000
index 90ccb3b..0000000
--- a/docs/components/modules/others/pages/leveldb-legacy.adoc
+++ /dev/null
@@ -1 +0,0 @@
-../../../../../components/camel-leveldb-legacy/src/main/docs/leveldb-legacy.adoc
\ No newline at end of file
diff --git a/parent/pom.xml b/parent/pom.xml
index 2f38c7b..eb3d862 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -1904,11 +1904,6 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
- <artifactId>camel-leveldb-legacy</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.camel</groupId>
<artifactId>camel-leveldb</artifactId>
<version>${project.version}</version>
</dependency>