camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/2] git commit: CAMEL-6418: Added separator option. Fixed and improved when using String as body with # placeholders to honor qutes and use new separator option.
Date Mon, 03 Jun 2013 14:01:27 GMT
Updated Branches:
  refs/heads/camel-2.11.x 51501fd94 -> 6f10a76bc
  refs/heads/master f6c6bba8f -> e14c66357


CAMEL-6418: Added separator option. Fixed and improved when using String as body with # placeholders
to honor qutes and use new separator option.


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

Branch: refs/heads/master
Commit: e14c663577cd49a6d81ec04d46cdfda377dd4e9e
Parents: f6c6bba
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Mon Jun 3 15:48:00 2013 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Mon Jun 3 15:48:00 2013 +0200

----------------------------------------------------------------------
 .../sql/DefaultSqlPrepareStatementStrategy.java    |   24 ++++-
 .../sql/DefaultSqlProcessingStrategy.java          |   13 ++-
 .../apache/camel/component/sql/SqlConsumer.java    |   13 ++-
 .../apache/camel/component/sql/SqlEndpoint.java    |   21 +++-
 .../apache/camel/component/sql/SqlProducer.java    |   15 ++-
 .../component/sql/SqlProducerSeparatorTest.java    |   79 +++++++++++++++
 6 files changed, 145 insertions(+), 20 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/e14c6635/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlPrepareStatementStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlPrepareStatementStrategy.java
b/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlPrepareStatementStrategy.java
index 0540c1d..5928f64 100644
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlPrepareStatementStrategy.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlPrepareStatementStrategy.java
@@ -18,13 +18,16 @@ package org.apache.camel.component.sql;
 
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.util.Arrays;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 import org.apache.camel.Exchange;
 import org.apache.camel.RuntimeExchangeException;
+import org.apache.camel.util.StringQuoteHelper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,6 +37,15 @@ import org.slf4j.LoggerFactory;
 public class DefaultSqlPrepareStatementStrategy implements SqlPrepareStatementStrategy {
 
     private static final Logger LOG = LoggerFactory.getLogger(DefaultSqlPrepareStatementStrategy.class);
+    private final char separator;
+
+    public DefaultSqlPrepareStatementStrategy() {
+        this(',');
+    }
+
+    public DefaultSqlPrepareStatementStrategy(char separator) {
+        this.separator = separator;
+    }
 
     @Override
     public String prepareQuery(String query, boolean allowNamedParameters) throws SQLException
{
@@ -111,8 +123,16 @@ public class DefaultSqlPrepareStatementStrategy implements SqlPrepareStatementSt
 
 
         } else {
-            // just use a regular iterator
-            return exchange.getContext().getTypeConverter().convertTo(Iterator.class, value);
+            // is the body a String
+            if (value instanceof String) {
+                // if the body is a String then honor quotes etc.
+                String[] tokens = StringQuoteHelper.splitSafeQuote((String)value, separator,
true);
+                List<String> list = Arrays.asList(tokens);
+                return list.iterator();
+            } else {
+                // just use a regular iterator
+                return exchange.getContext().getTypeConverter().convertTo(Iterator.class,
value);
+            }
         }
     }
 

http://git-wip-us.apache.org/repos/asf/camel/blob/e14c6635/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlProcessingStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlProcessingStrategy.java
b/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlProcessingStrategy.java
index 787c3d6..d641c0f 100644
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlProcessingStrategy.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlProcessingStrategy.java
@@ -32,19 +32,24 @@ import org.springframework.jdbc.core.PreparedStatementCallback;
 public class DefaultSqlProcessingStrategy implements SqlProcessingStrategy {
 
     private static final Logger LOG = LoggerFactory.getLogger(DefaultSqlProcessingStrategy.class);
+    private final SqlPrepareStatementStrategy sqlPrepareStatementStrategy;
+
+    public DefaultSqlProcessingStrategy(SqlPrepareStatementStrategy sqlPrepareStatementStrategy)
{
+        this.sqlPrepareStatementStrategy = sqlPrepareStatementStrategy;
+    }
 
     @Override
     public int commit(final SqlEndpoint endpoint, final Exchange exchange, final Object data,
final JdbcTemplate jdbcTemplate, final String query) throws Exception {
 
-        final String preparedQuery = endpoint.getPrepareStatementStrategy().prepareQuery(query,
endpoint.isAllowNamedParameters());
+        final String preparedQuery = sqlPrepareStatementStrategy.prepareQuery(query, endpoint.isAllowNamedParameters());
 
         return jdbcTemplate.execute(preparedQuery, new PreparedStatementCallback<Integer>()
{
             public Integer doInPreparedStatement(PreparedStatement ps) throws SQLException
{
                 int expected = ps.getParameterMetaData().getParameterCount();
 
-                Iterator<?> iterator = endpoint.getPrepareStatementStrategy().createPopulateIterator(query,
preparedQuery, expected, exchange, data);
+                Iterator<?> iterator = sqlPrepareStatementStrategy.createPopulateIterator(query,
preparedQuery, expected, exchange, data);
                 if (iterator != null) {
-                    endpoint.getPrepareStatementStrategy().populateStatement(ps, iterator,
expected);
+                    sqlPrepareStatementStrategy.populateStatement(ps, iterator, expected);
                     LOG.trace("Execute query {}", query);
                     ps.execute();
 
@@ -62,7 +67,7 @@ public class DefaultSqlProcessingStrategy implements SqlProcessingStrategy
{
 
     @Override
     public int commitBatchComplete(final SqlEndpoint endpoint, final JdbcTemplate jdbcTemplate,
final String query) throws Exception {
-        final String preparedQuery = endpoint.getPrepareStatementStrategy().prepareQuery(query,
endpoint.isAllowNamedParameters());
+        final String preparedQuery = sqlPrepareStatementStrategy.prepareQuery(query, endpoint.isAllowNamedParameters());
 
         return jdbcTemplate.execute(preparedQuery, new PreparedStatementCallback<Integer>()
{
             public Integer doInPreparedStatement(PreparedStatement ps) throws SQLException
{

http://git-wip-us.apache.org/repos/asf/camel/blob/e14c6635/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
index 7b0a1bc..2bb1f4a 100644
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
@@ -45,6 +45,8 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
 
     private final String query;
     private final JdbcTemplate jdbcTemplate;
+    private final SqlPrepareStatementStrategy sqlPrepareStatementStrategy;
+    private final SqlProcessingStrategy sqlProcessingStrategy;
 
     @UriParam
     private String onConsume;
@@ -69,10 +71,13 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
         }
     }
 
-    public SqlConsumer(SqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate,
String query) {
+    public SqlConsumer(SqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate,
String query,
+                       SqlPrepareStatementStrategy sqlPrepareStatementStrategy, SqlProcessingStrategy
sqlProcessingStrategy) {
         super(endpoint, processor);
         this.jdbcTemplate = jdbcTemplate;
         this.query = query;
+        this.sqlPrepareStatementStrategy = sqlPrepareStatementStrategy;
+        this.sqlProcessingStrategy = sqlProcessingStrategy;
     }
 
     @Override
@@ -86,7 +91,7 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
         shutdownRunningTask = null;
         pendingExchanges = 0;
 
-        final String preparedQuery = getEndpoint().getPrepareStatementStrategy().prepareQuery(query,
getEndpoint().isAllowNamedParameters());
+        final String preparedQuery = sqlPrepareStatementStrategy.prepareQuery(query, getEndpoint().isAllowNamedParameters());
 
         Integer messagePolled = jdbcTemplate.execute(preparedQuery, new PreparedStatementCallback<Integer>()
{
             @Override
@@ -179,7 +184,7 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
             try {
                 // we can only run on consume if there was data
                 if (data != null && sql != null) {
-                    int updateCount = getEndpoint().getProcessingStrategy().commit(getEndpoint(),
exchange, data, jdbcTemplate, sql);
+                    int updateCount = sqlProcessingStrategy.commit(getEndpoint(), exchange,
data, jdbcTemplate, sql);
                     if (expectedUpdateCount > -1 && updateCount != expectedUpdateCount)
{
                         String msg = "Expected update count " + expectedUpdateCount + " but
was " + updateCount + " executing query: " + sql;
                         throw new SQLException(msg);
@@ -196,7 +201,7 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
 
         try {
             if (onConsumeBatchComplete != null) {
-                int updateCount = getEndpoint().getProcessingStrategy().commitBatchComplete(getEndpoint(),
jdbcTemplate, onConsumeBatchComplete);
+                int updateCount = sqlProcessingStrategy.commitBatchComplete(getEndpoint(),
jdbcTemplate, onConsumeBatchComplete);
                 log.debug("onConsumeBatchComplete update count {}", updateCount);
             }
         } catch (Exception e) {

http://git-wip-us.apache.org/repos/asf/camel/blob/e14c6635/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlEndpoint.java
b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlEndpoint.java
index 1b9ec33..300079d 100644
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlEndpoint.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlEndpoint.java
@@ -42,9 +42,9 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
     @UriParam
     private int maxMessagesPerPoll;
     @UriParam
-    private SqlProcessingStrategy processingStrategy = new DefaultSqlProcessingStrategy();
+    private SqlProcessingStrategy processingStrategy;
     @UriParam
-    private SqlPrepareStatementStrategy prepareStatementStrategy = new DefaultSqlPrepareStatementStrategy();
+    private SqlPrepareStatementStrategy prepareStatementStrategy;
     @UriParam
     private String onConsume;
     @UriParam
@@ -55,6 +55,8 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
     private boolean allowNamedParameters = true;
     @UriParam
     private boolean alwaysPopulateStatement;
+    @UriParam
+    private char separator = ',';
 
     public SqlEndpoint() {
     }
@@ -66,7 +68,9 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
     }
 
     public Consumer createConsumer(Processor processor) throws Exception {
-        SqlConsumer consumer = new SqlConsumer(this, processor, jdbcTemplate, query);
+        SqlPrepareStatementStrategy prepareStrategy = prepareStatementStrategy != null ?
prepareStatementStrategy : new DefaultSqlPrepareStatementStrategy(separator);
+        SqlProcessingStrategy proStrategy = processingStrategy != null ? processingStrategy
: new DefaultSqlProcessingStrategy(prepareStrategy);
+        SqlConsumer consumer = new SqlConsumer(this, processor, jdbcTemplate, query, prepareStrategy,
proStrategy);
         consumer.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
         consumer.setOnConsume(getOnConsume());
         consumer.setOnConsumeFailed(getOnConsumeFailed());
@@ -76,7 +80,8 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
     }
 
     public Producer createProducer() throws Exception {
-        return new SqlProducer(this, query, jdbcTemplate, batch, alwaysPopulateStatement);
+        SqlPrepareStatementStrategy prepareStrategy = prepareStatementStrategy != null ?
prepareStatementStrategy : new DefaultSqlPrepareStatementStrategy(separator);
+        return new SqlProducer(this, query, jdbcTemplate, prepareStrategy, batch, alwaysPopulateStatement);
     }
 
     public boolean isSingleton() {
@@ -180,6 +185,14 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
         this.alwaysPopulateStatement = alwaysPopulateStatement;
     }
 
+    public char getSeparator() {
+        return separator;
+    }
+
+    public void setSeparator(char separator) {
+        this.separator = separator;
+    }
+
     @Override
     protected String createEndpointUri() {
         // Make sure it's properly encoded

http://git-wip-us.apache.org/repos/asf/camel/blob/e14c6635/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProducer.java
b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProducer.java
index 1fa6864..97ff150 100644
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProducer.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProducer.java
@@ -34,10 +34,13 @@ public class SqlProducer extends DefaultProducer {
     private JdbcTemplate jdbcTemplate;
     private boolean batch;
     private boolean alwaysPopulateStatement;
+    private SqlPrepareStatementStrategy sqlPrepareStatementStrategy;
 
-    public SqlProducer(SqlEndpoint endpoint, String query, JdbcTemplate jdbcTemplate, boolean
batch, boolean alwaysPopulateStatement) {
+    public SqlProducer(SqlEndpoint endpoint, String query, JdbcTemplate jdbcTemplate, SqlPrepareStatementStrategy
sqlPrepareStatementStrategy,
+                       boolean batch, boolean alwaysPopulateStatement) {
         super(endpoint);
         this.jdbcTemplate = jdbcTemplate;
+        this.sqlPrepareStatementStrategy = sqlPrepareStatementStrategy;
         this.query = query;
         this.batch = batch;
         this.alwaysPopulateStatement = alwaysPopulateStatement;
@@ -52,7 +55,7 @@ public class SqlProducer extends DefaultProducer {
         String queryHeader = exchange.getIn().getHeader(SqlConstants.SQL_QUERY, String.class);
 
         final String sql = queryHeader != null ? queryHeader : query;
-        final String preparedQuery = getEndpoint().getPrepareStatementStrategy().prepareQuery(sql,
getEndpoint().isAllowNamedParameters());
+        final String preparedQuery = sqlPrepareStatementStrategy.prepareQuery(sql, getEndpoint().isAllowNamedParameters());
 
         jdbcTemplate.execute(preparedQuery, new PreparedStatementCallback<Map<?, ?>>()
{
             public Map<?, ?> doInPreparedStatement(PreparedStatement ps) throws SQLException
{
@@ -65,13 +68,13 @@ public class SqlProducer extends DefaultProducer {
                         Iterator<?> iterator = exchange.getIn().getBody(Iterator.class);
                         while (iterator != null && iterator.hasNext()) {
                             Object value = iterator.next();
-                            Iterator<?> i = getEndpoint().getPrepareStatementStrategy().createPopulateIterator(sql,
preparedQuery, expected, exchange, value);
-                            getEndpoint().getPrepareStatementStrategy().populateStatement(ps,
i, expected);
+                            Iterator<?> i = sqlPrepareStatementStrategy.createPopulateIterator(sql,
preparedQuery, expected, exchange, value);
+                            sqlPrepareStatementStrategy.populateStatement(ps, i, expected);
                             ps.addBatch();
                         }
                     } else {
-                        Iterator<?> i = getEndpoint().getPrepareStatementStrategy().createPopulateIterator(sql,
preparedQuery, expected, exchange, exchange.getIn().getBody());
-                        getEndpoint().getPrepareStatementStrategy().populateStatement(ps,
i, expected);
+                        Iterator<?> i = sqlPrepareStatementStrategy.createPopulateIterator(sql,
preparedQuery, expected, exchange, exchange.getIn().getBody());
+                        sqlPrepareStatementStrategy.populateStatement(ps, i, expected);
                     }
                 }
 

http://git-wip-us.apache.org/repos/asf/camel/blob/e14c6635/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlProducerSeparatorTest.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlProducerSeparatorTest.java
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlProducerSeparatorTest.java
new file mode 100755
index 0000000..3182cd1
--- /dev/null
+++ b/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlProducerSeparatorTest.java
@@ -0,0 +1,79 @@
+/**
+ * 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.sql;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+
+/**
+ * @version 
+ */
+public class SqlProducerSeparatorTest extends CamelTestSupport {
+
+    private EmbeddedDatabase db;
+    private JdbcTemplate jdbcTemplate;
+
+    @Before
+    public void setUp() throws Exception {
+        db = new EmbeddedDatabaseBuilder()
+            .setType(EmbeddedDatabaseType.DERBY).addScript("sql/createAndPopulateDatabase.sql").build();
+
+        jdbcTemplate = new JdbcTemplate(db);
+
+        super.setUp();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        super.tearDown();
+        
+        db.shutdown();
+    }
+
+    @Test
+    public void testSeparator() throws Exception {
+        MockEndpoint mock = getMockEndpoint("mock:result");
+        mock.expectedMessageCount(1);
+
+        template.sendBody("direct:start", "4;'Food, Inc';'LGPL'");
+
+        mock.assertIsSatisfied();
+
+        assertEquals(4, jdbcTemplate.queryForInt("select count(*) from projects"));
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            public void configure() {
+                getContext().getComponent("sql", SqlComponent.class).setDataSource(db);
+
+                from("direct:start")
+                    .to("sql:insert into projects (id, project, license) values (#, #, #)?separator=;")
+                    .to("mock:result");
+            }
+        };
+    }
+}


Mime
View raw message