You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2014/12/23 12:25:04 UTC
[4/5] camel git commit: CAMEL-2939: Fixed CS
CAMEL-2939: Fixed CS
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/e0b14cd4
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/e0b14cd4
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/e0b14cd4
Branch: refs/heads/master
Commit: e0b14cd4fa4875a17b899bfedb334a97e7c4eaf9
Parents: 7be546a
Author: Claus Ibsen <da...@apache.org>
Authored: Tue Dec 23 11:40:45 2014 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Tue Dec 23 11:40:45 2014 +0100
----------------------------------------------------------------------
.../component/cassandra/CassandraComponent.java | 31 ++-
.../component/cassandra/CassandraConstants.java | 20 +-
.../component/cassandra/CassandraConsumer.java | 18 +-
.../component/cassandra/CassandraEndpoint.java | 25 ++-
.../component/cassandra/CassandraProducer.java | 36 +++-
.../ResultSetConversionStrategies.java | 53 +++--
.../cassandra/ResultSetConversionStrategy.java | 14 +-
.../CassandraAggregationException.java | 16 +-
.../CassandraAggregationRepository.java | 83 +++++---
.../cassandra/CassandraCamelCodec.java | 22 +-
.../NamedCassandraAggregationRepository.java | 19 +-
.../CassandraIdempotentRepository.java | 36 ++--
.../NamedCassandraIdempotentRepository.java | 22 +-
.../utils/cassandra/CassandraSessionHolder.java | 19 +-
.../camel/utils/cassandra/CassandraUtils.java | 42 ++--
.../src/main/resources/META-INF/LICENSE.txt | 203 +++++++++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
.../services/org/apache/camel/component/cql | 17 ++
.../CassandraComponentBeanRefTest.java | 49 +++--
.../CassandraComponentClusterBuilderTest.java | 59 +++---
.../CassandraComponentConsumerTest.java | 52 +++--
.../CassandraComponentProducerTest.java | 63 ++++--
.../component/cassandra/CassandraUnitUtils.java | 23 ++-
.../ResultSetConversionStrategiesTest.java | 46 +++--
.../CassandraAggregationRepositoryTest.java | 104 ++++++----
.../cassandra/CassandraAggregationTest.java | 42 ++--
.../CassandraIdempotentRepositoryTest.java | 82 ++++----
.../cassandra/CassandraIdempotentTest.java | 31 +--
.../src/test/resources/log4j.properties | 44 +++-
29 files changed, 889 insertions(+), 393 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraComponent.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraComponent.java
index 761a5d7..9a1dac6 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraComponent.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraComponent.java
@@ -1,11 +1,27 @@
+/**
+ * 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.cassandra;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.Session;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Session;
import org.apache.camel.Endpoint;
import org.apache.camel.impl.DefaultComponent;
import org.apache.camel.util.EndpointHelper;
@@ -13,7 +29,7 @@ import org.apache.camel.util.EndpointHelper;
/**
* Represents the component that manages {@link CassandraEndpoint}. This
* component is based on Datastax Java Driver for Cassandra.
- *
+ * <p/>
* URI examples:
* <ul>
* <li>cql:localhost/keyspace</li>
@@ -46,8 +62,8 @@ public class CassandraComponent extends DefaultComponent {
*/
private static final Pattern HOSTS_PORT_KEYSPACE_PATTERN = Pattern.compile(
"^(" + HOSTS_PATTERN + ")?" // Hosts
- + "(?::(" + PORT_PATTERN + "))?" // Port
- + "(?:/(" + KEYSPACE_PATTERN + "))?$"); // Keyspace
+ + "(?::(" + PORT_PATTERN + "))?" // Port
+ + "(?:/(" + KEYSPACE_PATTERN + "))?$"); // Keyspace
/**
* Regular expression for parsing URI bean:sessionRef
*/
@@ -64,7 +80,7 @@ public class CassandraComponent extends DefaultComponent {
if (beanRefMatcher.matches()) {
String beanRefName = beanRefMatcher.group(1);
keyspace = beanRefMatcher.group(2);
- Object bean = EndpointHelper.resolveParameter(getCamelContext(), "#"+beanRefName, Object.class);
+ Object bean = EndpointHelper.resolveParameter(getCamelContext(), "#" + beanRefName, Object.class);
if (bean instanceof Session) {
session = (Session) bean;
cluster = session.getCluster();
@@ -86,6 +102,7 @@ public class CassandraComponent extends DefaultComponent {
setProperties(endpoint, parameters);
return endpoint;
}
+
/**
* Parse URI of the form cql://host1,host2:9042/keyspace and create a
* {@link Cluster.Builder}
@@ -123,7 +140,7 @@ public class CassandraComponent extends DefaultComponent {
// Username and password
String username = getAndRemoveOrResolveReferenceParameter(parameters, "username", String.class);
String password = getAndRemoveOrResolveReferenceParameter(parameters, "password", String.class);
- if (username != null && !username.isEmpty() && password!=null) {
+ if (username != null && !username.isEmpty() && password != null) {
clusterBuilder.withCredentials(username, password);
}
return clusterBuilder;
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConstants.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConstants.java
index 1c7722f..2abd513 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConstants.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConstants.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,16 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.component.cassandra;
/**
* Cassandra QL Endpoint constants
*/
-public class CassandraConstants {
+public final class CassandraConstants {
+
/**
* In Message header: CQL Query
*/
public static final String CQL_QUERY = "CamelCqlQuery";
+ private CassandraConstants() {
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConsumer.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConsumer.java
index 20b0812..952eae3 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConsumer.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraConsumer.java
@@ -1,3 +1,19 @@
+/**
+ * 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.cassandra;
import com.datastax.driver.core.PreparedStatement;
@@ -35,7 +51,7 @@ public class CassandraConsumer extends ScheduledPollConsumer {
preparedStatement = getEndpoint().prepareStatement();
}
ResultSet resultSet = session.execute(preparedStatement.bind());
-
+
// Create message from ResultSet
Exchange exchange = getEndpoint().createExchange();
Message message = exchange.getIn();
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraEndpoint.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraEndpoint.java
index c7f7b11..49d3a71 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraEndpoint.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraEndpoint.java
@@ -1,3 +1,19 @@
+/**
+ * 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.cassandra;
import com.datastax.driver.core.Cluster;
@@ -34,14 +50,15 @@ public class CassandraEndpoint extends DefaultEndpoint {
* How many rows should be retrieved in message body
*/
private ResultSetConversionStrategy resultSetConversionStrategy = ResultSetConversionStrategies.all();
+
/**
* Cassandra URI
*
* @param uri
* @param component Parent component
- * @param cluster Cluster (required)
- * @param session Session (optional)
- * @param keyspace Keyspace (optional)
+ * @param cluster Cluster (required)
+ * @param session Session (optional)
+ * @param keyspace Keyspace (optional)
*/
public CassandraEndpoint(String uri, CassandraComponent component, Cluster cluster, Session session, String keyspace) {
super(uri, component);
@@ -122,12 +139,14 @@ public class CassandraEndpoint extends DefaultEndpoint {
}
return preparedStatement;
}
+
/**
* Create and configure a Prepared CQL statement
*/
protected PreparedStatement prepareStatement() {
return prepareStatement(cql);
}
+
/**
* Copy ResultSet into Message.
*/
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraProducer.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraProducer.java
index ac429e0..9921191 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraProducer.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/CassandraProducer.java
@@ -1,10 +1,26 @@
+/**
+ * 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.cassandra;
-import com.datastax.driver.core.ConsistencyLevel;
+import java.util.Collection;
+
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
-import java.util.Collection;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.impl.DefaultProducer;
@@ -14,16 +30,17 @@ import org.slf4j.LoggerFactory;
/**
* Cassandra 2 CQL3 producer.
* <dl>
- * <dt>In Message</dt>
- * <dd>Bound parameters: Collection of Objects, Array of Objects, Simple Object<dd>
- * <dt>Out Message</dt>
- * <dd>List of all Rows<dd>
+ * <dt>In Message</dt>
+ * <dd>Bound parameters: Collection of Objects, Array of Objects, Simple Object<dd>
+ * <dt>Out Message</dt>
+ * <dd>List of all Rows<dd>
* <dl>
*/
public class CassandraProducer extends DefaultProducer {
private static final Logger LOG = LoggerFactory.getLogger(CassandraProducer.class);
private PreparedStatement preparedStatement;
+
public CassandraProducer(CassandraEndpoint endpoint) {
super(endpoint);
}
@@ -47,14 +64,15 @@ public class CassandraProducer extends DefaultProducer {
}
return cqlParams;
}
+
/**
* Execute CQL query using incoming message body has statement parameters.
*/
private ResultSet execute(Message message) {
String messageCql = message.getHeader(CassandraConstants.CQL_QUERY, String.class);
Object[] cqlParams = getCqlParams(message);
-
- ResultSet resultSet;
+
+ ResultSet resultSet;
PreparedStatement lPreparedStatement;
if (messageCql == null || messageCql.isEmpty()) {
// URI CQL
@@ -71,7 +89,7 @@ public class CassandraProducer extends DefaultProducer {
resultSet = session.execute(lPreparedStatement.bind());
} else {
resultSet = session.execute(lPreparedStatement.bind(cqlParams));
- }
+ }
return resultSet;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategies.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategies.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategies.java
index 51ed9c4..29aa7a2 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategies.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategies.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,27 +14,41 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.component.cassandra;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+
/**
* Implementations of {@link ResultSetConversionStrategy}
*/
-public class ResultSetConversionStrategies {
+public final class ResultSetConversionStrategies {
+
+ private static final Pattern LIMIT_NAME_PATTERN = Pattern.compile("^LIMIT_(\\d+)$");
+
private static final ResultSetConversionStrategy ALL = new ResultSetConversionStrategy() {
@Override
public Object getBody(ResultSet resultSet) {
return resultSet.all();
}
};
+
+ private static final ResultSetConversionStrategy ONE = new ResultSetConversionStrategy() {
+ @Override
+ public Object getBody(ResultSet resultSet) {
+ return resultSet.one();
+ }
+ };
+
+ private ResultSetConversionStrategies() {
+ }
+
/**
* Retrieve all rows.
* Message body contains a big list of {@link Row}s
@@ -41,12 +56,7 @@ public class ResultSetConversionStrategies {
public static ResultSetConversionStrategy all() {
return ALL;
}
- private static final ResultSetConversionStrategy ONE = new ResultSetConversionStrategy() {
- @Override
- public Object getBody(ResultSet resultSet) {
- return resultSet.one();
- }
- };
+
/**
* Retrieve a single row.
* Message body contains a single {@link Row}
@@ -54,8 +64,10 @@ public class ResultSetConversionStrategies {
public static ResultSetConversionStrategy one() {
return ONE;
}
+
private static class LimitResultSetConversionStrategy implements ResultSetConversionStrategy {
private final int rowMax;
+
public LimitResultSetConversionStrategy(int rowMax) {
this.rowMax = rowMax;
}
@@ -65,13 +77,14 @@ public class ResultSetConversionStrategies {
List<Row> rows = new ArrayList<Row>(rowMax);
int rowCount = 0;
Iterator<Row> rowIter = resultSet.iterator();
- while(rowIter.hasNext() && rowCount<rowMax) {
+ while (rowIter.hasNext() && rowCount < rowMax) {
rows.add(rowIter.next());
rowCount++;
}
return rows;
- }
+ }
}
+
/**
* Retrieve a limited list of rows.
* Message body contains a list of {@link Row} containing at most rowMax rows.
@@ -79,7 +92,7 @@ public class ResultSetConversionStrategies {
public static ResultSetConversionStrategy limit(int rowMax) {
return new LimitResultSetConversionStrategy(rowMax);
}
- private static final Pattern LIMIT_NAME_PATTERN=Pattern.compile("^LIMIT_(\\d+)$");
+
/**
* Get {@link ResultSetConversionStrategy} from String
*/
@@ -98,6 +111,6 @@ public class ResultSetConversionStrategies {
int limit = Integer.parseInt(matcher.group(1));
return limit(limit);
}
- throw new IllegalArgumentException("Unknown conversion strategy "+name);
+ throw new IllegalArgumentException("Unknown conversion strategy " + name);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategy.java b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategy.java
index 5874d5a..0bf8366 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategy.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/component/cassandra/ResultSetConversionStrategy.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,7 +14,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.component.cassandra;
import com.datastax.driver.core.ResultSet;
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationException.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationException.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationException.java
index cfc0049..459bafe 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationException.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationException.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,7 +14,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.processor.aggregate.cassandra;
import org.apache.camel.CamelExecutionException;
@@ -22,7 +22,7 @@ import org.apache.camel.Exchange;
/**
* Error occurred in Cassandra Aggregation repository
*/
-public class CassandraAggregationException extends CamelExecutionException{
+public class CassandraAggregationException extends CamelExecutionException {
public CassandraAggregationException(String message, Exchange exchange, Throwable cause) {
super(message, exchange, cause);
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepository.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepository.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepository.java
index db900ef..d075e59 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepository.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraAggregationRepository.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,19 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.processor.aggregate.cassandra;
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ConsistencyLevel;
-import com.datastax.driver.core.PreparedStatement;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.spi.AggregationRepository;
@@ -33,14 +34,20 @@ import org.apache.camel.support.ServiceSupport;
import org.apache.camel.utils.cassandra.CassandraSessionHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.camel.utils.cassandra.CassandraUtils.*;
+
+import static org.apache.camel.utils.cassandra.CassandraUtils.append;
+import static org.apache.camel.utils.cassandra.CassandraUtils.applyConsistencyLevel;
+import static org.apache.camel.utils.cassandra.CassandraUtils.concat;
+import static org.apache.camel.utils.cassandra.CassandraUtils.generateDelete;
+import static org.apache.camel.utils.cassandra.CassandraUtils.generateInsert;
+import static org.apache.camel.utils.cassandra.CassandraUtils.generateSelect;
/**
* Implementation of {@link AggregationRepository} using Cassandra table to store
* exchanges.
* Advice: use LeveledCompaction for this table and tune read/write consistency levels.
* Warning: Cassandra is not the best tool for queuing use cases
- * @see http://www.datastax.com/dev/blog/cassandra-anti-patterns-queues-and-queue-like-datasets
+ * See: http://www.datastax.com/dev/blog/cassandra-anti-patterns-queues-and-queue-like-datasets
*/
public abstract class CassandraAggregationRepository extends ServiceSupport implements AggregationRepository {
/**
@@ -58,11 +65,11 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
/**
* Exchange Id column name
*/
- private String exchangeIdColumn="EXCHANGE_ID";
+ private String exchangeIdColumn = "EXCHANGE_ID";
/**
* Exchange column name
*/
- private String exchangeColumn="EXCHANGE";
+ private String exchangeColumn = "EXCHANGE";
/**
* Primary key columns
*/
@@ -83,7 +90,7 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
* Read consistency level
*/
private ConsistencyLevel readConsistencyLevel;
-
+
private PreparedStatement insertStatement;
private PreparedStatement selectStatement;
private PreparedStatement deleteStatement;
@@ -102,31 +109,36 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
public CassandraAggregationRepository(Session session) {
this.sessionHolder = new CassandraSessionHolder(session);
}
+
public CassandraAggregationRepository(Cluster cluster, String keyspace) {
this.sessionHolder = new CassandraSessionHolder(cluster, keyspace);
}
+
/**
* Get fixed primary key values.
*/
protected abstract Object[] getPKValues();
+
/**
* Generate primary key values: fixed + aggregation key.
*/
protected Object[] getPKValues(String key) {
return append(getPKValues(), key);
}
+
/**
* Get aggregation key colum name.
*/
private String getKeyColumn() {
- return pkColumns[pkColumns.length-1];
+ return pkColumns[pkColumns.length - 1];
}
+
private String[] getAllColumns() {
return append(pkColumns, exchangeIdColumn, exchangeColumn);
}
//--------------------------------------------------------------------------
// Service support
-
+
@Override
protected void doStart() throws Exception {
sessionHolder.start();
@@ -146,12 +158,13 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
// Add exchange to repository
private void initInsertStatement() {
- String cql = generateInsert(table,
- getAllColumns(),
+ String cql = generateInsert(table,
+ getAllColumns(),
false, ttl).toString();
LOGGER.debug("Generated Insert {}", cql);
insertStatement = applyConsistencyLevel(getSession().prepare(cql), writeConsistencyLevel);
}
+
/**
* Insert or update exchange in aggregation table.
*/
@@ -173,12 +186,13 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
// Get exchange from repository
protected void initSelectStatement() {
- String cql = generateSelect(table,
- getAllColumns(),
+ String cql = generateSelect(table,
+ getAllColumns(),
pkColumns).toString();
LOGGER.debug("Generated Select {}", cql);
selectStatement = applyConsistencyLevel(getSession().prepare(cql), readConsistencyLevel);
}
+
/**
* Get exchange from aggregation table by aggregation key.
*/
@@ -188,7 +202,7 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
LOGGER.debug("Selecting key {} ", pkValues);
Row row = getSession().execute(selectStatement.bind(pkValues)).one();
Exchange exchange = null;
- if (row!=null) {
+ if (row != null) {
try {
exchange = exchangeCodec.unmarshallExchange(camelContext, row.getBytes(exchangeColumn));
} catch (IOException iOException) {
@@ -216,10 +230,10 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
@Override
public void confirm(CamelContext camelContext, String exchangeId) {
Object[] pkValues = getPKValues();
- String keyColumn= getKeyColumn();
+ String keyColumn = getKeyColumn();
LOGGER.debug("Selecting Ids {} ", pkValues);
List<Row> rows = selectKeyIds();
- for(Row row:rows) {
+ for (Row row : rows) {
if (row.getString(exchangeIdColumn).equals(exchangeId)) {
String key = row.getString(keyColumn);
Object[] cqlParams = append(pkValues, key, exchangeId);
@@ -247,11 +261,12 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
LOGGER.debug("Deleting key {}", (Object) idValues);
getSession().execute(deleteStatement.bind(idValues));
}
+
// -------------------------------------------------------------------------
- private void initSelectKeyIdStatement() {
- String cql = generateSelect(table,
+ private void initSelectKeyIdStatement() {
+ String cql = generateSelect(table,
new String[]{getKeyColumn(), exchangeIdColumn}, // Key + Exchange Id columns
- pkColumns, pkColumns.length-1).toString(); // Where fixed PK columns
+ pkColumns, pkColumns.length - 1).toString(); // Where fixed PK columns
LOGGER.debug("Generated Select keys {}", cql);
selectKeyIdStatement = applyConsistencyLevel(getSession().prepare(cql), readConsistencyLevel);
}
@@ -270,7 +285,7 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
List<Row> rows = selectKeyIds();
Set<String> keys = new HashSet<String>(rows.size());
String keyColumnName = getPKColumns()[1];
- for(Row row:rows) {
+ for (Row row : rows) {
keys.add(row.getString(keyColumnName));
}
return keys;
@@ -295,10 +310,12 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
public void setTable(String table) {
this.table = table;
}
+
public String[] getPKColumns() {
return pkColumns;
}
- public void setPKColumns(String ... pkColumns) {
+
+ public void setPKColumns(String... pkColumns) {
this.pkColumns = pkColumns;
}
@@ -325,7 +342,7 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
public void setReadConsistencyLevel(ConsistencyLevel readConsistencyLevel) {
this.readConsistencyLevel = readConsistencyLevel;
}
-
+
public String getExchangeColumn() {
return exchangeColumn;
}
@@ -341,5 +358,5 @@ public abstract class CassandraAggregationRepository extends ServiceSupport impl
public void setTtl(Integer ttl) {
this.ttl = ttl;
}
-
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraCamelCodec.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraCamelCodec.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraCamelCodec.java
index e060678..af5b2e5 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraCamelCodec.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/CassandraCamelCodec.java
@@ -1,18 +1,18 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with this
- * work for additional information regarding copyright ownership. The ASF
- * licenses this file to You under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * 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
+ * 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.
+ * 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.processor.aggregate.cassandra;
@@ -31,7 +31,7 @@ import org.apache.camel.impl.DefaultExchangeHolder;
/**
* Marshall/unmarshall Exchange to/from a ByteBuffer.
- * Inspired from JdbcCamelCodec. TODO Find how to share code
+ * Inspired from JdbcCamelCodec.
*/
public class CassandraCamelCodec {
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepository.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepository.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepository.java
index 5448207..347ed75 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepository.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/aggregate/cassandra/NamedCassandraAggregationRepository.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,14 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.processor.aggregate.cassandra;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
/**
- * Concrete implementation of {@link CassandraAggregationRepository} using 2
+ * Concrete implementation of {@link CassandraAggregationRepository} using 2
* columns as primary key: name (partition key) and key (clustering key).
*/
public class NamedCassandraAggregationRepository extends CassandraAggregationRepository {
@@ -28,14 +28,17 @@ public class NamedCassandraAggregationRepository extends CassandraAggregationRep
* Aggregation repository name
*/
private String name;
+
public NamedCassandraAggregationRepository() {
setPKColumns("NAME", "KEY");
}
+
public NamedCassandraAggregationRepository(Session session, String name) {
super(session);
this.name = name;
setPKColumns("NAME", "KEY");
}
+
public NamedCassandraAggregationRepository(Cluster cluster, String keyspace, String name) {
super(cluster, keyspace);
this.name = name;
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepository.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepository.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepository.java
index 4bdf496..ddb7168 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepository.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/CassandraIdempotentRepository.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -26,14 +27,19 @@ import org.apache.camel.support.ServiceSupport;
import org.apache.camel.utils.cassandra.CassandraSessionHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.camel.utils.cassandra.CassandraUtils.*;
+
+import static org.apache.camel.utils.cassandra.CassandraUtils.applyConsistencyLevel;
+import static org.apache.camel.utils.cassandra.CassandraUtils.generateDelete;
+import static org.apache.camel.utils.cassandra.CassandraUtils.generateInsert;
+import static org.apache.camel.utils.cassandra.CassandraUtils.generateSelect;
/**
* Implementation of {@link IdempotentRepository} using Cassandra table to store
* message ids.
* Advice: use LeveledCompaction for this table and tune read/write consistency levels.
* Warning: Cassandra is not the best tool for queuing use cases
- * @see http://www.datastax.com/dev/blog/cassandra-anti-patterns-queues-and-queue-like-datasets
+ * See http://www.datastax.com/dev/blog/cassandra-anti-patterns-queues-and-queue-like-datasets
+ *
* @param <K> Message Id
*/
public abstract class CassandraIdempotentRepository<K> extends ServiceSupport implements IdempotentRepository<K> {
@@ -75,20 +81,22 @@ public abstract class CassandraIdempotentRepository<K> extends ServiceSupport im
public CassandraIdempotentRepository(Session session) {
this.sessionHolder = new CassandraSessionHolder(session);
}
+
public CassandraIdempotentRepository(Cluster cluster, String keyspace) {
this.sessionHolder = new CassandraSessionHolder(cluster, keyspace);
}
private boolean isKey(ResultSet resultSet) {
- Row row = resultSet.one();
- if (row==null) {
+ Row row = resultSet.one();
+ if (row == null) {
LOGGER.debug("No row to check key");
return false;
} else {
- LOGGER.debug("Row with {} columns to check key", row.getColumnDefinitions());
- return row.getColumnDefinitions().size()>1;
+ LOGGER.debug("Row with {} columns to check key", row.getColumnDefinitions());
+ return row.getColumnDefinitions().size() > 1;
}
}
+
protected abstract Object[] getPKValues(K key);
// -------------------------------------------------------------------------
// Lifecycle methods
@@ -182,7 +190,7 @@ public abstract class CassandraIdempotentRepository<K> extends ServiceSupport im
}
public void setPKColumns(String... pkColumns) {
- this.pkColumns=pkColumns;
+ this.pkColumns = pkColumns;
}
public Integer getTtl() {
@@ -208,5 +216,5 @@ public abstract class CassandraIdempotentRepository<K> extends ServiceSupport im
public void setReadConsistencyLevel(ConsistencyLevel readConsistencyLevel) {
this.readConsistencyLevel = readConsistencyLevel;
}
-
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepository.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepository.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepository.java
index 03d73ec..73219fc 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepository.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/idempotent/cassandra/NamedCassandraIdempotentRepository.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,17 +14,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.processor.idempotent.cassandra;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
/**
- * Concrete implementation of {@link CassandraIdempotentRepository} using 2
+ * Concrete implementation of {@link CassandraIdempotentRepository} using 2
* columns as primary key: name (partition key) and key (clustering key).
*/
-public class NamedCassandraIdempotentRepository<K> extends CassandraIdempotentRepository<K>{
+public class NamedCassandraIdempotentRepository<K> extends CassandraIdempotentRepository<K> {
/**
* Idempotent repository name
*/
@@ -32,17 +32,19 @@ public class NamedCassandraIdempotentRepository<K> extends CassandraIdempotentRe
public NamedCassandraIdempotentRepository() {
setPKColumns("NAME", "KEY");
}
+
public NamedCassandraIdempotentRepository(Session session, String name) {
super(session);
this.name = name;
setPKColumns("NAME", "KEY");
}
+
public NamedCassandraIdempotentRepository(Cluster cluster, String keyspace, String name) {
super(cluster, keyspace);
this.name = name;
setPKColumns("NAME", "KEY");
}
-
+
@Override
protected Object[] getPKValues(K key) {
return new Object[]{name, key};
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraSessionHolder.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraSessionHolder.java b/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraSessionHolder.java
index 82a7a72..d16c3fa 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraSessionHolder.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraSessionHolder.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,7 +14,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.utils.cassandra;
import com.datastax.driver.core.Cluster;
@@ -52,6 +52,7 @@ public class CassandraSessionHolder {
this.keyspace = session.getLoggedKeyspace();
this.managedSession = false;
}
+
public void start() {
if (managedSession) {
if (keyspace == null) {
@@ -61,12 +62,14 @@ public class CassandraSessionHolder {
}
}
}
+
public void stop() {
if (!managedSession) {
session.close();
session = null;
}
}
+
public Session getSession() {
return session;
}
@@ -78,5 +81,5 @@ public class CassandraSessionHolder {
public String getKeyspace() {
return keyspace;
}
-
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraUtils.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraUtils.java b/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraUtils.java
index 4b18d8c..defc169 100644
--- a/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraUtils.java
+++ b/components/camel-cassandraql/src/main/java/org/apache/camel/utils/cassandra/CassandraUtils.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,7 +14,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.utils.cassandra;
import com.datastax.driver.core.ConsistencyLevel;
@@ -27,41 +27,46 @@ public class CassandraUtils {
* Apply consistency level if provided, else leave default.
*/
public static PreparedStatement applyConsistencyLevel(PreparedStatement statement, ConsistencyLevel consistencyLevel) {
- if (consistencyLevel!=null) {
+ if (consistencyLevel != null) {
statement.setConsistencyLevel(consistencyLevel);
}
return statement;
}
+
/**
* Concatenate 2 arrays.
*/
public static Object[] concat(Object[] array1, Object[] array2) {
- Object[] array = new Object[array1.length+array2.length];
+ Object[] array = new Object[array1.length + array2.length];
System.arraycopy(array1, 0, array, 0, array1.length);
System.arraycopy(array2, 0, array, array1.length, array2.length);
return array;
}
+
/**
* Concatenate 2 arrays.
*/
public static String[] concat(String[] array1, String[] array2) {
- String[] array = new String[array1.length+array2.length];
+ String[] array = new String[array1.length + array2.length];
System.arraycopy(array1, 0, array, 0, array1.length);
System.arraycopy(array2, 0, array, array1.length, array2.length);
return array;
}
+
/**
* Append values to given array.
*/
- public static Object[] append(Object[] array1, Object ... array2) {
+ public static Object[] append(Object[] array1, Object... array2) {
return concat(array1, array2);
}
+
/**
* Append values to given array.
*/
- public static String[] append(String[] array1, String ... array2) {
+ public static String[] append(String[] array1, String... array2) {
return concat(array1, array2);
}
+
/**
* Append columns to CQL.
*/
@@ -77,7 +82,7 @@ public class CassandraUtils {
/**
* Append columns to CQL.
*/
- public static void appendColumns(StringBuilder cqlBuilder, String[] columns, String sep) {
+ public static void appendColumns(StringBuilder cqlBuilder, String[] columns, String sep) {
appendColumns(cqlBuilder, columns, sep, columns.length);
}
@@ -89,12 +94,14 @@ public class CassandraUtils {
appendColumns(cqlBuilder, columns, "=? and ", maxColumnIndex);
cqlBuilder.append("=?");
}
+
/**
* Append where columns = ? to CQL.
*/
public void appendWhere(StringBuilder cqlBuilder, String[] columns) {
appendWhere(cqlBuilder, columns, columns.length);
}
+
/**
* Append ?,? to CQL.
*/
@@ -120,11 +127,12 @@ public class CassandraUtils {
if (ifNotExists) {
cqlBuilder.append(" if not exists");
}
- if (ttl!=null) {
+ if (ttl != null) {
cqlBuilder.append(" using ttl=").append(ttl);
}
- return cqlBuilder;
+ return cqlBuilder;
}
+
/**
* Generate select where columns = ? CQL.
*/
@@ -142,18 +150,20 @@ public class CassandraUtils {
appendWhere(cqlBuilder, whereColumns, whereColumnsMaxIndex);
return cqlBuilder;
}
+
/**
* Generate delete where columns = ? CQL.
*/
public static StringBuilder generateDelete(String table, String[] whereColumns, boolean ifExists) {
return generateDelete(table, whereColumns, whereColumns.length, ifExists);
}
+
/**
* Generate delete where columns = ? CQL.
*/
public static StringBuilder generateDelete(String table, String[] whereColumns, int whereColumnsMaxIndex, boolean ifExists) {
StringBuilder cqlBuilder = new StringBuilder("delete from ")
- .append(table);
+ .append(table);
appendWhere(cqlBuilder, whereColumns, whereColumnsMaxIndex);
if (ifExists) {
cqlBuilder.append(" if exists");
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/resources/META-INF/LICENSE.txt
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/resources/META-INF/LICENSE.txt b/components/camel-cassandraql/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/components/camel-cassandraql/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
+
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/resources/META-INF/NOTICE.txt
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/resources/META-INF/NOTICE.txt b/components/camel-cassandraql/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/components/camel-cassandraql/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/main/resources/META-INF/services/org/apache/camel/component/cql
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/main/resources/META-INF/services/org/apache/camel/component/cql b/components/camel-cassandraql/src/main/resources/META-INF/services/org/apache/camel/component/cql
index 8d80a4f..66c5093 100644
--- a/components/camel-cassandraql/src/main/resources/META-INF/services/org/apache/camel/component/cql
+++ b/components/camel-cassandraql/src/main/resources/META-INF/services/org/apache/camel/component/cql
@@ -1 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
class=org.apache.camel.component.cassandra.CassandraComponent
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentBeanRefTest.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentBeanRefTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentBeanRefTest.java
index f3efc27..a184401 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentBeanRefTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentBeanRefTest.java
@@ -1,18 +1,27 @@
+/**
+ * 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.cassandra;
-import org.apache.camel.component.cassandra.CassandraEndpoint;
import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import java.net.URLEncoder;
-import java.util.Arrays;
import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.JndiRegistry;
import org.apache.camel.test.junit4.CamelTestSupport;
-import org.apache.http.client.utils.URLEncodedUtils;
import org.cassandraunit.CassandraCQLUnit;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -20,14 +29,21 @@ import org.junit.Rule;
import org.junit.Test;
public class CassandraComponentBeanRefTest extends CamelTestSupport {
- @Rule
- public CassandraCQLUnit cassandra=CassandraUnitUtils.cassandraCQLUnit();
+ public static final String CQL = "insert into camel_user(login, first_name, last_name) values (?, ?, ?)";
+ public static final String SESSION_URI = "cql:bean:cassandraSession?cql=#insertCql";
+ public static final String CLUSTER_URI = "cql:bean:cassandraCluster/camel_ks?cql=#insertCql";
+
@Produce(uri = "direct:input")
- ProducerTemplate producerTemplate;
+ public ProducerTemplate producerTemplate;
+
+ @Rule
+ public CassandraCQLUnit cassandra = CassandraUnitUtils.cassandraCQLUnit();
+
@BeforeClass
public static void setUpClass() throws Exception {
CassandraUnitUtils.startEmbeddedCassandra();
}
+
@AfterClass
public static void tearDownClass() throws Exception {
CassandraUnitUtils.cleanEmbeddedCassandra();
@@ -44,21 +60,19 @@ public class CassandraComponentBeanRefTest extends CamelTestSupport {
registry.bind("insertCql", CQL);
return registry;
}
- public static final String CQL = "insert into camel_user(login, first_name, last_name) values (?, ?, ?)";
- public static final String SESSION_URI = "cql:bean:cassandraSession?cql=#insertCql";
- public static final String CLUSTER_URI = "cql:bean:cassandraCluster/camel_ks?cql=#insertCql";
-
+
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
- public void configure() {
+ public void configure() {
from("direct:inputSession")
- .to(SESSION_URI);
+ .to(SESSION_URI);
from("direct:inputCluster")
- .to(CLUSTER_URI);
+ .to(CLUSTER_URI);
}
};
}
+
@Test
public void testSession() throws Exception {
CassandraEndpoint endpoint = getMandatoryEndpoint(SESSION_URI, CassandraEndpoint.class);
@@ -66,6 +80,7 @@ public class CassandraComponentBeanRefTest extends CamelTestSupport {
assertEquals("camel_ks", endpoint.getKeyspace());
assertEquals(CQL, endpoint.getCql());
}
+
@Test
public void testCluster() throws Exception {
CassandraEndpoint endpoint = getMandatoryEndpoint(CLUSTER_URI, CassandraEndpoint.class);
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentClusterBuilderTest.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentClusterBuilderTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentClusterBuilderTest.java
index 17753f7..826cd17 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentClusterBuilderTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentClusterBuilderTest.java
@@ -1,9 +1,10 @@
-/*
- * Copyright 2014 The Apache Software Foundation.
- *
- * Licensed 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
+/**
+ * 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
*
@@ -13,62 +14,68 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.component.cassandra;
-import org.apache.camel.component.cassandra.CassandraComponent;
+import java.util.HashMap;
+import java.util.Map;
+
import com.datastax.driver.core.Cluster;
-import java.util.*;
import org.apache.camel.impl.DefaultCamelContext;
-import static org.hamcrest.Matchers.*;
-import org.junit.Test;
-import static org.junit.Assert.*;
import org.junit.Before;
+import org.junit.Test;
+
+import static org.hamcrest.Matchers.isOneOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
/**
* Unit test for {@link CassandraComponent}
*/
public class CassandraComponentClusterBuilderTest {
private final CassandraComponent component = new CassandraComponent();
+
@Before
public void setUp() {
component.setCamelContext(new DefaultCamelContext());
}
+
@Test
- public void testClusterBuilder_Basic() {
+ public void testClusterBuilderBasic() {
Map<String, Object> params = new HashMap<String, Object>();
params.put("clusterName", "cluster");
Cluster.Builder clusterBuilder = component.clusterBuilder("127.0.0.1,127.0.0.2/keyspace", params);
-
+
assertEquals(2, clusterBuilder.getContactPoints().size());
- assertThat(clusterBuilder.getContactPoints().get(0).getHostName(), isOneOf("127.0.0.1","localhost"));
- assertThat(clusterBuilder.getContactPoints().get(1).getHostName(), isOneOf("127.0.0.2","localhost"));
+ assertThat(clusterBuilder.getContactPoints().get(0).getHostName(), isOneOf("127.0.0.1", "localhost"));
+ assertThat(clusterBuilder.getContactPoints().get(1).getHostName(), isOneOf("127.0.0.2", "localhost"));
assertEquals("cluster", clusterBuilder.getClusterName());
- assertEquals("keyspace",params.get("keyspace"));
+ assertEquals("keyspace", params.get("keyspace"));
}
+
@Test
- public void testClusterBuilder_Port() {
+ public void testClusterBuilderPort() {
Map<String, Object> params = new HashMap<String, Object>();
params.put("clusterName", "cluster");
Cluster.Builder clusterBuilder = component.clusterBuilder("127.0.0.1,127.0.0.2:1234/keyspace", params);
-
+
assertEquals(2, clusterBuilder.getContactPoints().size());
- assertThat(clusterBuilder.getContactPoints().get(0).getHostName(), isOneOf("127.0.0.1","localhost"));
+ assertThat(clusterBuilder.getContactPoints().get(0).getHostName(), isOneOf("127.0.0.1", "localhost"));
assertEquals(1234, clusterBuilder.getContactPoints().get(0).getPort());
- assertThat(clusterBuilder.getContactPoints().get(1).getHostName(), isOneOf("127.0.0.2","localhost"));
+ assertThat(clusterBuilder.getContactPoints().get(1).getHostName(), isOneOf("127.0.0.2", "localhost"));
assertEquals(1234, clusterBuilder.getConfiguration().getProtocolOptions().getPort());
assertEquals("cluster", clusterBuilder.getClusterName());
- assertEquals("keyspace",params.get("keyspace"));
+ assertEquals("keyspace", params.get("keyspace"));
}
@Test
- public void testClusterBuilder_Simplest() {
+ public void testClusterBuilderSimplest() {
Map<String, Object> params = new HashMap<String, Object>();
Cluster.Builder clusterBuilder = component.clusterBuilder("127.0.0.1", params);
-
+
assertEquals(1, clusterBuilder.getContactPoints().size());
- assertThat(clusterBuilder.getContactPoints().get(0).getHostName(), isOneOf("127.0.0.1","localhost"));
+ assertThat(clusterBuilder.getContactPoints().get(0).getHostName(), isOneOf("127.0.0.1", "localhost"));
assertNull(params.get("keyspace"));
}
-
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e0b14cd4/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentConsumerTest.java b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentConsumerTest.java
index 0b4f0ed..76d303d 100644
--- a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentConsumerTest.java
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentConsumerTest.java
@@ -1,8 +1,25 @@
+/**
+ * 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.cassandra;
-import com.datastax.driver.core.Row;
import java.util.List;
import java.util.concurrent.TimeUnit;
+
+import com.datastax.driver.core.Row;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
@@ -10,7 +27,6 @@ import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.cassandraunit.CassandraCQLUnit;
import org.junit.AfterClass;
-import static org.junit.Assert.assertTrue;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
@@ -18,21 +34,26 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CassandraComponentConsumerTest extends CamelTestSupport {
+
+ private static final String CQL = "select login, first_name, last_name from camel_user";
+
@Rule
- public CassandraCQLUnit cassandra=CassandraUnitUtils.cassandraCQLUnit();
- private final Logger logger = LoggerFactory.getLogger(getClass());
+ public CassandraCQLUnit cassandra = CassandraUnitUtils.cassandraCQLUnit();
+
@BeforeClass
public static void setUpClass() throws Exception {
CassandraUnitUtils.startEmbeddedCassandra();
}
+
@AfterClass
public static void tearDownClass() throws Exception {
CassandraUnitUtils.cleanEmbeddedCassandra();
}
+
@Test
- public void testConsume_All() throws Exception {
+ public void testConsumeAll() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:resultAll");
- mock.expectedMinimumMessageCount(1);
+ mock.expectedMinimumMessageCount(1);
mock.whenAnyExchangeReceived(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
@@ -43,10 +64,11 @@ public class CassandraComponentConsumerTest extends CamelTestSupport {
mock.await(1, TimeUnit.SECONDS);
assertMockEndpointsSatisfied();
}
+
@Test
- public void testConsume_One() throws Exception {
+ public void testConsumeOne() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:resultOne");
- mock.expectedMinimumMessageCount(1);
+ mock.expectedMinimumMessageCount(1);
mock.whenAnyExchangeReceived(new Processor() {
@Override
public void process(Exchange exchange) throws Exception {
@@ -55,18 +77,18 @@ public class CassandraComponentConsumerTest extends CamelTestSupport {
}
});
mock.await(1, TimeUnit.SECONDS);
-
+
assertMockEndpointsSatisfied();
}
- private static final String CQL = "select login, first_name, last_name from camel_user";
+
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
- public void configure() {
- from("cql://localhost/camel_ks?cql="+CQL)
- .to("mock:resultAll");
- from("cql://localhost/camel_ks?cql="+CQL+"&resultSetConversionStrategy=ONE")
- .to("mock:resultOne");
+ public void configure() {
+ from("cql://localhost/camel_ks?cql=" + CQL)
+ .to("mock:resultAll");
+ from("cql://localhost/camel_ks?cql=" + CQL + "&resultSetConversionStrategy=ONE")
+ .to("mock:resultOne");
}
};
}