camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/9] camel git commit: CAMEL-2939: Allow to use unprepared statements
Date Wed, 18 Feb 2015 09:11:29 GMT
Repository: camel
Updated Branches:
  refs/heads/master 69934ac79 -> db5e6be5a


CAMEL-2939: Allow to use unprepared statements


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/742eda6b
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/742eda6b
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/742eda6b

Branch: refs/heads/master
Commit: 742eda6b021f67a63fd7bdae83d9c3d4fda7faa5
Parents: 0592f0c
Author: Gerald Quintana <gerald.quintana@zenika.com>
Authored: Wed Jan 14 17:54:08 2015 +0100
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Wed Feb 18 10:00:48 2015 +0100

----------------------------------------------------------------------
 .../component/cassandra/CassandraConsumer.java  |  21 +++-
 .../component/cassandra/CassandraEndpoint.java  |  13 ++
 .../component/cassandra/CassandraProducer.java  |  59 ++++++++-
 .../CassandraComponentConsumerTest.java         |  17 +++
 ...assandraComponentProducerUnpreparedTest.java | 119 +++++++++++++++++++
 5 files changed, 220 insertions(+), 9 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/742eda6b/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 168a7c2..fdad0ae 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
@@ -47,7 +47,12 @@ public class CassandraConsumer extends ScheduledPollConsumer {
     protected int poll() throws Exception {
         // Execute CQL Query
         Session session = getEndpoint().getSessionHolder().getSession();
-        ResultSet resultSet = session.execute(preparedStatement.bind());
+        ResultSet resultSet;
+        if (isPrepareStatements()) {
+            resultSet = session.execute(preparedStatement.bind());
+        } else {
+            resultSet = session.execute(getEndpoint().getCql());
+        }
 
         // Create message from ResultSet
         Exchange exchange = getEndpoint().createExchange();
@@ -69,9 +74,19 @@ public class CassandraConsumer extends ScheduledPollConsumer {
     @Override
     protected void doStart() throws Exception {
         super.doStart();
-
-        if (preparedStatement == null) {
+        if (isPrepareStatements()) {
             preparedStatement = getEndpoint().prepareStatement();
         }
     }
+
+    @Override
+    protected void doStop() throws Exception {
+        this.preparedStatement = null;
+        super.doStop();
+    }
+
+    public boolean isPrepareStatements() {
+        return getEndpoint().isPrepareStatements();
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/742eda6b/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 5d6ad1f..8d70b9f 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
@@ -51,6 +51,11 @@ public class CassandraEndpoint extends DefaultEndpoint {
     private String keyspace;
     @UriParam
     private String cql;
+    /**
+     * Use PreparedStatements or normal Statements
+     */
+    @UriParam
+    private boolean prepareStatements=true;
     @UriParam
     private String clusterName;
     @UriParam
@@ -282,4 +287,12 @@ public class CassandraEndpoint extends DefaultEndpoint {
     public void setResultSetConversionStrategy(ResultSetConversionStrategy resultSetConversionStrategy)
{
         this.resultSetConversionStrategy = resultSetConversionStrategy;
     }
+
+    public boolean isPrepareStatements() {
+        return prepareStatements;
+    }
+
+    public void setPrepareStatements(boolean prepareStatements) {
+        this.prepareStatements = prepareStatements;
+    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/742eda6b/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 824d0eb..4d1d39b 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
@@ -46,10 +46,26 @@ public class CassandraProducer extends DefaultProducer {
     }
 
     @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+        if (isPrepareStatements()) {
+            this.preparedStatement = getEndpoint().prepareStatement();
+        }
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        this.preparedStatement = null;
+        super.doStop();
+    }
+
+    @Override
     public CassandraEndpoint getEndpoint() {
         return (CassandraEndpoint) super.getEndpoint();
     }
-
+    public boolean isPrepareStatements() {
+        return getEndpoint().isPrepareStatements();
+    }
     private Object[] getCqlParams(Message message) {
         Object cqlParamsObj = message.getBody();
         Object[] cqlParams;
@@ -75,19 +91,29 @@ public class CassandraProducer extends DefaultProducer {
         Object[] cqlParams = getCqlParams(message);
 
         ResultSet resultSet;
+        Session session = getEndpoint().getSessionHolder().getSession();
+        if (isPrepareStatements()) {
+            resultSet = executePreparedStatement(session, messageCql, cqlParams);
+        } else {
+            resultSet = executeStatement(session, messageCql, cqlParams);
+        }
+        return resultSet;
+    }
+
+    /**
+     * Execute CQL as PreparedStatement
+     */
+    private ResultSet executePreparedStatement(Session session, String messageCql, Object[]
cqlParams) {
+        ResultSet resultSet;
         PreparedStatement lPreparedStatement;
         if (messageCql == null || messageCql.isEmpty()) {
             // URI CQL
-            if (preparedStatement == null) {
-                this.preparedStatement = getEndpoint().prepareStatement();
-            }
             lPreparedStatement = this.preparedStatement;
         } else {
             // Message CQL
             lPreparedStatement = getEndpoint().prepareStatement(messageCql);
         }
-        Session session = getEndpoint().getSessionHolder().getSession();
-        if (cqlParams == null || cqlParams.length==0) {
+        if (cqlParams == null || cqlParams.length == 0) {
             resultSet = session.execute(lPreparedStatement.bind());
         } else {
             resultSet = session.execute(lPreparedStatement.bind(cqlParams));
@@ -95,6 +121,27 @@ public class CassandraProducer extends DefaultProducer {
         return resultSet;
     }
 
+    /**
+     * Execute CQL as is
+     */
+    private ResultSet executeStatement(Session session, String messageCql, Object[] cqlParams)
{
+        ResultSet resultSet;
+        String cql;
+        if (messageCql == null || messageCql.isEmpty()) {
+            // URI CQL
+            cql = getEndpoint().getCql();
+        } else {
+            // Message CQL
+            cql = messageCql;
+        }
+        if (cqlParams == null || cqlParams.length == 0) {
+            resultSet = session.execute(cql);
+        } else {
+            resultSet = session.execute(cql, cqlParams);
+        }
+        return resultSet;
+    }
+
     public void process(Exchange exchange) throws Exception {
         ResultSet resultSet = execute(exchange.getIn());
         getEndpoint().fillMessage(resultSet, exchange.getOut());

http://git-wip-us.apache.org/repos/asf/camel/blob/742eda6b/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 76d303d..788844e 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
@@ -66,6 +66,21 @@ public class CassandraComponentConsumerTest extends CamelTestSupport {
     }
 
     @Test
+    public void testConsumeUnprepared() throws Exception {
+        MockEndpoint mock = getMockEndpoint("mock:resultUnprepared");
+        mock.expectedMinimumMessageCount(1);
+        mock.whenAnyExchangeReceived(new Processor() {
+            @Override
+            public void process(Exchange exchange) throws Exception {
+                Object body = exchange.getIn().getBody();
+                assertTrue(body instanceof List);
+            }
+        });
+        mock.await(1, TimeUnit.SECONDS);
+        assertMockEndpointsSatisfied();
+    }
+
+    @Test
     public void testConsumeOne() throws Exception {
         MockEndpoint mock = getMockEndpoint("mock:resultOne");
         mock.expectedMinimumMessageCount(1);
@@ -87,6 +102,8 @@ public class CassandraComponentConsumerTest extends CamelTestSupport {
             public void configure() {
                 from("cql://localhost/camel_ks?cql=" + CQL)
                         .to("mock:resultAll");
+                from("cql://localhost/camel_ks?cql=" + CQL + "&prepareStatements=false")
+                        .to("mock:resultUnprepared");
                 from("cql://localhost/camel_ks?cql=" + CQL + "&resultSetConversionStrategy=ONE")
                         .to("mock:resultOne");
             }

http://git-wip-us.apache.org/repos/asf/camel/blob/742eda6b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerUnpreparedTest.java
----------------------------------------------------------------------
diff --git a/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerUnpreparedTest.java
b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerUnpreparedTest.java
new file mode 100644
index 0000000..f8702d1
--- /dev/null
+++ b/components/camel-cassandraql/src/test/java/org/apache/camel/component/cassandra/CassandraComponentProducerUnpreparedTest.java
@@ -0,0 +1,119 @@
+/**
+ * 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.*;
+import org.apache.camel.Produce;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.cassandraunit.CassandraCQLUnit;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class CassandraComponentProducerUnpreparedTest extends CamelTestSupport {
+
+    private static final String CQL = "insert into camel_user(login, first_name, last_name)
values (?, ?, ?)";
+    private static final String NO_PARAMETER_CQL = "select login, first_name, last_name from
camel_user";
+
+    @Rule
+    public CassandraCQLUnit cassandra = CassandraUnitUtils.cassandraCQLUnit();
+
+    @Produce(uri = "direct:input")
+    ProducerTemplate producerTemplate;
+
+    @Produce(uri = "direct:inputNoParameter")
+    ProducerTemplate noParameterProducerTemplate;
+
+    @BeforeClass
+    public static void setUpClass() throws Exception {
+        CassandraUnitUtils.startEmbeddedCassandra();
+    }
+
+    @AfterClass
+    public static void tearDownClass() throws Exception {
+        CassandraUnitUtils.cleanEmbeddedCassandra();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            public void configure() {
+
+                from("direct:input")
+                        .to("cql://localhost/camel_ks?cql=" + CQL+"&prepareStatements=false");
+                from("direct:inputNoParameter")
+                        .to("cql://localhost/camel_ks?cql=" + NO_PARAMETER_CQL+"&prepareStatements=false");
+            }
+        };
+    }
+
+    @Test
+    public void testRequestUriCql() throws Exception {
+        Object response = producerTemplate.requestBody(Arrays.asList("w_jiang", "Willem",
"Jiang"));
+
+        Cluster cluster = CassandraUnitUtils.cassandraCluster();
+        Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
+        ResultSet resultSet = session.execute("select login, first_name, last_name from camel_user
where login = ?", "w_jiang");
+        Row row = resultSet.one();
+        assertNotNull(row);
+        assertEquals("Willem", row.getString("first_name"));
+        assertEquals("Jiang", row.getString("last_name"));
+        session.close();
+        cluster.close();
+    }
+
+    @Test
+    public void testRequestNoParameter_Null() throws Exception {
+        Object response = noParameterProducerTemplate.requestBody(null);
+
+        assertNotNull(response);
+        assertIsInstanceOf(List.class, response);
+        List<Row> rows = (List<Row>) response;
+    }
+
+    @Test
+    public void testRequestNoParameter_Empty() throws Exception {
+        Object response = noParameterProducerTemplate.requestBody(null);
+
+        assertNotNull(response);
+        assertIsInstanceOf(List.class, response);
+        List<Row> rows = (List<Row>) response;
+    }
+
+    @Test
+    public void testRequestMessageCql() throws Exception {
+        Object response = producerTemplate.requestBodyAndHeader(new Object[]{"Claus 2", "Ibsen
2", "c_ibsen"},
+                CassandraConstants.CQL_QUERY, "update camel_user set first_name=?, last_name=?
where login=?");
+
+        Cluster cluster = CassandraUnitUtils.cassandraCluster();
+        Session session = cluster.connect(CassandraUnitUtils.KEYSPACE);
+        ResultSet resultSet = session.execute("select login, first_name, last_name from camel_user
where login = ?", "c_ibsen");
+        Row row = resultSet.one();
+        assertNotNull(row);
+        assertEquals("Claus 2", row.getString("first_name"));
+        assertEquals("Ibsen 2", row.getString("last_name"));
+        session.close();
+        cluster.close();
+    }
+
+}


Mime
View raw message