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 2015/10/05 11:50:26 UTC
[01/10] camel git commit: CAMEL-9162: camel-elsql component
Repository: camel
Updated Branches:
refs/heads/master f5614efb8 -> 5fc96ef45
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/ed0f42a1
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/ed0f42a1
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/ed0f42a1
Branch: refs/heads/master
Commit: ed0f42a1434cda23b85a140a40a54bdf3ae0c8b5
Parents: 6cf9956
Author: Claus Ibsen <da...@apache.org>
Authored: Sun Oct 4 14:33:06 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 10:54:53 2015 +0200
----------------------------------------------------------------------
components/camel-elsql/pom.xml | 1 +
.../camel/component/elsql/ElsqlComponent.java | 55 +++++++------
.../camel/component/elsql/ElsqlEndpoint.java | 21 ++---
.../camel/component/elsql/ElsqlProducer.java | 19 +++--
.../component/elsql/ElsqlSqlMapSource.java | 54 +++++++++++++
.../camel/component/elsql/ElsqlSqlParams.java | 68 ----------------
.../elsql/ElsqlSqlProcessingStrategy.java | 6 +-
.../component/elsql/ElSqlDataSourceTest.java | 85 ++++++++++++++++++++
.../src/test/resources/elsql/projects.elsql | 5 ++
.../src/test/resources/log4j.properties | 39 +++++++++
.../resources/sql/createAndPopulateDatabase.sql | 23 ++++++
.../camel/component/sql/SqlComponent.java | 4 +
.../apache/camel/component/sql/SqlEndpoint.java | 73 ++++++++++-------
13 files changed, 314 insertions(+), 139 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-elsql/pom.xml b/components/camel-elsql/pom.xml
index d82bb05..f45f7e6 100644
--- a/components/camel-elsql/pom.xml
+++ b/components/camel-elsql/pom.xml
@@ -34,6 +34,7 @@
org.apache.camel.component.elsql.*;${camel.osgi.version}
</camel.osgi.export.pkg>
<camel.osgi.import.pkg>
+ org.apache.camel.component.sql.*,
!org.apache.camel.component.elsql.*,
${camel.osgi.import.defaults},
*
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
index ddea39d..40e6530 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
@@ -20,20 +20,20 @@ import java.util.Map;
import javax.sql.DataSource;
import com.opengamma.elsql.ElSqlConfig;
-import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
-import org.apache.camel.component.sql.SqlComponent;
+import org.apache.camel.impl.UriEndpointComponent;
import org.apache.camel.util.CamelContextHelper;
import org.apache.camel.util.IntrospectionSupport;
-import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
-public class ElsqlComponent extends SqlComponent {
+public class ElsqlComponent extends UriEndpointComponent {
+ private DataSource dataSource;
private ElSqlConfig elSqlConfig;
private String resourceUri;
- public ElsqlComponent(CamelContext context) {
- super(context, ElsqlEndpoint.class);
+ public ElsqlComponent() {
+ super(ElsqlEndpoint.class);
}
@Override
@@ -57,12 +57,12 @@ public class ElsqlComponent extends SqlComponent {
throw new IllegalArgumentException("DataSource must be configured");
}
- JdbcTemplate jdbcTemplate = new JdbcTemplate(target);
+ NamedParameterJdbcTemplate jdbcTemplate = new NamedParameterJdbcTemplate(target);
IntrospectionSupport.setProperties(jdbcTemplate, parameters, "template.");
String elsqlName = remaining;
String resUri = resourceUri;
- String[] part = remaining.split("/");
+ String[] part = remaining.split(":");
if (part.length == 2) {
elsqlName = part[0];
resUri = part[1];
@@ -104,6 +104,27 @@ public class ElsqlComponent extends SqlComponent {
return endpoint;
}
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ if (elSqlConfig == null) {
+ elSqlConfig = ElSqlConfig.DEFAULT;
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ }
+
+ public DataSource getDataSource() {
+ return dataSource;
+ }
+
+ /**
+ * Sets the DataSource to use to communicate with the database.
+ */
public ElSqlConfig getElSqlConfig() {
return elSqlConfig;
}
@@ -126,22 +147,4 @@ public class ElsqlComponent extends SqlComponent {
this.resourceUri = resourceUri;
}
- /**
- * Sets the DataSource to use to communicate with the database.
- */
- @Override
- public void setDataSource(DataSource dataSource) {
- super.setDataSource(dataSource);
- }
-
- /**
- * Sets whether to use placeholder and replace all placeholder characters with ? sign in the SQL queries.
- * <p/>
- * This option is default <tt>true</tt>
- */
- @Override
- public void setUsePlaceholder(boolean usePlaceholder) {
- super.setUsePlaceholder(usePlaceholder);
- }
-
}
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
index 8ef9845..9025d30 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
@@ -25,20 +25,19 @@ import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.component.sql.SqlEndpoint;
-import org.apache.camel.component.sql.SqlPrepareStatementStrategy;
-import org.apache.camel.component.sql.SqlProcessingStrategy;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ResourceHelper;
-import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
-@UriEndpoint(scheme = "elsql", title = "SQL", syntax = "elsql:elsqlName/resourceUri", consumerClass = ElsqlConsumer.class, label = "database,sql")
+@UriEndpoint(scheme = "elsql", title = "SQL", syntax = "elsql:elsqlName:resourceUri", consumerClass = ElsqlConsumer.class, label = "database,sql")
public class ElsqlEndpoint extends SqlEndpoint {
private volatile ElSql elSql;
+ private NamedParameterJdbcTemplate jdbcTemplate;
@UriPath
@Metadata(required = "true")
@@ -48,18 +47,21 @@ public class ElsqlEndpoint extends SqlEndpoint {
@UriParam
private ElSqlConfig elSqlConfig;
- public ElsqlEndpoint(String uri, Component component, JdbcTemplate jdbcTemplate, String elsqlName, String resourceUri) {
- super(uri, component, jdbcTemplate, null);
+ public ElsqlEndpoint(String uri, Component component, NamedParameterJdbcTemplate jdbcTemplate, String elsqlName, String resourceUri) {
+ super(uri, component, null, null);
this.elsqlName = elsqlName;
this.resourceUri = resourceUri;
+ this.jdbcTemplate = jdbcTemplate;
}
@Override
public Consumer createConsumer(Processor processor) throws Exception {
+ // TODO: must be named
+ /*
SqlProcessingStrategy proStrategy = new ElsqlSqlProcessingStrategy(elsqlName, elSql);
SqlPrepareStatementStrategy preStategy = new ElsqlSqlPrepareStatementStrategy();
- ElsqlConsumer consumer = new ElsqlConsumer(this, processor, getJdbcTemplate(), elsqlName, preStategy, proStrategy);
+ ElsqlConsumer consumer = new ElsqlConsumer(this, processor, jdbcTemplate, elsqlName, preStategy, proStrategy);
consumer.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
consumer.setOnConsume(getOnConsume());
consumer.setOnConsumeFailed(getOnConsumeFailed());
@@ -70,11 +72,13 @@ public class ElsqlEndpoint extends SqlEndpoint {
consumer.setRouteEmptyResultSet(isRouteEmptyResultSet());
configureConsumer(consumer);
return consumer;
+ */
+ return null;
}
@Override
public Producer createProducer() throws Exception {
- ElsqlProducer result = new ElsqlProducer(this, elSql, elsqlName, getJdbcTemplate());
+ ElsqlProducer result = new ElsqlProducer(this, elSql, elsqlName, jdbcTemplate);
return result;
}
@@ -86,7 +90,6 @@ public class ElsqlEndpoint extends SqlEndpoint {
ObjectHelper.notNull(resourceUri, "resourceUri", this);
URL url = ResourceHelper.resolveMandatoryResourceAsUrl(getCamelContext().getClassResolver(), resourceUri);
-
elSql = ElSql.parse(elSqlConfig, url);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
index 73925d3..fa3f5df 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
@@ -22,14 +22,16 @@ import java.sql.SQLException;
import java.util.List;
import com.opengamma.elsql.ElSql;
+import com.opengamma.elsql.SpringSqlParams;
import org.apache.camel.Exchange;
import org.apache.camel.component.sql.SqlConstants;
import org.apache.camel.component.sql.SqlEndpoint;
import org.apache.camel.component.sql.SqlOutputType;
import org.apache.camel.impl.DefaultProducer;
import org.springframework.dao.DataAccessException;
-import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementCallback;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import static org.springframework.jdbc.support.JdbcUtils.closeResultSet;
@@ -37,9 +39,9 @@ public class ElsqlProducer extends DefaultProducer {
private final ElSql elSql;
private final String elSqlName;
- private final JdbcTemplate jdbcTemplate;
+ private final NamedParameterJdbcTemplate jdbcTemplate;
- public ElsqlProducer(SqlEndpoint endpoint, ElSql elSql, String elSqlName, JdbcTemplate jdbcTemplate) {
+ public ElsqlProducer(SqlEndpoint endpoint, ElSql elSql, String elSqlName, NamedParameterJdbcTemplate jdbcTemplate) {
super(endpoint);
this.elSql = elSql;
this.elSqlName = elSqlName;
@@ -55,19 +57,22 @@ public class ElsqlProducer extends DefaultProducer {
public void process(final Exchange exchange) throws Exception {
Object data = exchange.getIn().getBody();
- final String sql = elSql.getSql(elSqlName, new ElsqlSqlParams(exchange, data));
+ final SqlParameterSource param = new ElsqlSqlMapSource(exchange, data);
+ final String sql = elSql.getSql(elSqlName, new SpringSqlParams(param));
+ log.debug("ElSql @{} using sql: {}", elSqlName, sql);
- jdbcTemplate.execute(sql, new PreparedStatementCallback<Object>() {
+ jdbcTemplate.execute(sql, param, new PreparedStatementCallback<Object>() {
@Override
public Object doInPreparedStatement(PreparedStatement ps) throws SQLException, DataAccessException {
ResultSet rs = null;
try {
boolean isResultSet = ps.execute();
if (isResultSet) {
+ rs = ps.getResultSet();
+
// preserve headers first, so we can override the SQL_ROW_COUNT header
exchange.getOut().getHeaders().putAll(exchange.getIn().getHeaders());
- rs = ps.getResultSet();
SqlOutputType outputType = getEndpoint().getOutputType();
log.trace("Got result list from query: {}, outputType={}", rs, outputType);
if (outputType == SqlOutputType.SelectList) {
@@ -99,8 +104,6 @@ public class ElsqlProducer extends DefaultProducer {
} else {
throw new IllegalArgumentException("Invalid outputType=" + outputType);
}
- } else {
- exchange.getIn().setHeader(SqlConstants.SQL_UPDATE_COUNT, ps.getUpdateCount());
}
} finally {
closeResultSet(rs);
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
new file mode 100644
index 0000000..e8035b0
--- /dev/null
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
@@ -0,0 +1,54 @@
+/**
+ * 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.elsql;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.camel.Exchange;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+
+public class ElsqlSqlMapSource extends MapSqlParameterSource {
+
+ private final Exchange exchange;
+ private final Map<?, ?> bodyMap;
+ private final Map<?, ?> headersMap;
+
+ public ElsqlSqlMapSource(Exchange exchange, Object body) {
+ this.exchange = exchange;
+ this.bodyMap = safeMap(exchange.getContext().getTypeConverter().tryConvertTo(Map.class, body));
+ this.headersMap = safeMap(exchange.getIn().getHeaders());
+
+ addValue("body", body);
+
+ for (Map.Entry<?, ?> entry : bodyMap.entrySet()) {
+ String name = entry.getKey().toString();
+ Object value = entry.getValue();
+ addValue(name, value);
+ }
+ for (Map.Entry<?, ?> entry : headersMap.entrySet()) {
+ String name = entry.getKey().toString();
+ Object value = entry.getValue();
+ addValue(name, value);
+ }
+ }
+
+ private static Map<?, ?> safeMap(Map<?, ?> map) {
+ return (map == null || map.isEmpty()) ? Collections.emptyMap() : map;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlParams.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlParams.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlParams.java
deleted file mode 100644
index 6822e8b..0000000
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlParams.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.camel.component.elsql;
-
-import java.util.Collections;
-import java.util.Map;
-
-import com.opengamma.elsql.SqlParams;
-import org.apache.camel.Exchange;
-import org.apache.camel.language.simple.SimpleLanguage;
-
-public class ElsqlSqlParams implements SqlParams {
-
- private final Exchange exchange;
- private final Map<?, ?> bodyMap;
- private final Map<?, ?> headersMap;
-
- public ElsqlSqlParams(Exchange exchange, Object body) {
- this.exchange = exchange;
- this.bodyMap = safeMap(exchange.getContext().getTypeConverter().tryConvertTo(Map.class, body));
- this.headersMap = safeMap(exchange.getIn().getHeaders());
- }
-
- private static Map<?, ?> safeMap(Map<?, ?> map) {
- return (map == null || map.isEmpty()) ? Collections.emptyMap() : map;
- }
-
- @Override
- public boolean contains(String variable) {
- if (variable.startsWith("${") && variable.endsWith("}")) {
- return SimpleLanguage.expression(variable).evaluate(exchange, Object.class) != null;
- } else if (bodyMap.containsKey(variable)) {
- return true;
- } else if (headersMap.containsKey(variable)) {
- return true;
- }
-
- return false;
- }
-
- @Override
- public Object get(String variable) {
- if (variable.startsWith("${") && variable.endsWith("}")) {
- return SimpleLanguage.expression(variable).evaluate(exchange, Object.class) != null;
- } else if (bodyMap.containsKey(variable)) {
- return bodyMap.get(variable);
- } else if (headersMap.containsKey(variable)) {
- return headersMap.get(variable);
- }
-
- return null;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
index d22c33e..1134088 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
@@ -20,6 +20,7 @@ import java.sql.PreparedStatement;
import java.sql.SQLException;
import com.opengamma.elsql.ElSql;
+import com.opengamma.elsql.SpringSqlParams;
import org.apache.camel.Exchange;
import org.apache.camel.component.sql.SqlEndpoint;
import org.apache.camel.component.sql.SqlProcessingStrategy;
@@ -28,6 +29,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementCallback;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
public class ElsqlSqlProcessingStrategy implements SqlProcessingStrategy {
@@ -42,7 +44,9 @@ public class ElsqlSqlProcessingStrategy implements SqlProcessingStrategy {
@Override
public int commit(final SqlEndpoint endpoint, final Exchange exchange, final Object data, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
- String sql = elSql.getSql(elSqlName, new ElsqlSqlParams(exchange, data));
+ final SqlParameterSource param = new ElsqlSqlMapSource(exchange, data);
+ final String sql = elSql.getSql(elSqlName, new SpringSqlParams(param));
+ LOG.debug("ElSql @{} using sql: {}", elSqlName, sql);
return jdbcTemplate.execute(sql, new PreparedStatementCallback<Integer>() {
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlDataSourceTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlDataSourceTest.java b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlDataSourceTest.java
new file mode 100644
index 0000000..858daf4
--- /dev/null
+++ b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlDataSourceTest.java
@@ -0,0 +1,85 @@
+/**
+ * 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.elsql;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.After;
+import org.junit.Test;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+
+public class ElSqlDataSourceTest extends CamelTestSupport {
+
+ private EmbeddedDatabase db;
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry jndi = super.createRegistry();
+
+ // this is the database we create with some initial data for our unit test
+ db = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.DERBY).addScript("sql/createAndPopulateDatabase.sql").build();
+
+ jndi.bind("dataSource", db);
+
+ return jndi;
+ }
+
+ @Test
+ public void testSimpleBody() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ template.sendBody("direct:simple", "XXX");
+
+ mock.assertIsSatisfied();
+
+ // the result is a List
+ List<?> received = assertIsInstanceOf(List.class, mock.getReceivedExchanges().get(0).getIn().getBody());
+
+ // and each row in the list is a Map
+ Map<?, ?> row = assertIsInstanceOf(Map.class, received.get(0));
+
+ // and we should be able the get the project from the map that should be Linux
+ assertEquals("Linux", row.get("PROJECT"));
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ super.tearDown();
+
+ db.shutdown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:simple")
+ .to("elsql:projects:elsql/projects.elsql?dataSource=dataSource")
+ .to("mock:result");
+ }
+ };
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/test/resources/elsql/projects.elsql
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/resources/elsql/projects.elsql b/components/camel-elsql/src/test/resources/elsql/projects.elsql
new file mode 100644
index 0000000..75dfe51
--- /dev/null
+++ b/components/camel-elsql/src/test/resources/elsql/projects.elsql
@@ -0,0 +1,5 @@
+@NAME(projects)
+ SELECT *
+ FROM projects
+ WHERE license = :body
+ ORDER BY id
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/resources/log4j.properties b/components/camel-elsql/src/test/resources/log4j.properties
new file mode 100755
index 0000000..d5af410
--- /dev/null
+++ b/components/camel-elsql/src/test/resources/log4j.properties
@@ -0,0 +1,39 @@
+## ------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ------------------------------------------------------------------------
+
+#
+# The logging properties used for testing
+#
+log4j.rootLogger=INFO, out
+
+#log4j.logger.org.apache.camel.component.sql=DEBUG
+#log4j.logger.org.apache.camel.component.sql=TRACE
+log4j.logger.org.apache.camel.component.elsql=DEBUG
+#log4j.logger.org.apache.camel.component.elsql=TRACE
+
+# CONSOLE appender not used by default
+log4j.appender.out=org.apache.log4j.ConsoleAppender
+log4j.appender.out.layout=org.apache.log4j.PatternLayout
+log4j.appender.out.layout.ConversionPattern=[%30.30t] %-30.30c{1} %-5p %m%n
+#log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+# File appender
+log4j.appender.file=org.apache.log4j.FileAppender
+log4j.appender.file.layout=org.apache.log4j.PatternLayout
+log4j.appender.file.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.file.file=target/camel-elsql-test.log
+log4j.appender.file.append=true
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-elsql/src/test/resources/sql/createAndPopulateDatabase.sql
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/resources/sql/createAndPopulateDatabase.sql b/components/camel-elsql/src/test/resources/sql/createAndPopulateDatabase.sql
new file mode 100644
index 0000000..65206c2
--- /dev/null
+++ b/components/camel-elsql/src/test/resources/sql/createAndPopulateDatabase.sql
@@ -0,0 +1,23 @@
+-- ------------------------------------------------------------------------
+-- 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.
+-- ------------------------------------------------------------------------
+
+-- START SNIPPET: e1
+create table projects (id integer primary key, project varchar(10), license varchar(5));
+insert into projects values (1, 'Camel', 'ASF');
+insert into projects values (2, 'AMQ', 'ASF');
+insert into projects values (3, 'Linux', 'XXX');
+-- END SNIPPET: e1
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java
index 1d9ada5..7236392 100755
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java
@@ -41,6 +41,10 @@ public class SqlComponent extends UriEndpointComponent {
super(SqlEndpoint.class);
}
+ public SqlComponent(Class<? extends Endpoint> endpointClass) {
+ super(endpointClass);
+ }
+
public SqlComponent(CamelContext context) {
super(context, SqlEndpoint.class);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/ed0f42a1/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 af0d4ad..761c125 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
@@ -49,52 +49,73 @@ import org.springframework.jdbc.core.RowMapperResultSetExtractor;
public class SqlEndpoint extends DefaultPollingEndpoint {
private JdbcTemplate jdbcTemplate;
- @UriPath @Metadata(required = "true")
+ @UriPath(description = "Sets the SQL query to perform") @Metadata(required = "true")
+
private String query;
- @UriParam
+ @UriParam(description = "Sets the reference to a DataSource to lookup from the registry, to use for communicating with the database.")
@Deprecated
private String dataSourceRef;
- @UriParam
+ @UriParam(description = "Sets the DataSource to use to communicate with the database.")
private DataSource dataSource;
- @UriParam(label = "producer")
+ @UriParam(label = "producer", description = "Enables or disables batch mode")
private boolean batch;
- @UriParam(label = "consumer")
+ @UriParam(label = "consumer", description = "Sets the maximum number of messages to poll")
private int maxMessagesPerPoll;
- @UriParam(label = "consumer,advanced")
+ @UriParam(label = "consumer,advanced",
+ description = "Allows to plugin to use a custom org.apache.camel.component.sql.SqlProcessingStrategy to execute queries when the consumer has processed the rows/batch.")
private SqlProcessingStrategy processingStrategy;
- @UriParam(label = "advanced")
+ @UriParam(label = "advanced",
+ description = "Allows to plugin to use a custom org.apache.camel.component.sql.SqlPrepareStatementStrategy to control preparation of the query and prepared statement.")
private SqlPrepareStatementStrategy prepareStatementStrategy;
- @UriParam(label = "consumer")
+ @UriParam(label = "consumer",
+ description = "After processing each row then this query can be executed, if the Exchange was processed successfully, for example to mark the row as processed. The query can have parameter.")
private String onConsume;
- @UriParam(label = "consumer")
+ @UriParam(label = "consumer",
+ description = "After processing each row then this query can be executed, if the Exchange failed, for example to mark the row as failed. The query can have parameter.")
private String onConsumeFailed;
- @UriParam(label = "consumer")
+ @UriParam(label = "consumer",
+ description = "After processing the entire batch, this query can be executed to bulk update rows etc. The query cannot have parameters.")
private String onConsumeBatchComplete;
- @UriParam(label = "consumer", defaultValue = "true")
+ @UriParam(label = "consumer", defaultValue = "true",
+ description = "Sets how resultset should be delivered to route. Indicates delivery as either a list or individual object. defaults to true.")
private boolean useIterator = true;
- @UriParam(label = "consumer")
+ @UriParam(label = "consumer",
+ description = "Sets whether empty resultset should be allowed to be sent to the next hop. Defaults to false. So the empty resultset will be filtered out.")
private boolean routeEmptyResultSet;
- @UriParam(label = "consumer", defaultValue = "-1")
+ @UriParam(label = "consumer", defaultValue = "-1", description = "Sets an expected update count to validate when using onConsume.")
private int expectedUpdateCount = -1;
- @UriParam(label = "consumer")
+ @UriParam(label = "consumer", description = "Sets whether to break batch if onConsume failed.")
private boolean breakBatchOnConsumeFail;
- @UriParam(defaultValue = "true")
+ @UriParam(defaultValue = "true", description = "Whether to allow using named parameters in the queries.")
private boolean allowNamedParameters = true;
- @UriParam(label = "producer,advanced")
+ @UriParam(label = "producer,advanced",
+ description = "If enabled then the populateStatement method from org.apache.camel.component.sql.SqlPrepareStatementStrategy is always invoked, "
+ + "also if there is no expected parameters to be prepared. When this is false then the populateStatement is only invoked if there is 1"
+ + " or more expected parameters to be set; for example this avoids reading the message body/headers for SQL queries with no parameters.")
private boolean alwaysPopulateStatement;
- @UriParam(defaultValue = ",")
+ @UriParam(defaultValue = ",",
+ description = "The separator to use when parameter values is taken from message body (if the body is a String type), to be inserted at # placeholders."
+ + "Notice if you use named parameters, then a Map type is used instead. The default value is ,")
private char separator = ',';
- @UriParam(defaultValue = "SelectList")
+ @UriParam(defaultValue = "SelectList", description = "Make the output of consumer or producer to SelectList as List of Map, or SelectOne as single Java object in the following way:"
+ + "a) If the query has only single column, then that JDBC Column object is returned. (such as SELECT COUNT( * ) FROM PROJECT will return a Long object."
+ + "b) If the query has more than one column, then it will return a Map of that result."
+ + "c) If the outputClass is set, then it will convert the query result into an Java bean object by calling all the setters that match the column names."
+ + "It will assume your class has a default constructor to create instance with."
+ + "d) If the query resulted in more than one rows, it throws an non-unique result exception.")
private SqlOutputType outputType = SqlOutputType.SelectList;
- @UriParam
+ @UriParam(description = "Specify the full package and class name to use as conversion when outputType=SelectOne.")
private String outputClass;
- @UriParam(label = "producer,advanced")
+ @UriParam(label = "producer,advanced", description = "If set greater than zero, then Camel will use this count value of parameters to replace instead of"
+ + " querying via JDBC metadata API. This is useful if the JDBC vendor could not return correct parameters count, then user may override instead.")
private int parametersCount;
- @UriParam(label = "producer")
+ @UriParam(label = "producer", description = "If set, will ignore the results of the SQL query and use the existing IN message as the OUT message for the continuation of processing")
private boolean noop;
- @UriParam
+ @UriParam(description = "Store the query result in a header instead of the message body. By default, outputHeader == null and the query result is stored"
+ + " in the message body, any existing content in the message body is discarded. If outputHeader is set, the value is used as the name of the header"
+ + " to store the query result and the original message body is preserved.")
private String outputHeader;
- @UriParam(label = "producer")
+ @UriParam(label = "producer", description = "Whether to use the message body as the SQL and then headers for parameters. If this option is enabled then the SQL in the uri is not used.")
private boolean useMessageBodyForSql;
public SqlEndpoint() {
@@ -372,9 +393,7 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
}
/**
- * Sets how resultset should be delivered to route.
- * Indicates delivery as either a list or individual object.
- * defaults to true.
+ * Sets how resultset should be delivered to route. Indicates delivery as either a list or individual object. defaults to true.
*/
public void setUseIterator(boolean useIterator) {
this.useIterator = useIterator;
@@ -386,7 +405,7 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
/**
* Sets whether empty resultset should be allowed to be sent to the next hop.
- * defaults to false. So the empty resultset will be filtered out.
+ * Defaults to false. So the empty resultset will be filtered out.
*/
public void setRouteEmptyResultSet(boolean routeEmptyResultSet) {
this.routeEmptyResultSet = routeEmptyResultSet;
[03/10] camel git commit: CAMEL-9162: camel-elsql component
Posted by da...@apache.org.
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/3dd80567
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/3dd80567
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/3dd80567
Branch: refs/heads/master
Commit: 3dd805678d6ef0c15b3c6739e0f2af24d1356e35
Parents: 0b71c42
Author: Claus Ibsen <da...@apache.org>
Authored: Mon Oct 5 08:25:42 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 10:54:54 2015 +0200
----------------------------------------------------------------------
.../camel/component/elsql/ElsqlSqlProcessingStrategy.java | 6 +++---
.../camel/component/sql/DefaultSqlProcessingStrategy.java | 4 ++--
.../main/java/org/apache/camel/component/sql/SqlConsumer.java | 4 ++--
.../org/apache/camel/component/sql/SqlProcessingStrategy.java | 4 ++--
4 files changed, 9 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/3dd80567/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
index 1134088..ea933d8 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
@@ -22,7 +22,7 @@ import java.sql.SQLException;
import com.opengamma.elsql.ElSql;
import com.opengamma.elsql.SpringSqlParams;
import org.apache.camel.Exchange;
-import org.apache.camel.component.sql.SqlEndpoint;
+import org.apache.camel.component.sql.DefaultSqlEndpoint;
import org.apache.camel.component.sql.SqlProcessingStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,7 +43,7 @@ public class ElsqlSqlProcessingStrategy implements SqlProcessingStrategy {
}
@Override
- public int commit(final SqlEndpoint endpoint, final Exchange exchange, final Object data, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
+ public int commit(final DefaultSqlEndpoint endpoint, final Exchange exchange, final Object data, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
final SqlParameterSource param = new ElsqlSqlMapSource(exchange, data);
final String sql = elSql.getSql(elSqlName, new SpringSqlParams(param));
LOG.debug("ElSql @{} using sql: {}", elSqlName, sql);
@@ -63,7 +63,7 @@ public class ElsqlSqlProcessingStrategy implements SqlProcessingStrategy {
}
@Override
- public int commitBatchComplete(final SqlEndpoint endpoint, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
+ public int commitBatchComplete(final DefaultSqlEndpoint endpoint, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
return 0;
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/3dd80567/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 d641c0f..1d7f96d 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
@@ -39,7 +39,7 @@ public class DefaultSqlProcessingStrategy implements SqlProcessingStrategy {
}
@Override
- public int commit(final SqlEndpoint endpoint, final Exchange exchange, final Object data, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
+ public int commit(final DefaultSqlEndpoint endpoint, final Exchange exchange, final Object data, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
final String preparedQuery = sqlPrepareStatementStrategy.prepareQuery(query, endpoint.isAllowNamedParameters());
@@ -66,7 +66,7 @@ public class DefaultSqlProcessingStrategy implements SqlProcessingStrategy {
}
@Override
- public int commitBatchComplete(final SqlEndpoint endpoint, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
+ public int commitBatchComplete(final DefaultSqlEndpoint endpoint, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
final String preparedQuery = sqlPrepareStatementStrategy.prepareQuery(query, endpoint.isAllowNamedParameters());
return jdbcTemplate.execute(preparedQuery, new PreparedStatementCallback<Integer>() {
http://git-wip-us.apache.org/repos/asf/camel/blob/3dd80567/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 3774dc2..1187881 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
@@ -69,8 +69,8 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
}
@Override
- public SqlEndpoint getEndpoint() {
- return (SqlEndpoint) super.getEndpoint();
+ public DefaultSqlEndpoint getEndpoint() {
+ return (DefaultSqlEndpoint) super.getEndpoint();
}
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/3dd80567/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProcessingStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProcessingStrategy.java b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProcessingStrategy.java
index 6175bd8..6117cb2 100644
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProcessingStrategy.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlProcessingStrategy.java
@@ -35,7 +35,7 @@ public interface SqlProcessingStrategy {
* @return the update count if the query returned an update count
* @throws Exception can be thrown in case of error
*/
- int commit(SqlEndpoint endpoint, Exchange exchange, Object data, JdbcTemplate jdbcTemplate, String query) throws Exception;
+ int commit(DefaultSqlEndpoint endpoint, Exchange exchange, Object data, JdbcTemplate jdbcTemplate, String query) throws Exception;
/**
* Commit callback when the batch is complete. This allows you to do one extra query after all rows has been processed in the batch.
@@ -46,6 +46,6 @@ public interface SqlProcessingStrategy {
* @return the update count if the query returned an update count
* @throws Exception can be thrown in case of error
*/
- int commitBatchComplete(SqlEndpoint endpoint, JdbcTemplate jdbcTemplate, String query) throws Exception;
+ int commitBatchComplete(DefaultSqlEndpoint endpoint, JdbcTemplate jdbcTemplate, String query) throws Exception;
}
[08/10] camel git commit: CAMEL-9162: camel-elsql component
Posted by da...@apache.org.
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f6fcecf9
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f6fcecf9
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f6fcecf9
Branch: refs/heads/master
Commit: f6fcecf93d48740226197934b11ec6fc408aa407
Parents: f65b049
Author: Claus Ibsen <da...@apache.org>
Authored: Mon Oct 5 10:45:13 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 10:54:55 2015 +0200
----------------------------------------------------------------------
apache-camel/pom.xml | 4 ++++
apache-camel/src/main/descriptors/common-bin.xml | 1 +
parent/pom.xml | 5 +++++
3 files changed, 10 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/f6fcecf9/apache-camel/pom.xml
----------------------------------------------------------------------
diff --git a/apache-camel/pom.xml b/apache-camel/pom.xml
index 0510f6b..3bdeef1 100644
--- a/apache-camel/pom.xml
+++ b/apache-camel/pom.xml
@@ -213,6 +213,10 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-elsql</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-eventadmin</artifactId>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/camel/blob/f6fcecf9/apache-camel/src/main/descriptors/common-bin.xml
----------------------------------------------------------------------
diff --git a/apache-camel/src/main/descriptors/common-bin.xml b/apache-camel/src/main/descriptors/common-bin.xml
index e281d9d..857a801 100644
--- a/apache-camel/src/main/descriptors/common-bin.xml
+++ b/apache-camel/src/main/descriptors/common-bin.xml
@@ -66,6 +66,7 @@
<include>org.apache.camel:camel-eclipse</include>
<include>org.apache.camel:camel-ejb</include>
<include>org.apache.camel:camel-elasticsearch</include>
+ <include>org.apache.camel:camel-elsql</include>
<include>org.apache.camel:camel-eventadmin</include>
<include>org.apache.camel:camel-exec</include>
<include>org.apache.camel:camel-facebook</include>
http://git-wip-us.apache.org/repos/asf/camel/blob/f6fcecf9/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index 38e838f..b1c8655 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -838,6 +838,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-elsql</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-eventadmin</artifactId>
<version>${project.version}</version>
</dependency>
[04/10] camel git commit: CAMEL-9162: camel-elsql component
Posted by da...@apache.org.
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/0b71c42f
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/0b71c42f
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/0b71c42f
Branch: refs/heads/master
Commit: 0b71c42f2c2e1704fc61ca2bcd7385ef0ff84677
Parents: f6cefc7
Author: Claus Ibsen <da...@apache.org>
Authored: Mon Oct 5 08:21:20 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 10:54:54 2015 +0200
----------------------------------------------------------------------
.../camel/component/elsql/ElsqlConsumer.java | 4 +-
.../camel/component/elsql/ElsqlEndpoint.java | 6 +-
.../camel/component/elsql/ElsqlProducer.java | 3 +-
.../camel/component/sql/DefaultSqlEndpoint.java | 437 +++++++++++++++++++
.../apache/camel/component/sql/SqlConsumer.java | 2 +-
.../apache/camel/component/sql/SqlEndpoint.java | 428 +-----------------
6 files changed, 456 insertions(+), 424 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/0b71c42f/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
index 0f1c6d7..9459241 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
@@ -17,15 +17,15 @@
package org.apache.camel.component.elsql;
import org.apache.camel.Processor;
+import org.apache.camel.component.sql.DefaultSqlEndpoint;
import org.apache.camel.component.sql.SqlConsumer;
-import org.apache.camel.component.sql.SqlEndpoint;
import org.apache.camel.component.sql.SqlPrepareStatementStrategy;
import org.apache.camel.component.sql.SqlProcessingStrategy;
import org.springframework.jdbc.core.JdbcTemplate;
public class ElsqlConsumer extends SqlConsumer {
- public ElsqlConsumer(SqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate, String query,
+ public ElsqlConsumer(DefaultSqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate, String query,
SqlPrepareStatementStrategy sqlPrepareStatementStrategy, SqlProcessingStrategy sqlProcessingStrategy) {
super(endpoint, processor, jdbcTemplate, query, sqlPrepareStatementStrategy, sqlProcessingStrategy);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/0b71c42f/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
index 352dc37..a07b93e 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
@@ -24,7 +24,7 @@ import org.apache.camel.Component;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
-import org.apache.camel.component.sql.SqlEndpoint;
+import org.apache.camel.component.sql.DefaultSqlEndpoint;
import org.apache.camel.component.sql.SqlPrepareStatementStrategy;
import org.apache.camel.component.sql.SqlProcessingStrategy;
import org.apache.camel.spi.Metadata;
@@ -37,7 +37,7 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
@UriEndpoint(scheme = "elsql", title = "SQL", syntax = "elsql:elsqlName:resourceUri", consumerClass = ElsqlConsumer.class, label = "database,sql")
-public class ElsqlEndpoint extends SqlEndpoint {
+public class ElsqlEndpoint extends DefaultSqlEndpoint {
private volatile ElSql elSql;
private NamedParameterJdbcTemplate namedJdbcTemplate;
@@ -51,7 +51,7 @@ public class ElsqlEndpoint extends SqlEndpoint {
private ElSqlConfig elSqlConfig;
public ElsqlEndpoint(String uri, Component component, NamedParameterJdbcTemplate namedJdbcTemplate, String elsqlName, String resourceUri) {
- super(uri, component, null, null);
+ super(uri, component, null);
this.elsqlName = elsqlName;
this.resourceUri = resourceUri;
this.namedJdbcTemplate = namedJdbcTemplate;
http://git-wip-us.apache.org/repos/asf/camel/blob/0b71c42f/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
index fa3f5df..4353d9a 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
@@ -25,7 +25,6 @@ import com.opengamma.elsql.ElSql;
import com.opengamma.elsql.SpringSqlParams;
import org.apache.camel.Exchange;
import org.apache.camel.component.sql.SqlConstants;
-import org.apache.camel.component.sql.SqlEndpoint;
import org.apache.camel.component.sql.SqlOutputType;
import org.apache.camel.impl.DefaultProducer;
import org.springframework.dao.DataAccessException;
@@ -41,7 +40,7 @@ public class ElsqlProducer extends DefaultProducer {
private final String elSqlName;
private final NamedParameterJdbcTemplate jdbcTemplate;
- public ElsqlProducer(SqlEndpoint endpoint, ElSql elSql, String elSqlName, NamedParameterJdbcTemplate jdbcTemplate) {
+ public ElsqlProducer(ElsqlEndpoint endpoint, ElSql elSql, String elSqlName, NamedParameterJdbcTemplate jdbcTemplate) {
super(endpoint);
this.elSql = elSql;
this.elSqlName = elSqlName;
http://git-wip-us.apache.org/repos/asf/camel/blob/0b71c42f/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlEndpoint.java b/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlEndpoint.java
new file mode 100644
index 0000000..c67ecc1
--- /dev/null
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/DefaultSqlEndpoint.java
@@ -0,0 +1,437 @@
+/**
+ * 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 java.sql.ResultSet;
+import java.sql.SQLDataException;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Map;
+import javax.sql.DataSource;
+
+import org.apache.camel.Component;
+import org.apache.camel.impl.DefaultPollingEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.springframework.jdbc.core.BeanPropertyRowMapper;
+import org.springframework.jdbc.core.ColumnMapRowMapper;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.core.RowMapperResultSetExtractor;
+
+/**
+ * Base class for SQL endpoints.
+ */
+public abstract class DefaultSqlEndpoint extends DefaultPollingEndpoint {
+ private JdbcTemplate jdbcTemplate;
+
+ @UriParam(description = "Sets the reference to a DataSource to lookup from the registry, to use for communicating with the database.")
+ @Deprecated
+ private String dataSourceRef;
+ @UriParam(description = "Sets the DataSource to use to communicate with the database.")
+ private DataSource dataSource;
+ @UriParam(label = "producer", description = "Enables or disables batch mode")
+ private boolean batch;
+ @UriParam(label = "consumer", description = "Sets the maximum number of messages to poll")
+ private int maxMessagesPerPoll;
+ @UriParam(label = "consumer,advanced",
+ description = "Allows to plugin to use a custom org.apache.camel.component.sql.SqlProcessingStrategy to execute queries when the consumer has processed the rows/batch.")
+ private SqlProcessingStrategy processingStrategy;
+ @UriParam(label = "advanced",
+ description = "Allows to plugin to use a custom org.apache.camel.component.sql.SqlPrepareStatementStrategy to control preparation of the query and prepared statement.")
+ private SqlPrepareStatementStrategy prepareStatementStrategy;
+ @UriParam(label = "consumer",
+ description = "After processing each row then this query can be executed, if the Exchange was processed successfully, for example to mark the row as processed. The query can have parameter.")
+ private String onConsume;
+ @UriParam(label = "consumer",
+ description = "After processing each row then this query can be executed, if the Exchange failed, for example to mark the row as failed. The query can have parameter.")
+ private String onConsumeFailed;
+ @UriParam(label = "consumer",
+ description = "After processing the entire batch, this query can be executed to bulk update rows etc. The query cannot have parameters.")
+ private String onConsumeBatchComplete;
+ @UriParam(label = "consumer", defaultValue = "true",
+ description = "Sets how resultset should be delivered to route. Indicates delivery as either a list or individual object. defaults to true.")
+ private boolean useIterator = true;
+ @UriParam(label = "consumer",
+ description = "Sets whether empty resultset should be allowed to be sent to the next hop. Defaults to false. So the empty resultset will be filtered out.")
+ private boolean routeEmptyResultSet;
+ @UriParam(label = "consumer", defaultValue = "-1", description = "Sets an expected update count to validate when using onConsume.")
+ private int expectedUpdateCount = -1;
+ @UriParam(label = "consumer", description = "Sets whether to break batch if onConsume failed.")
+ private boolean breakBatchOnConsumeFail;
+ @UriParam(defaultValue = "true", description = "Whether to allow using named parameters in the queries.")
+ private boolean allowNamedParameters = true;
+ @UriParam(label = "producer,advanced",
+ description = "If enabled then the populateStatement method from org.apache.camel.component.sql.SqlPrepareStatementStrategy is always invoked, "
+ + "also if there is no expected parameters to be prepared. When this is false then the populateStatement is only invoked if there is 1"
+ + " or more expected parameters to be set; for example this avoids reading the message body/headers for SQL queries with no parameters.")
+ private boolean alwaysPopulateStatement;
+ @UriParam(defaultValue = ",",
+ description = "The separator to use when parameter values is taken from message body (if the body is a String type), to be inserted at # placeholders."
+ + "Notice if you use named parameters, then a Map type is used instead. The default value is ,")
+ private char separator = ',';
+ @UriParam(defaultValue = "SelectList", description = "Make the output of consumer or producer to SelectList as List of Map, or SelectOne as single Java object in the following way:"
+ + "a) If the query has only single column, then that JDBC Column object is returned. (such as SELECT COUNT( * ) FROM PROJECT will return a Long object."
+ + "b) If the query has more than one column, then it will return a Map of that result."
+ + "c) If the outputClass is set, then it will convert the query result into an Java bean object by calling all the setters that match the column names."
+ + "It will assume your class has a default constructor to create instance with."
+ + "d) If the query resulted in more than one rows, it throws an non-unique result exception.")
+ private SqlOutputType outputType = SqlOutputType.SelectList;
+ @UriParam(description = "Specify the full package and class name to use as conversion when outputType=SelectOne.")
+ private String outputClass;
+ @UriParam(label = "producer,advanced", description = "If set greater than zero, then Camel will use this count value of parameters to replace instead of"
+ + " querying via JDBC metadata API. This is useful if the JDBC vendor could not return correct parameters count, then user may override instead.")
+ private int parametersCount;
+ @UriParam(label = "producer", description = "If set, will ignore the results of the SQL query and use the existing IN message as the OUT message for the continuation of processing")
+ private boolean noop;
+ @UriParam(description = "Store the query result in a header instead of the message body. By default, outputHeader == null and the query result is stored"
+ + " in the message body, any existing content in the message body is discarded. If outputHeader is set, the value is used as the name of the header"
+ + " to store the query result and the original message body is preserved.")
+ private String outputHeader;
+ @UriParam(label = "producer", description = "Whether to use the message body as the SQL and then headers for parameters. If this option is enabled then the SQL in the uri is not used.")
+ private boolean useMessageBodyForSql;
+
+ public DefaultSqlEndpoint() {
+ }
+
+ public DefaultSqlEndpoint(String uri, Component component, JdbcTemplate jdbcTemplate) {
+ super(uri, component);
+ this.jdbcTemplate = jdbcTemplate;
+ }
+
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public JdbcTemplate getJdbcTemplate() {
+ return jdbcTemplate;
+ }
+
+ public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
+ this.jdbcTemplate = jdbcTemplate;
+ }
+
+ public boolean isBatch() {
+ return batch;
+ }
+
+ /**
+ * Enables or disables batch mode
+ */
+ public void setBatch(boolean batch) {
+ this.batch = batch;
+ }
+
+ public int getMaxMessagesPerPoll() {
+ return maxMessagesPerPoll;
+ }
+
+ /**
+ * Sets the maximum number of messages to poll
+ */
+ public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
+ this.maxMessagesPerPoll = maxMessagesPerPoll;
+ }
+
+ public SqlProcessingStrategy getProcessingStrategy() {
+ return processingStrategy;
+ }
+
+ /**
+ * Allows to plugin to use a custom org.apache.camel.component.sql.SqlProcessingStrategy to execute queries when the consumer has processed the rows/batch.
+ */
+ public void setProcessingStrategy(SqlProcessingStrategy processingStrategy) {
+ this.processingStrategy = processingStrategy;
+ }
+
+ public SqlPrepareStatementStrategy getPrepareStatementStrategy() {
+ return prepareStatementStrategy;
+ }
+
+ /**
+ * Allows to plugin to use a custom org.apache.camel.component.sql.SqlPrepareStatementStrategy to control preparation of the query and prepared statement.
+ */
+ public void setPrepareStatementStrategy(SqlPrepareStatementStrategy prepareStatementStrategy) {
+ this.prepareStatementStrategy = prepareStatementStrategy;
+ }
+
+ public String getOnConsume() {
+ return onConsume;
+ }
+
+ /**
+ * After processing each row then this query can be executed, if the Exchange was processed successfully, for example to mark the row as processed. The query can have parameter.
+ */
+ public void setOnConsume(String onConsume) {
+ this.onConsume = onConsume;
+ }
+
+ public String getOnConsumeFailed() {
+ return onConsumeFailed;
+ }
+
+ /**
+ * After processing each row then this query can be executed, if the Exchange failed, for example to mark the row as failed. The query can have parameter.
+ */
+ public void setOnConsumeFailed(String onConsumeFailed) {
+ this.onConsumeFailed = onConsumeFailed;
+ }
+
+ public String getOnConsumeBatchComplete() {
+ return onConsumeBatchComplete;
+ }
+
+ /**
+ * After processing the entire batch, this query can be executed to bulk update rows etc. The query cannot have parameters.
+ */
+ public void setOnConsumeBatchComplete(String onConsumeBatchComplete) {
+ this.onConsumeBatchComplete = onConsumeBatchComplete;
+ }
+
+ public boolean isAllowNamedParameters() {
+ return allowNamedParameters;
+ }
+
+ /**
+ * Whether to allow using named parameters in the queries.
+ */
+ public void setAllowNamedParameters(boolean allowNamedParameters) {
+ this.allowNamedParameters = allowNamedParameters;
+ }
+
+ public boolean isAlwaysPopulateStatement() {
+ return alwaysPopulateStatement;
+ }
+
+ /**
+ * If enabled then the populateStatement method from org.apache.camel.component.sql.SqlPrepareStatementStrategy is always invoked,
+ * also if there is no expected parameters to be prepared. When this is false then the populateStatement is only invoked if there
+ * is 1 or more expected parameters to be set; for example this avoids reading the message body/headers for SQL queries with no parameters.
+ */
+ public void setAlwaysPopulateStatement(boolean alwaysPopulateStatement) {
+ this.alwaysPopulateStatement = alwaysPopulateStatement;
+ }
+
+ public char getSeparator() {
+ return separator;
+ }
+
+ /**
+ * The separator to use when parameter values is taken from message body (if the body is a String type), to be inserted at # placeholders.
+ * Notice if you use named parameters, then a Map type is used instead.
+ * <p/>
+ * The default value is ,
+ */
+ public void setSeparator(char separator) {
+ this.separator = separator;
+ }
+
+ public SqlOutputType getOutputType() {
+ return outputType;
+ }
+
+ /**
+ * Make the output of consumer or producer to SelectList as List of Map, or SelectOne as single Java object in the following way:
+ * a) If the query has only single column, then that JDBC Column object is returned. (such as SELECT COUNT( * ) FROM PROJECT will return a Long object.
+ * b) If the query has more than one column, then it will return a Map of that result.
+ * c) If the outputClass is set, then it will convert the query result into an Java bean object by calling all the setters that match the column names.
+ * It will assume your class has a default constructor to create instance with.
+ * d) If the query resulted in more than one rows, it throws an non-unique result exception.
+ */
+ public void setOutputType(SqlOutputType outputType) {
+ this.outputType = outputType;
+ }
+
+ public String getOutputClass() {
+ return outputClass;
+ }
+
+ /**
+ * Specify the full package and class name to use as conversion when outputType=SelectOne.
+ */
+ public void setOutputClass(String outputClass) {
+ this.outputClass = outputClass;
+ }
+
+ public int getParametersCount() {
+ return parametersCount;
+ }
+
+ /**
+ * If set greater than zero, then Camel will use this count value of parameters to replace instead of querying via JDBC metadata API.
+ * This is useful if the JDBC vendor could not return correct parameters count, then user may override instead.
+ */
+ public void setParametersCount(int parametersCount) {
+ this.parametersCount = parametersCount;
+ }
+
+ public boolean isNoop() {
+ return noop;
+ }
+
+ /**
+ * If set, will ignore the results of the SQL query and use the existing IN message as the OUT message for the continuation of processing
+ */
+ public void setNoop(boolean noop) {
+ this.noop = noop;
+ }
+
+ public String getOutputHeader() {
+ return outputHeader;
+ }
+
+ /**
+ * Store the query result in a header instead of the message body.
+ * By default, outputHeader == null and the query result is stored in the message body,
+ * any existing content in the message body is discarded.
+ * If outputHeader is set, the value is used as the name of the header to store the
+ * query result and the original message body is preserved.
+ */
+ public void setOutputHeader(String outputHeader) {
+ this.outputHeader = outputHeader;
+ }
+
+ public boolean isUseMessageBodyForSql() {
+ return useMessageBodyForSql;
+ }
+
+ /**
+ * Whether to use the message body as the SQL and then headers for parameters.
+ * <p/>
+ * If this option is enabled then the SQL in the uri is not used.
+ */
+ public void setUseMessageBodyForSql(boolean useMessageBodyForSql) {
+ this.useMessageBodyForSql = useMessageBodyForSql;
+ }
+
+ public String getDataSourceRef() {
+ return dataSourceRef;
+ }
+
+ /**
+ * Sets the reference to a DataSource to lookup from the registry, to use for communicating with the database.
+ */
+ public void setDataSourceRef(String dataSourceRef) {
+ this.dataSourceRef = dataSourceRef;
+ }
+
+ public DataSource getDataSource() {
+ return dataSource;
+ }
+
+ /**
+ * Sets the DataSource to use to communicate with the database.
+ */
+ public void setDataSource(DataSource dataSource) {
+ this.dataSource = dataSource;
+ }
+
+ public boolean isUseIterator() {
+ return useIterator;
+ }
+
+ /**
+ * Sets how resultset should be delivered to route. Indicates delivery as either a list or individual object. defaults to true.
+ */
+ public void setUseIterator(boolean useIterator) {
+ this.useIterator = useIterator;
+ }
+
+ public boolean isRouteEmptyResultSet() {
+ return routeEmptyResultSet;
+ }
+
+ /**
+ * Sets whether empty resultset should be allowed to be sent to the next hop.
+ * Defaults to false. So the empty resultset will be filtered out.
+ */
+ public void setRouteEmptyResultSet(boolean routeEmptyResultSet) {
+ this.routeEmptyResultSet = routeEmptyResultSet;
+ }
+
+ public int getExpectedUpdateCount() {
+ return expectedUpdateCount;
+ }
+
+ /**
+ * Sets an expected update count to validate when using onConsume.
+ */
+ public void setExpectedUpdateCount(int expectedUpdateCount) {
+ this.expectedUpdateCount = expectedUpdateCount;
+ }
+
+ public boolean isBreakBatchOnConsumeFail() {
+ return breakBatchOnConsumeFail;
+ }
+
+ /**
+ * Sets whether to break batch if onConsume failed.
+ */
+ public void setBreakBatchOnConsumeFail(boolean breakBatchOnConsumeFail) {
+ this.breakBatchOnConsumeFail = breakBatchOnConsumeFail;
+ }
+
+ @SuppressWarnings("unchecked")
+ public List<?> queryForList(ResultSet rs, boolean allowMapToClass) throws SQLException {
+ if (allowMapToClass && outputClass != null) {
+ Class<?> outputClazz = getCamelContext().getClassResolver().resolveClass(outputClass);
+ RowMapper rowMapper = new BeanPropertyRowMapper(outputClazz);
+ RowMapperResultSetExtractor<?> mapper = new RowMapperResultSetExtractor(rowMapper);
+ List<?> data = mapper.extractData(rs);
+ return data;
+ } else {
+ ColumnMapRowMapper rowMapper = new ColumnMapRowMapper();
+ RowMapperResultSetExtractor<Map<String, Object>> mapper = new RowMapperResultSetExtractor<Map<String, Object>>(rowMapper);
+ List<Map<String, Object>> data = mapper.extractData(rs);
+ return data;
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public Object queryForObject(ResultSet rs) throws SQLException {
+ Object result = null;
+ if (outputClass == null) {
+ RowMapper rowMapper = new ColumnMapRowMapper();
+ RowMapperResultSetExtractor<Map<String, Object>> mapper = new RowMapperResultSetExtractor<Map<String, Object>>(rowMapper);
+ List<Map<String, Object>> data = mapper.extractData(rs);
+ if (data.size() > 1) {
+ throw new SQLDataException("Query result not unique for outputType=SelectOne. Got " + data.size() + " count instead.");
+ } else if (data.size() == 1) {
+ // Set content depend on number of column from query result
+ Map<String, Object> row = data.get(0);
+ if (row.size() == 1) {
+ result = row.values().iterator().next();
+ } else {
+ result = row;
+ }
+ }
+ } else {
+ Class<?> outputClzz = getCamelContext().getClassResolver().resolveClass(outputClass);
+ RowMapper rowMapper = new BeanPropertyRowMapper(outputClzz);
+ RowMapperResultSetExtractor<?> mapper = new RowMapperResultSetExtractor(rowMapper);
+ List<?> data = mapper.extractData(rs);
+ if (data.size() > 1) {
+ throw new SQLDataException("Query result not unique for outputType=SelectOne. Got " + data.size() + " count instead.");
+ } else if (data.size() == 1) {
+ result = data.get(0);
+ }
+ }
+
+ // If data.size is zero, let result be null.
+ return result;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/0b71c42f/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 29c3e07..3774dc2 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
@@ -59,7 +59,7 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
}
}
- public SqlConsumer(SqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate, String query,
+ public SqlConsumer(DefaultSqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate, String query,
SqlPrepareStatementStrategy sqlPrepareStatementStrategy, SqlProcessingStrategy sqlProcessingStrategy) {
super(endpoint, processor);
this.jdbcTemplate = jdbcTemplate;
http://git-wip-us.apache.org/repos/asf/camel/blob/0b71c42f/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 761c125..fdf049c 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
@@ -16,29 +16,15 @@
*/
package org.apache.camel.component.sql;
-import java.sql.ResultSet;
-import java.sql.SQLDataException;
-import java.sql.SQLException;
-import java.util.List;
-import java.util.Map;
-
-import javax.sql.DataSource;
-
import org.apache.camel.Component;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
-import org.apache.camel.impl.DefaultPollingEndpoint;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriEndpoint;
-import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.util.UnsafeUriCharactersEncoder;
-import org.springframework.jdbc.core.BeanPropertyRowMapper;
-import org.springframework.jdbc.core.ColumnMapRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.jdbc.core.RowMapper;
-import org.springframework.jdbc.core.RowMapperResultSetExtractor;
/**
* SQL Endpoint. Endpoint URI should contain valid SQL statement, but instead of
@@ -46,91 +32,23 @@ import org.springframework.jdbc.core.RowMapperResultSetExtractor;
* This is because in camel question mark has other meaning.
*/
@UriEndpoint(scheme = "sql", title = "SQL", syntax = "sql:query", consumerClass = SqlConsumer.class, label = "database,sql")
-public class SqlEndpoint extends DefaultPollingEndpoint {
- private JdbcTemplate jdbcTemplate;
+public class SqlEndpoint extends DefaultSqlEndpoint {
@UriPath(description = "Sets the SQL query to perform") @Metadata(required = "true")
-
private String query;
- @UriParam(description = "Sets the reference to a DataSource to lookup from the registry, to use for communicating with the database.")
- @Deprecated
- private String dataSourceRef;
- @UriParam(description = "Sets the DataSource to use to communicate with the database.")
- private DataSource dataSource;
- @UriParam(label = "producer", description = "Enables or disables batch mode")
- private boolean batch;
- @UriParam(label = "consumer", description = "Sets the maximum number of messages to poll")
- private int maxMessagesPerPoll;
- @UriParam(label = "consumer,advanced",
- description = "Allows to plugin to use a custom org.apache.camel.component.sql.SqlProcessingStrategy to execute queries when the consumer has processed the rows/batch.")
- private SqlProcessingStrategy processingStrategy;
- @UriParam(label = "advanced",
- description = "Allows to plugin to use a custom org.apache.camel.component.sql.SqlPrepareStatementStrategy to control preparation of the query and prepared statement.")
- private SqlPrepareStatementStrategy prepareStatementStrategy;
- @UriParam(label = "consumer",
- description = "After processing each row then this query can be executed, if the Exchange was processed successfully, for example to mark the row as processed. The query can have parameter.")
- private String onConsume;
- @UriParam(label = "consumer",
- description = "After processing each row then this query can be executed, if the Exchange failed, for example to mark the row as failed. The query can have parameter.")
- private String onConsumeFailed;
- @UriParam(label = "consumer",
- description = "After processing the entire batch, this query can be executed to bulk update rows etc. The query cannot have parameters.")
- private String onConsumeBatchComplete;
- @UriParam(label = "consumer", defaultValue = "true",
- description = "Sets how resultset should be delivered to route. Indicates delivery as either a list or individual object. defaults to true.")
- private boolean useIterator = true;
- @UriParam(label = "consumer",
- description = "Sets whether empty resultset should be allowed to be sent to the next hop. Defaults to false. So the empty resultset will be filtered out.")
- private boolean routeEmptyResultSet;
- @UriParam(label = "consumer", defaultValue = "-1", description = "Sets an expected update count to validate when using onConsume.")
- private int expectedUpdateCount = -1;
- @UriParam(label = "consumer", description = "Sets whether to break batch if onConsume failed.")
- private boolean breakBatchOnConsumeFail;
- @UriParam(defaultValue = "true", description = "Whether to allow using named parameters in the queries.")
- private boolean allowNamedParameters = true;
- @UriParam(label = "producer,advanced",
- description = "If enabled then the populateStatement method from org.apache.camel.component.sql.SqlPrepareStatementStrategy is always invoked, "
- + "also if there is no expected parameters to be prepared. When this is false then the populateStatement is only invoked if there is 1"
- + " or more expected parameters to be set; for example this avoids reading the message body/headers for SQL queries with no parameters.")
- private boolean alwaysPopulateStatement;
- @UriParam(defaultValue = ",",
- description = "The separator to use when parameter values is taken from message body (if the body is a String type), to be inserted at # placeholders."
- + "Notice if you use named parameters, then a Map type is used instead. The default value is ,")
- private char separator = ',';
- @UriParam(defaultValue = "SelectList", description = "Make the output of consumer or producer to SelectList as List of Map, or SelectOne as single Java object in the following way:"
- + "a) If the query has only single column, then that JDBC Column object is returned. (such as SELECT COUNT( * ) FROM PROJECT will return a Long object."
- + "b) If the query has more than one column, then it will return a Map of that result."
- + "c) If the outputClass is set, then it will convert the query result into an Java bean object by calling all the setters that match the column names."
- + "It will assume your class has a default constructor to create instance with."
- + "d) If the query resulted in more than one rows, it throws an non-unique result exception.")
- private SqlOutputType outputType = SqlOutputType.SelectList;
- @UriParam(description = "Specify the full package and class name to use as conversion when outputType=SelectOne.")
- private String outputClass;
- @UriParam(label = "producer,advanced", description = "If set greater than zero, then Camel will use this count value of parameters to replace instead of"
- + " querying via JDBC metadata API. This is useful if the JDBC vendor could not return correct parameters count, then user may override instead.")
- private int parametersCount;
- @UriParam(label = "producer", description = "If set, will ignore the results of the SQL query and use the existing IN message as the OUT message for the continuation of processing")
- private boolean noop;
- @UriParam(description = "Store the query result in a header instead of the message body. By default, outputHeader == null and the query result is stored"
- + " in the message body, any existing content in the message body is discarded. If outputHeader is set, the value is used as the name of the header"
- + " to store the query result and the original message body is preserved.")
- private String outputHeader;
- @UriParam(label = "producer", description = "Whether to use the message body as the SQL and then headers for parameters. If this option is enabled then the SQL in the uri is not used.")
- private boolean useMessageBodyForSql;
public SqlEndpoint() {
}
public SqlEndpoint(String uri, Component component, JdbcTemplate jdbcTemplate, String query) {
- super(uri, component);
- this.jdbcTemplate = jdbcTemplate;
+ super(uri, component, jdbcTemplate);
this.query = query;
}
public Consumer createConsumer(Processor processor) throws Exception {
- 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);
+ SqlPrepareStatementStrategy prepareStrategy = getPrepareStatementStrategy() != null ? getPrepareStatementStrategy() : new DefaultSqlPrepareStatementStrategy(getSeparator());
+ SqlProcessingStrategy proStrategy = getProcessingStrategy() != null ? getProcessingStrategy() : new DefaultSqlProcessingStrategy(prepareStrategy);
+ SqlConsumer consumer = new SqlConsumer(this, processor, getJdbcTemplate(), query, prepareStrategy, proStrategy);
consumer.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
consumer.setOnConsume(getOnConsume());
consumer.setOnConsumeFailed(getOnConsumeFailed());
@@ -144,22 +62,16 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
}
public Producer createProducer() throws Exception {
- SqlPrepareStatementStrategy prepareStrategy = prepareStatementStrategy != null ? prepareStatementStrategy : new DefaultSqlPrepareStatementStrategy(separator);
- SqlProducer result = new SqlProducer(this, query, jdbcTemplate, prepareStrategy, batch, alwaysPopulateStatement, useMessageBodyForSql);
- result.setParametersCount(parametersCount);
+ SqlPrepareStatementStrategy prepareStrategy = getPrepareStatementStrategy() != null ? getPrepareStatementStrategy() : new DefaultSqlPrepareStatementStrategy(getSeparator());
+ SqlProducer result = new SqlProducer(this, query, getJdbcTemplate(), prepareStrategy, isBatch(), isAlwaysPopulateStatement(), isUseMessageBodyForSql());
+ result.setParametersCount(getParametersCount());
return result;
}
- public boolean isSingleton() {
- return true;
- }
-
- public JdbcTemplate getJdbcTemplate() {
- return jdbcTemplate;
- }
-
- public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
- this.jdbcTemplate = jdbcTemplate;
+ @Override
+ protected String createEndpointUri() {
+ // Make sure it's properly encoded
+ return "sql:" + UnsafeUriCharactersEncoder.encode(query);
}
public String getQuery() {
@@ -173,320 +85,4 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
this.query = query;
}
- public boolean isBatch() {
- return batch;
- }
-
- /**
- * Enables or disables batch mode
- */
- public void setBatch(boolean batch) {
- this.batch = batch;
- }
-
- public int getMaxMessagesPerPoll() {
- return maxMessagesPerPoll;
- }
-
- /**
- * Sets the maximum number of messages to poll
- */
- public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
- this.maxMessagesPerPoll = maxMessagesPerPoll;
- }
-
- public SqlProcessingStrategy getProcessingStrategy() {
- return processingStrategy;
- }
-
- /**
- * Allows to plugin to use a custom org.apache.camel.component.sql.SqlProcessingStrategy to execute queries when the consumer has processed the rows/batch.
- */
- public void setProcessingStrategy(SqlProcessingStrategy processingStrategy) {
- this.processingStrategy = processingStrategy;
- }
-
- public SqlPrepareStatementStrategy getPrepareStatementStrategy() {
- return prepareStatementStrategy;
- }
-
- /**
- * Allows to plugin to use a custom org.apache.camel.component.sql.SqlPrepareStatementStrategy to control preparation of the query and prepared statement.
- */
- public void setPrepareStatementStrategy(SqlPrepareStatementStrategy prepareStatementStrategy) {
- this.prepareStatementStrategy = prepareStatementStrategy;
- }
-
- public String getOnConsume() {
- return onConsume;
- }
-
- /**
- * After processing each row then this query can be executed, if the Exchange was processed successfully, for example to mark the row as processed. The query can have parameter.
- */
- public void setOnConsume(String onConsume) {
- this.onConsume = onConsume;
- }
-
- public String getOnConsumeFailed() {
- return onConsumeFailed;
- }
-
- /**
- * After processing each row then this query can be executed, if the Exchange failed, for example to mark the row as failed. The query can have parameter.
- */
- public void setOnConsumeFailed(String onConsumeFailed) {
- this.onConsumeFailed = onConsumeFailed;
- }
-
- public String getOnConsumeBatchComplete() {
- return onConsumeBatchComplete;
- }
-
- /**
- * After processing the entire batch, this query can be executed to bulk update rows etc. The query cannot have parameters.
- */
- public void setOnConsumeBatchComplete(String onConsumeBatchComplete) {
- this.onConsumeBatchComplete = onConsumeBatchComplete;
- }
-
- public boolean isAllowNamedParameters() {
- return allowNamedParameters;
- }
-
- /**
- * Whether to allow using named parameters in the queries.
- */
- public void setAllowNamedParameters(boolean allowNamedParameters) {
- this.allowNamedParameters = allowNamedParameters;
- }
-
- public boolean isAlwaysPopulateStatement() {
- return alwaysPopulateStatement;
- }
-
- /**
- * If enabled then the populateStatement method from org.apache.camel.component.sql.SqlPrepareStatementStrategy is always invoked,
- * also if there is no expected parameters to be prepared. When this is false then the populateStatement is only invoked if there
- * is 1 or more expected parameters to be set; for example this avoids reading the message body/headers for SQL queries with no parameters.
- */
- public void setAlwaysPopulateStatement(boolean alwaysPopulateStatement) {
- this.alwaysPopulateStatement = alwaysPopulateStatement;
- }
-
- public char getSeparator() {
- return separator;
- }
-
- /**
- * The separator to use when parameter values is taken from message body (if the body is a String type), to be inserted at # placeholders.
- * Notice if you use named parameters, then a Map type is used instead.
- * <p/>
- * The default value is ,
- */
- public void setSeparator(char separator) {
- this.separator = separator;
- }
-
- public SqlOutputType getOutputType() {
- return outputType;
- }
-
- /**
- * Make the output of consumer or producer to SelectList as List of Map, or SelectOne as single Java object in the following way:
- * a) If the query has only single column, then that JDBC Column object is returned. (such as SELECT COUNT( * ) FROM PROJECT will return a Long object.
- * b) If the query has more than one column, then it will return a Map of that result.
- * c) If the outputClass is set, then it will convert the query result into an Java bean object by calling all the setters that match the column names.
- * It will assume your class has a default constructor to create instance with.
- * d) If the query resulted in more than one rows, it throws an non-unique result exception.
- */
- public void setOutputType(SqlOutputType outputType) {
- this.outputType = outputType;
- }
-
- public String getOutputClass() {
- return outputClass;
- }
-
- /**
- * Specify the full package and class name to use as conversion when outputType=SelectOne.
- */
- public void setOutputClass(String outputClass) {
- this.outputClass = outputClass;
- }
-
- public int getParametersCount() {
- return parametersCount;
- }
-
- /**
- * If set greater than zero, then Camel will use this count value of parameters to replace instead of querying via JDBC metadata API.
- * This is useful if the JDBC vendor could not return correct parameters count, then user may override instead.
- */
- public void setParametersCount(int parametersCount) {
- this.parametersCount = parametersCount;
- }
-
- public boolean isNoop() {
- return noop;
- }
-
- /**
- * If set, will ignore the results of the SQL query and use the existing IN message as the OUT message for the continuation of processing
- */
- public void setNoop(boolean noop) {
- this.noop = noop;
- }
-
- public String getOutputHeader() {
- return outputHeader;
- }
-
- /**
- * Store the query result in a header instead of the message body.
- * By default, outputHeader == null and the query result is stored in the message body,
- * any existing content in the message body is discarded.
- * If outputHeader is set, the value is used as the name of the header to store the
- * query result and the original message body is preserved.
- */
- public void setOutputHeader(String outputHeader) {
- this.outputHeader = outputHeader;
- }
-
- public boolean isUseMessageBodyForSql() {
- return useMessageBodyForSql;
- }
-
- /**
- * Whether to use the message body as the SQL and then headers for parameters.
- * <p/>
- * If this option is enabled then the SQL in the uri is not used.
- */
- public void setUseMessageBodyForSql(boolean useMessageBodyForSql) {
- this.useMessageBodyForSql = useMessageBodyForSql;
- }
-
- public String getDataSourceRef() {
- return dataSourceRef;
- }
-
- /**
- * Sets the reference to a DataSource to lookup from the registry, to use for communicating with the database.
- */
- public void setDataSourceRef(String dataSourceRef) {
- this.dataSourceRef = dataSourceRef;
- }
-
- public DataSource getDataSource() {
- return dataSource;
- }
-
- /**
- * Sets the DataSource to use to communicate with the database.
- */
- public void setDataSource(DataSource dataSource) {
- this.dataSource = dataSource;
- }
-
- public boolean isUseIterator() {
- return useIterator;
- }
-
- /**
- * Sets how resultset should be delivered to route. Indicates delivery as either a list or individual object. defaults to true.
- */
- public void setUseIterator(boolean useIterator) {
- this.useIterator = useIterator;
- }
-
- public boolean isRouteEmptyResultSet() {
- return routeEmptyResultSet;
- }
-
- /**
- * Sets whether empty resultset should be allowed to be sent to the next hop.
- * Defaults to false. So the empty resultset will be filtered out.
- */
- public void setRouteEmptyResultSet(boolean routeEmptyResultSet) {
- this.routeEmptyResultSet = routeEmptyResultSet;
- }
-
- public int getExpectedUpdateCount() {
- return expectedUpdateCount;
- }
-
- /**
- * Sets an expected update count to validate when using onConsume.
- */
- public void setExpectedUpdateCount(int expectedUpdateCount) {
- this.expectedUpdateCount = expectedUpdateCount;
- }
-
- public boolean isBreakBatchOnConsumeFail() {
- return breakBatchOnConsumeFail;
- }
-
- /**
- * Sets whether to break batch if onConsume failed.
- */
- public void setBreakBatchOnConsumeFail(boolean breakBatchOnConsumeFail) {
- this.breakBatchOnConsumeFail = breakBatchOnConsumeFail;
- }
-
- @Override
- protected String createEndpointUri() {
- // Make sure it's properly encoded
- return "sql:" + UnsafeUriCharactersEncoder.encode(query);
- }
-
- @SuppressWarnings("unchecked")
- public List<?> queryForList(ResultSet rs, boolean allowMapToClass) throws SQLException {
- if (allowMapToClass && outputClass != null) {
- Class<?> outputClazz = getCamelContext().getClassResolver().resolveClass(outputClass);
- RowMapper rowMapper = new BeanPropertyRowMapper(outputClazz);
- RowMapperResultSetExtractor<?> mapper = new RowMapperResultSetExtractor(rowMapper);
- List<?> data = mapper.extractData(rs);
- return data;
- } else {
- ColumnMapRowMapper rowMapper = new ColumnMapRowMapper();
- RowMapperResultSetExtractor<Map<String, Object>> mapper = new RowMapperResultSetExtractor<Map<String, Object>>(rowMapper);
- List<Map<String, Object>> data = mapper.extractData(rs);
- return data;
- }
- }
-
- @SuppressWarnings("unchecked")
- public Object queryForObject(ResultSet rs) throws SQLException {
- Object result = null;
- if (outputClass == null) {
- RowMapper rowMapper = new ColumnMapRowMapper();
- RowMapperResultSetExtractor<Map<String, Object>> mapper = new RowMapperResultSetExtractor<Map<String, Object>>(rowMapper);
- List<Map<String, Object>> data = mapper.extractData(rs);
- if (data.size() > 1) {
- throw new SQLDataException("Query result not unique for outputType=SelectOne. Got " + data.size() + " count instead.");
- } else if (data.size() == 1) {
- // Set content depend on number of column from query result
- Map<String, Object> row = data.get(0);
- if (row.size() == 1) {
- result = row.values().iterator().next();
- } else {
- result = row;
- }
- }
- } else {
- Class<?> outputClzz = getCamelContext().getClassResolver().resolveClass(outputClass);
- RowMapper rowMapper = new BeanPropertyRowMapper(outputClzz);
- RowMapperResultSetExtractor<?> mapper = new RowMapperResultSetExtractor(rowMapper);
- List<?> data = mapper.extractData(rs);
- if (data.size() > 1) {
- throw new SQLDataException("Query result not unique for outputType=SelectOne. Got " + data.size() + " count instead.");
- } else if (data.size() == 1) {
- result = data.get(0);
- }
- }
-
- // If data.size is zero, let result be null.
- return result;
- }
-
}
[05/10] camel git commit: CAMEL-9162: camel-elsql component
Posted by da...@apache.org.
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f65b0491
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f65b0491
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f65b0491
Branch: refs/heads/master
Commit: f65b0491f25a94089b547937d46ef5638580ee7a
Parents: 3dd8056
Author: Claus Ibsen <da...@apache.org>
Authored: Mon Oct 5 10:35:04 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 10:54:54 2015 +0200
----------------------------------------------------------------------
.../camel/component/elsql/ElsqlComponent.java | 24 ++---
.../camel/component/elsql/ElsqlConsumer.java | 7 +-
.../camel/component/elsql/ElsqlEndpoint.java | 16 ++-
.../camel/component/elsql/ElsqlProducer.java | 5 +-
.../component/elsql/ElsqlSqlMapSource.java | 49 ++++++---
.../elsql/ElsqlSqlProcessingStrategy.java | 51 ++++++++--
.../elsql/ElSqlConsumerDeleteTest.java | 100 +++++++++++++++++++
.../src/test/resources/elsql/projects.elsql | 4 +
.../src/test/resources/log4j.properties | 2 +-
.../apache/camel/component/sql/SqlConsumer.java | 48 +++++++--
.../sql/SqlNamedProcessingStrategy.java | 57 +++++++++++
11 files changed, 307 insertions(+), 56 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
index 40e6530..51142e8 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
@@ -70,34 +70,23 @@ public class ElsqlComponent extends UriEndpointComponent {
throw new IllegalArgumentException("Invalid uri. Must by elsql:elsqlName/resourceUri, was: " + uri);
}
- /* TODO: add this later
String onConsume = getAndRemoveParameter(parameters, "consumer.onConsume", String.class);
if (onConsume == null) {
onConsume = getAndRemoveParameter(parameters, "onConsume", String.class);
}
- if (onConsume != null && isUsePlaceholder()) {
- onConsume = onConsume.replaceAll(parameterPlaceholderSubstitute, "?");
- }
String onConsumeFailed = getAndRemoveParameter(parameters, "consumer.onConsumeFailed", String.class);
if (onConsumeFailed == null) {
onConsumeFailed = getAndRemoveParameter(parameters, "onConsumeFailed", String.class);
}
- if (onConsumeFailed != null && isUsePlaceholder()) {
- onConsumeFailed = onConsumeFailed.replaceAll(parameterPlaceholderSubstitute, "?");
- }
String onConsumeBatchComplete = getAndRemoveParameter(parameters, "consumer.onConsumeBatchComplete", String.class);
if (onConsumeBatchComplete == null) {
onConsumeBatchComplete = getAndRemoveParameter(parameters, "onConsumeBatchComplete", String.class);
}
- if (onConsumeBatchComplete != null && isUsePlaceholder()) {
- onConsumeBatchComplete = onConsumeBatchComplete.replaceAll(parameterPlaceholderSubstitute, "?");
- }
- */
ElsqlEndpoint endpoint = new ElsqlEndpoint(uri, this, jdbcTemplate, elsqlName, resUri);
-// endpoint.setOnConsume(onConsume);
-// endpoint.setOnConsumeFailed(onConsumeFailed);
-// endpoint.setOnConsumeBatchComplete(onConsumeBatchComplete);
+ endpoint.setOnConsume(onConsume);
+ endpoint.setOnConsumeFailed(onConsumeFailed);
+ endpoint.setOnConsumeBatchComplete(onConsumeBatchComplete);
endpoint.setDataSource(ds);
endpoint.setDataSourceRef(dataSourceRef);
endpoint.setElSqlConfig(elSqlConfig);
@@ -118,6 +107,13 @@ public class ElsqlComponent extends UriEndpointComponent {
super.doStop();
}
+ /**
+ * Sets the DataSource to use to communicate with the database.
+ */
+ public void setDataSource(DataSource dataSource) {
+ this.dataSource = dataSource;
+ }
+
public DataSource getDataSource() {
return dataSource;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
index 9459241..530dc23 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
@@ -21,12 +21,13 @@ import org.apache.camel.component.sql.DefaultSqlEndpoint;
import org.apache.camel.component.sql.SqlConsumer;
import org.apache.camel.component.sql.SqlPrepareStatementStrategy;
import org.apache.camel.component.sql.SqlProcessingStrategy;
-import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
public class ElsqlConsumer extends SqlConsumer {
- public ElsqlConsumer(DefaultSqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate, String query,
+ public ElsqlConsumer(DefaultSqlEndpoint endpoint, Processor processor, NamedParameterJdbcTemplate namedJdbcTemplate, String query, SqlParameterSource parameterSource,
SqlPrepareStatementStrategy sqlPrepareStatementStrategy, SqlProcessingStrategy sqlProcessingStrategy) {
- super(endpoint, processor, jdbcTemplate, query, sqlPrepareStatementStrategy, sqlProcessingStrategy);
+ super(endpoint, processor, namedJdbcTemplate, query, parameterSource, sqlPrepareStatementStrategy, sqlProcessingStrategy);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
index a07b93e..d2b2cbf 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
@@ -20,6 +20,7 @@ import java.net.URL;
import com.opengamma.elsql.ElSql;
import com.opengamma.elsql.ElSqlConfig;
+import com.opengamma.elsql.SpringSqlParams;
import org.apache.camel.Component;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
@@ -33,12 +34,17 @@ import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ResourceHelper;
-import org.springframework.jdbc.core.JdbcTemplate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.jdbc.core.namedparam.EmptySqlParameterSource;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
@UriEndpoint(scheme = "elsql", title = "SQL", syntax = "elsql:elsqlName:resourceUri", consumerClass = ElsqlConsumer.class, label = "database,sql")
public class ElsqlEndpoint extends DefaultSqlEndpoint {
+ private static final Logger LOG = LoggerFactory.getLogger(ElsqlEndpoint.class);
+
private volatile ElSql elSql;
private NamedParameterJdbcTemplate namedJdbcTemplate;
@@ -59,12 +65,14 @@ public class ElsqlEndpoint extends DefaultSqlEndpoint {
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- SqlProcessingStrategy proStrategy = new ElsqlSqlProcessingStrategy(elsqlName, elSql);
+ SqlProcessingStrategy proStrategy = new ElsqlSqlProcessingStrategy(elSql);
SqlPrepareStatementStrategy preStategy = new ElsqlSqlPrepareStatementStrategy();
- JdbcTemplate template = new JdbcTemplate(getDataSource());
+ final SqlParameterSource param = new EmptySqlParameterSource();
+ final String sql = elSql.getSql(elsqlName, new SpringSqlParams(param));
+ LOG.debug("ElsqlConsumer @{} using sql: {}", elsqlName, sql);
- ElsqlConsumer consumer = new ElsqlConsumer(this, processor, template, elsqlName, preStategy, proStrategy);
+ ElsqlConsumer consumer = new ElsqlConsumer(this, processor, namedJdbcTemplate, sql, param, preStategy, proStrategy);
consumer.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
consumer.setOnConsume(getOnConsume());
consumer.setOnConsumeFailed(getOnConsumeFailed());
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
index 4353d9a..78d0d2e 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
@@ -27,6 +27,8 @@ import org.apache.camel.Exchange;
import org.apache.camel.component.sql.SqlConstants;
import org.apache.camel.component.sql.SqlOutputType;
import org.apache.camel.impl.DefaultProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.PreparedStatementCallback;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
@@ -36,6 +38,7 @@ import static org.springframework.jdbc.support.JdbcUtils.closeResultSet;
public class ElsqlProducer extends DefaultProducer {
+ private static final Logger LOG = LoggerFactory.getLogger(ElsqlProducer.class);
private final ElSql elSql;
private final String elSqlName;
private final NamedParameterJdbcTemplate jdbcTemplate;
@@ -58,7 +61,7 @@ public class ElsqlProducer extends DefaultProducer {
final SqlParameterSource param = new ElsqlSqlMapSource(exchange, data);
final String sql = elSql.getSql(elSqlName, new SpringSqlParams(param));
- log.debug("ElSql @{} using sql: {}", elSqlName, sql);
+ LOG.debug("ElsqlProducer @{} using sql: {}", elSqlName, sql);
jdbcTemplate.execute(sql, param, new PreparedStatementCallback<Object>() {
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
index e8035b0..aec8d46 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
@@ -20,10 +20,16 @@ import java.util.Collections;
import java.util.Map;
import org.apache.camel.Exchange;
-import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.jdbc.core.namedparam.AbstractSqlParameterSource;
-public class ElsqlSqlMapSource extends MapSqlParameterSource {
+/**
+ * A {@link org.springframework.jdbc.core.namedparam.SqlParameterSource} that is used by {@link com.opengamma.elsql.ElSql}
+ * to lookup parameter values. This source will lookup in the Camel {@link Exchange} and {@link org.apache.camel.Message}
+ * assuming they are Map based.
+ */
+public class ElsqlSqlMapSource extends AbstractSqlParameterSource {
+ // use the maps from the Camel Message as they are case insensitive which makes it easier for end users to work with
private final Exchange exchange;
private final Map<?, ?> bodyMap;
private final Map<?, ?> headersMap;
@@ -32,23 +38,36 @@ public class ElsqlSqlMapSource extends MapSqlParameterSource {
this.exchange = exchange;
this.bodyMap = safeMap(exchange.getContext().getTypeConverter().tryConvertTo(Map.class, body));
this.headersMap = safeMap(exchange.getIn().getHeaders());
-
- addValue("body", body);
-
- for (Map.Entry<?, ?> entry : bodyMap.entrySet()) {
- String name = entry.getKey().toString();
- Object value = entry.getValue();
- addValue(name, value);
- }
- for (Map.Entry<?, ?> entry : headersMap.entrySet()) {
- String name = entry.getKey().toString();
- Object value = entry.getValue();
- addValue(name, value);
- }
}
private static Map<?, ?> safeMap(Map<?, ?> map) {
return (map == null || map.isEmpty()) ? Collections.emptyMap() : map;
}
+ @Override
+ public boolean hasValue(String paramName) {
+ if ("body".equals(paramName)) {
+ return true;
+ } else if (bodyMap.containsKey(paramName)) {
+ return true;
+ } else if (headersMap.containsKey(paramName)) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public Object getValue(String paramName) throws IllegalArgumentException {
+ Object answer;
+ if ("body".equals(paramName)) {
+ answer = exchange.getIn().getBody();
+ } else {
+ answer = bodyMap.get(paramName);
+ if (answer == null) {
+ headersMap.get(paramName);
+ }
+ }
+ return answer;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
index ea933d8..4180edd 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
@@ -23,32 +23,56 @@ import com.opengamma.elsql.ElSql;
import com.opengamma.elsql.SpringSqlParams;
import org.apache.camel.Exchange;
import org.apache.camel.component.sql.DefaultSqlEndpoint;
-import org.apache.camel.component.sql.SqlProcessingStrategy;
+import org.apache.camel.component.sql.SqlNamedProcessingStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementCallback;
+import org.springframework.jdbc.core.namedparam.EmptySqlParameterSource;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
-public class ElsqlSqlProcessingStrategy implements SqlProcessingStrategy {
+public class ElsqlSqlProcessingStrategy implements SqlNamedProcessingStrategy {
private static final Logger LOG = LoggerFactory.getLogger(ElsqlSqlProcessingStrategy.class);
- private final String elSqlName;
private final ElSql elSql;
- public ElsqlSqlProcessingStrategy(String elSqlName, ElSql elSql) {
- this.elSqlName = elSqlName;
+ public ElsqlSqlProcessingStrategy(ElSql elSql) {
this.elSql = elSql;
}
@Override
- public int commit(final DefaultSqlEndpoint endpoint, final Exchange exchange, final Object data, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
+ public int commit(DefaultSqlEndpoint defaultSqlEndpoint, Exchange exchange, Object data, NamedParameterJdbcTemplate jdbcTemplate,
+ SqlParameterSource parameterSource, String query) throws Exception {
+
final SqlParameterSource param = new ElsqlSqlMapSource(exchange, data);
- final String sql = elSql.getSql(elSqlName, new SpringSqlParams(param));
- LOG.debug("ElSql @{} using sql: {}", elSqlName, sql);
+ final String sql = elSql.getSql(query, new SpringSqlParams(param));
+ LOG.debug("commit @{} using sql: {}", query, sql);
+
+ return jdbcTemplate.execute(sql, param, new PreparedStatementCallback<Integer>() {
+ @Override
+ public Integer doInPreparedStatement(PreparedStatement ps) throws SQLException, DataAccessException {
+ ps.execute();
+
+ int updateCount = ps.getUpdateCount();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Update count {}", updateCount);
+ }
+ return updateCount;
+ }
+ });
+ }
+
+ @Override
+ public int commitBatchComplete(DefaultSqlEndpoint endpoint, NamedParameterJdbcTemplate namedJdbcTemplate,
+ SqlParameterSource parameterSource, String query) throws Exception {
+
+ final SqlParameterSource param = new EmptySqlParameterSource();
+ final String sql = elSql.getSql(query, new SpringSqlParams(param));
+ LOG.debug("commitBatchComplete @{} using sql: {}", query, sql);
- return jdbcTemplate.execute(sql, new PreparedStatementCallback<Integer>() {
+ return namedJdbcTemplate.execute(sql, param, new PreparedStatementCallback<Integer>() {
@Override
public Integer doInPreparedStatement(PreparedStatement ps) throws SQLException, DataAccessException {
ps.execute();
@@ -63,7 +87,12 @@ public class ElsqlSqlProcessingStrategy implements SqlProcessingStrategy {
}
@Override
- public int commitBatchComplete(final DefaultSqlEndpoint endpoint, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
- return 0;
+ public int commit(DefaultSqlEndpoint defaultSqlEndpoint, Exchange exchange, Object data, JdbcTemplate jdbcTemplate, String query) throws Exception {
+ throw new UnsupportedOperationException("Should not be called");
+ }
+
+ @Override
+ public int commitBatchComplete(DefaultSqlEndpoint defaultSqlEndpoint, JdbcTemplate jdbcTemplate, String query) throws Exception {
+ throw new UnsupportedOperationException("Should not be called");
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlConsumerDeleteTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlConsumerDeleteTest.java b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlConsumerDeleteTest.java
new file mode 100644
index 0000000..a381bf5
--- /dev/null
+++ b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlConsumerDeleteTest.java
@@ -0,0 +1,100 @@
+/**
+ * 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.elsql;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.Exchange;
+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;
+
+/**
+ *
+ */
+public class ElSqlConsumerDeleteTest 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 testConsume() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(3);
+
+ assertMockEndpointsSatisfied();
+
+ List<Exchange> exchanges = mock.getReceivedExchanges();
+ assertEquals(3, exchanges.size());
+
+ assertEquals(1, exchanges.get(0).getIn().getBody(Map.class).get("ID"));
+ assertEquals("Camel", exchanges.get(0).getIn().getBody(Map.class).get("PROJECT"));
+ assertEquals(2, exchanges.get(1).getIn().getBody(Map.class).get("ID"));
+ assertEquals("AMQ", exchanges.get(1).getIn().getBody(Map.class).get("PROJECT"));
+ assertEquals(3, exchanges.get(2).getIn().getBody(Map.class).get("ID"));
+ assertEquals("Linux", exchanges.get(2).getIn().getBody(Map.class).get("PROJECT"));
+
+ // some servers may be a bit slow for this
+ for (int i = 0; i < 5; i++) {
+ // give it a little tine to delete
+ Thread.sleep(1000);
+ int rows = jdbcTemplate.queryForObject("select count(*) from projects", Integer.class);
+ if (rows == 0) {
+ break;
+ }
+ }
+ assertEquals("Should have deleted all 3 rows", new Integer(0), jdbcTemplate.queryForObject("select count(*) from projects", Integer.class));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ getContext().getComponent("elsql", ElsqlComponent.class).setDataSource(db);
+
+ from("elsql:allProjects:elsql/projects.elsql?consumer.onConsume=deleteProject")
+ .to("mock:result");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/components/camel-elsql/src/test/resources/elsql/projects.elsql
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/resources/elsql/projects.elsql b/components/camel-elsql/src/test/resources/elsql/projects.elsql
index de60eef..1d957ef 100644
--- a/components/camel-elsql/src/test/resources/elsql/projects.elsql
+++ b/components/camel-elsql/src/test/resources/elsql/projects.elsql
@@ -7,3 +7,7 @@
SELECT *
FROM projects
ORDER BY id
+@NAME(deleteProject)
+ DELETE
+ FROM projects
+ WHERE id = :id
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/components/camel-elsql/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/resources/log4j.properties b/components/camel-elsql/src/test/resources/log4j.properties
index d5af410..82e5ef4 100755
--- a/components/camel-elsql/src/test/resources/log4j.properties
+++ b/components/camel-elsql/src/test/resources/log4j.properties
@@ -18,7 +18,7 @@
#
# The logging properties used for testing
#
-log4j.rootLogger=INFO, out
+log4j.rootLogger=INFO, file
#log4j.logger.org.apache.camel.component.sql=DEBUG
#log4j.logger.org.apache.camel.component.sql=TRACE
http://git-wip-us.apache.org/repos/asf/camel/blob/f65b0491/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 1187881..0f52280 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
@@ -33,6 +33,8 @@ import org.apache.camel.util.ObjectHelper;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementCallback;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import static org.springframework.jdbc.support.JdbcUtils.closeResultSet;
@@ -40,6 +42,8 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
private final String query;
private final JdbcTemplate jdbcTemplate;
+ private final NamedParameterJdbcTemplate namedJdbcTemplate;
+ private final SqlParameterSource parameterSource;
private final SqlPrepareStatementStrategy sqlPrepareStatementStrategy;
private final SqlProcessingStrategy sqlProcessingStrategy;
@@ -59,11 +63,23 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
}
}
- public SqlConsumer(DefaultSqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate, String query,
- SqlPrepareStatementStrategy sqlPrepareStatementStrategy, SqlProcessingStrategy sqlProcessingStrategy) {
+ public SqlConsumer(DefaultSqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate, String query, SqlPrepareStatementStrategy sqlPrepareStatementStrategy, SqlProcessingStrategy sqlProcessingStrategy) {
super(endpoint, processor);
this.jdbcTemplate = jdbcTemplate;
+ this.namedJdbcTemplate = null;
+ this.query = query;
+ this.parameterSource = null;
+ this.sqlPrepareStatementStrategy = sqlPrepareStatementStrategy;
+ this.sqlProcessingStrategy = sqlProcessingStrategy;
+ }
+
+ public SqlConsumer(DefaultSqlEndpoint endpoint, Processor processor, NamedParameterJdbcTemplate namedJdbcTemplate, String query, SqlParameterSource parameterSource,
+ SqlPrepareStatementStrategy sqlPrepareStatementStrategy, SqlProcessingStrategy sqlProcessingStrategy) {
+ super(endpoint, processor);
+ this.jdbcTemplate = null;
+ this.namedJdbcTemplate = namedJdbcTemplate;
this.query = query;
+ this.parameterSource = parameterSource;
this.sqlPrepareStatementStrategy = sqlPrepareStatementStrategy;
this.sqlProcessingStrategy = sqlProcessingStrategy;
}
@@ -80,8 +96,7 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
pendingExchanges = 0;
final String preparedQuery = sqlPrepareStatementStrategy.prepareQuery(query, getEndpoint().isAllowNamedParameters());
-
- Integer messagePolled = jdbcTemplate.execute(preparedQuery, new PreparedStatementCallback<Integer>() {
+ final PreparedStatementCallback<Integer> callback = new PreparedStatementCallback<Integer>() {
@Override
public Integer doInPreparedStatement(PreparedStatement preparedStatement) throws SQLException, DataAccessException {
Queue<DataHolder> answer = new LinkedList<DataHolder>();
@@ -114,7 +129,14 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
throw ObjectHelper.wrapRuntimeCamelException(e);
}
}
- });
+ };
+
+ Integer messagePolled;
+ if (namedJdbcTemplate != null) {
+ messagePolled = namedJdbcTemplate.execute(preparedQuery, parameterSource, callback);
+ } else {
+ messagePolled = jdbcTemplate.execute(preparedQuery, callback);
+ }
return messagePolled;
}
@@ -189,7 +211,13 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
try {
// we can only run on consume if there was data
if (data != null && sql != null) {
- int updateCount = sqlProcessingStrategy.commit(getEndpoint(), exchange, data, jdbcTemplate, sql);
+ int updateCount;
+ if (namedJdbcTemplate != null && sqlProcessingStrategy instanceof SqlNamedProcessingStrategy) {
+ SqlNamedProcessingStrategy namedProcessingStrategy = (SqlNamedProcessingStrategy) sqlProcessingStrategy;
+ updateCount = namedProcessingStrategy.commit(getEndpoint(), exchange, data, namedJdbcTemplate, parameterSource, sql);
+ } else {
+ 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);
@@ -206,7 +234,13 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
try {
if (onConsumeBatchComplete != null) {
- int updateCount = sqlProcessingStrategy.commitBatchComplete(getEndpoint(), jdbcTemplate, onConsumeBatchComplete);
+ int updateCount;
+ if (namedJdbcTemplate != null && sqlProcessingStrategy instanceof SqlNamedProcessingStrategy) {
+ SqlNamedProcessingStrategy namedProcessingStrategy = (SqlNamedProcessingStrategy) sqlProcessingStrategy;
+ updateCount = namedProcessingStrategy.commitBatchComplete(getEndpoint(), namedJdbcTemplate, parameterSource, onConsumeBatchComplete);
+ } else {
+ 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/f65b0491/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlNamedProcessingStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlNamedProcessingStrategy.java b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlNamedProcessingStrategy.java
new file mode 100644
index 0000000..cae9389
--- /dev/null
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlNamedProcessingStrategy.java
@@ -0,0 +1,57 @@
+/**
+ * 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.Exchange;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.jdbc.core.namedparam.SqlParameterSource;
+
+/**
+ * Extended processing strategy for dealing with SQL when consuming, which uses a {@link NamedParameterJdbcTemplate}
+ * instead of plain {@link org.springframework.jdbc.core.JdbcTemplate}.
+ */
+public interface SqlNamedProcessingStrategy extends SqlProcessingStrategy {
+
+ /**
+ * Commit callback if there are a query to be run after processing.
+ *
+ * @param endpoint the endpoint
+ * @param exchange The exchange after it has been processed
+ * @param data The original data delivered to the route
+ * @param namedJdbcTemplate The JDBC template
+ * @param parameterSource Parameter sources for the named JDBC template
+ * @param query The SQL query to execute
+ * @return the update count if the query returned an update count
+ * @throws Exception can be thrown in case of error
+ */
+ int commit(DefaultSqlEndpoint endpoint, Exchange exchange, Object data,
+ NamedParameterJdbcTemplate namedJdbcTemplate, SqlParameterSource parameterSource, String query) throws Exception;
+
+ /**
+ * Commit callback when the batch is complete. This allows you to do one extra query after all rows has been processed in the batch.
+ *
+ * @param endpoint the endpoint
+ * @param namedJdbcTemplate The JDBC template
+ * @param parameterSource Parameter sources for the named JDBC template
+ * @param query The SQL query to execute
+ * @return the update count if the query returned an update count
+ * @throws Exception can be thrown in case of error
+ */
+ int commitBatchComplete(DefaultSqlEndpoint endpoint, NamedParameterJdbcTemplate namedJdbcTemplate,
+ SqlParameterSource parameterSource, String query) throws Exception;
+
+}
[07/10] camel git commit: CAMEL-9162: camel-elsql component
Posted by da...@apache.org.
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/7241cadb
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/7241cadb
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/7241cadb
Branch: refs/heads/master
Commit: 7241cadbec7096a6536a5e57e6260c7bfc3c45d3
Parents: f6fcecf
Author: Claus Ibsen <da...@apache.org>
Authored: Mon Oct 5 10:53:04 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 10:54:55 2015 +0200
----------------------------------------------------------------------
components/camel-elsql/pom.xml | 2 +-
parent/pom.xml | 1 +
.../features/src/main/resources/features.xml | 5 +++
.../camel/itest/karaf/CamelElsqlTest.java | 40 ++++++++++++++++++++
4 files changed, 47 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/7241cadb/components/camel-elsql/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-elsql/pom.xml b/components/camel-elsql/pom.xml
index f45f7e6..8b22a3b 100644
--- a/components/camel-elsql/pom.xml
+++ b/components/camel-elsql/pom.xml
@@ -56,7 +56,7 @@
<dependency>
<groupId>com.opengamma</groupId>
<artifactId>elsql</artifactId>
- <version>1.1</version>
+ <version>${elsql-version}</version>
</dependency>
<!-- test dependencies -->
http://git-wip-us.apache.org/repos/asf/camel/blob/7241cadb/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index b1c8655..c3a30f8 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -146,6 +146,7 @@
<egit-github-core-bundle-version>2.1.5_1</egit-github-core-bundle-version>
<elasticsearch-bundle-version>1.7.1_1</elasticsearch-bundle-version>
<elasticsearch-version>1.7.1</elasticsearch-version>
+ <elsql-version>1.1</elsql-version>
<el-api-1.0-version>1.0.1</el-api-1.0-version>
<embedmongo-version>1.50.0</embedmongo-version>
<exec-maven-plugin-version>1.2.1</exec-maven-plugin-version>
http://git-wip-us.apache.org/repos/asf/camel/blob/7241cadb/platforms/karaf/features/src/main/resources/features.xml
----------------------------------------------------------------------
diff --git a/platforms/karaf/features/src/main/resources/features.xml b/platforms/karaf/features/src/main/resources/features.xml
index c42e201..083a478 100644
--- a/platforms/karaf/features/src/main/resources/features.xml
+++ b/platforms/karaf/features/src/main/resources/features.xml
@@ -381,6 +381,11 @@
<bundle dependency='true'>wrap:mvn:com.squareup.okio/okio/${squareup-okio-version}</bundle>
<bundle>mvn:org.apache.camel/camel-dropbox/${project.version}</bundle>
</feature>
+ <feature name='camel-elsql' version='${project.version}' resolver='(obr)' start-level='50'>
+ <feature version="${project.version}">camel-sql</feature>
+ <bundle dependency="true">wrap:mvn:com.opengamma/elsql/${elsql-version}</bundle>
+ <bundle>mvn:org.apache.camel/camel-elsql/${project.version}</bundle>
+ </feature>
<feature name='camel-elasticsearch' version='${project.version}' resolver='(obr)' start-level='50'>
<feature version='${project.version}'>camel-core</feature>
<feature>http</feature>
http://git-wip-us.apache.org/repos/asf/camel/blob/7241cadb/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelElsqlTest.java
----------------------------------------------------------------------
diff --git a/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelElsqlTest.java b/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelElsqlTest.java
new file mode 100644
index 0000000..78eb7b2
--- /dev/null
+++ b/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelElsqlTest.java
@@ -0,0 +1,40 @@
+/**
+ * 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.itest.karaf;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+
+@RunWith(PaxExam.class)
+public class CamelElsqlTest extends AbstractFeatureTest {
+
+ public static final String COMPONENT = extractName(CamelElsqlTest.class);
+
+ @Test
+ public void test() throws Exception {
+ testComponent(COMPONENT);
+ }
+
+ @Configuration
+ public static Option[] configure() {
+ return configure(COMPONENT);
+ }
+
+}
[10/10] camel git commit: CAMEL-9162: camel-elsql component
Posted by da...@apache.org.
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/5fc96ef4
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/5fc96ef4
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/5fc96ef4
Branch: refs/heads/master
Commit: 5fc96ef456ec86736b323ef1fdaa9fadf1d750de
Parents: 913efcc
Author: Claus Ibsen <da...@apache.org>
Authored: Mon Oct 5 11:52:16 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 11:52:16 2015 +0200
----------------------------------------------------------------------
.../camel/component/elsql/ElSqlDatabaseVendor.java | 16 ++++++++++++++++
.../camel/component/elsql/ElsqlSqlMapSource.java | 6 +-----
2 files changed, 17 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/5fc96ef4/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElSqlDatabaseVendor.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElSqlDatabaseVendor.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElSqlDatabaseVendor.java
index 28ce85a..a7405d6 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElSqlDatabaseVendor.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElSqlDatabaseVendor.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.elsql;
import com.opengamma.elsql.ElSqlConfig;
http://git-wip-us.apache.org/repos/asf/camel/blob/5fc96ef4/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
index aec8d46..725022e 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlMapSource.java
@@ -48,12 +48,8 @@ public class ElsqlSqlMapSource extends AbstractSqlParameterSource {
public boolean hasValue(String paramName) {
if ("body".equals(paramName)) {
return true;
- } else if (bodyMap.containsKey(paramName)) {
- return true;
- } else if (headersMap.containsKey(paramName)) {
- return true;
} else {
- return false;
+ return bodyMap.containsKey(paramName) || headersMap.containsKey(paramName);
}
}
[02/10] camel git commit: CAMEL-9162: camel-elsql component
Posted by da...@apache.org.
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/6cf99562
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/6cf99562
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/6cf99562
Branch: refs/heads/master
Commit: 6cf9956245fa59208d113f0045d31c82ae2afa98
Parents: f5614ef
Author: Claus Ibsen <da...@apache.org>
Authored: Sun Oct 4 13:19:01 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 10:54:53 2015 +0200
----------------------------------------------------------------------
components/camel-elsql/pom.xml | 106 ++++++++++
.../camel/component/elsql/ElsqlComponent.java | 147 ++++++++++++++
.../camel/component/elsql/ElsqlConsumer.java | 32 +++
.../camel/component/elsql/ElsqlEndpoint.java | 121 +++++++++++
.../camel/component/elsql/ElsqlProducer.java | 113 +++++++++++
.../camel/component/elsql/ElsqlSqlParams.java | 68 +++++++
.../elsql/ElsqlSqlPrepareStatementStrategy.java | 39 ++++
.../elsql/ElsqlSqlProcessingStrategy.java | 65 ++++++
.../src/main/resources/META-INF/LICENSE.txt | 203 +++++++++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
.../services/org/apache/camel/component/elsql | 18 ++
.../camel/component/sql/SqlComponent.java | 4 +
.../apache/camel/component/sql/SqlEndpoint.java | 4 +-
13 files changed, 929 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-elsql/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-elsql/pom.xml b/components/camel-elsql/pom.xml
new file mode 100644
index 0000000..d82bb05
--- /dev/null
+++ b/components/camel-elsql/pom.xml
@@ -0,0 +1,106 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>2.16.1-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-elsql</artifactId>
+ <packaging>bundle</packaging>
+ <name>Camel :: ElSql</name>
+ <description>Camel ElSql support</description>
+
+ <properties>
+ <camel.osgi.export.pkg>
+ org.apache.camel.component.elsql.*;${camel.osgi.version}
+ </camel.osgi.export.pkg>
+ <camel.osgi.import.pkg>
+ !org.apache.camel.component.elsql.*,
+ ${camel.osgi.import.defaults},
+ *
+ </camel.osgi.import.pkg>
+ <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=elsql</camel.osgi.export.service>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+ <!-- this component extends camel-sql -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-sql</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.opengamma</groupId>
+ <artifactId>elsql</artifactId>
+ <version>1.1</version>
+ </dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-spring</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.derby</groupId>
+ <artifactId>derby</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.hsqldb</groupId>
+ <artifactId>hsqldb</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <!-- use per test fork mode to avoid side effects -->
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <forkCount>1</forkCount>
+ <reuseForks>false</reuseForks>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
new file mode 100644
index 0000000..ddea39d
--- /dev/null
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
@@ -0,0 +1,147 @@
+/**
+ * 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.elsql;
+
+import java.util.Map;
+import javax.sql.DataSource;
+
+import com.opengamma.elsql.ElSqlConfig;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.sql.SqlComponent;
+import org.apache.camel.util.CamelContextHelper;
+import org.apache.camel.util.IntrospectionSupport;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+public class ElsqlComponent extends SqlComponent {
+
+ private ElSqlConfig elSqlConfig;
+ private String resourceUri;
+
+ public ElsqlComponent(CamelContext context) {
+ super(context, ElsqlEndpoint.class);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ DataSource target = null;
+
+ // endpoint options overrule component configured datasource
+ DataSource ds = resolveAndRemoveReferenceParameter(parameters, "dataSource", DataSource.class);
+ if (ds != null) {
+ target = ds;
+ }
+ String dataSourceRef = getAndRemoveParameter(parameters, "dataSourceRef", String.class);
+ if (target == null && dataSourceRef != null) {
+ target = CamelContextHelper.mandatoryLookup(getCamelContext(), dataSourceRef, DataSource.class);
+ }
+ if (target == null) {
+ // fallback and use component
+ target = getDataSource();
+ }
+ if (target == null) {
+ throw new IllegalArgumentException("DataSource must be configured");
+ }
+
+ JdbcTemplate jdbcTemplate = new JdbcTemplate(target);
+ IntrospectionSupport.setProperties(jdbcTemplate, parameters, "template.");
+
+ String elsqlName = remaining;
+ String resUri = resourceUri;
+ String[] part = remaining.split("/");
+ if (part.length == 2) {
+ elsqlName = part[0];
+ resUri = part[1];
+ } else if (part.length > 2) {
+ throw new IllegalArgumentException("Invalid uri. Must by elsql:elsqlName/resourceUri, was: " + uri);
+ }
+
+ /* TODO: add this later
+ String onConsume = getAndRemoveParameter(parameters, "consumer.onConsume", String.class);
+ if (onConsume == null) {
+ onConsume = getAndRemoveParameter(parameters, "onConsume", String.class);
+ }
+ if (onConsume != null && isUsePlaceholder()) {
+ onConsume = onConsume.replaceAll(parameterPlaceholderSubstitute, "?");
+ }
+ String onConsumeFailed = getAndRemoveParameter(parameters, "consumer.onConsumeFailed", String.class);
+ if (onConsumeFailed == null) {
+ onConsumeFailed = getAndRemoveParameter(parameters, "onConsumeFailed", String.class);
+ }
+ if (onConsumeFailed != null && isUsePlaceholder()) {
+ onConsumeFailed = onConsumeFailed.replaceAll(parameterPlaceholderSubstitute, "?");
+ }
+ String onConsumeBatchComplete = getAndRemoveParameter(parameters, "consumer.onConsumeBatchComplete", String.class);
+ if (onConsumeBatchComplete == null) {
+ onConsumeBatchComplete = getAndRemoveParameter(parameters, "onConsumeBatchComplete", String.class);
+ }
+ if (onConsumeBatchComplete != null && isUsePlaceholder()) {
+ onConsumeBatchComplete = onConsumeBatchComplete.replaceAll(parameterPlaceholderSubstitute, "?");
+ }
+ */
+
+ ElsqlEndpoint endpoint = new ElsqlEndpoint(uri, this, jdbcTemplate, elsqlName, resUri);
+// endpoint.setOnConsume(onConsume);
+// endpoint.setOnConsumeFailed(onConsumeFailed);
+// endpoint.setOnConsumeBatchComplete(onConsumeBatchComplete);
+ endpoint.setDataSource(ds);
+ endpoint.setDataSourceRef(dataSourceRef);
+ endpoint.setElSqlConfig(elSqlConfig);
+ return endpoint;
+ }
+
+ public ElSqlConfig getElSqlConfig() {
+ return elSqlConfig;
+ }
+
+ /**
+ * To use the given ElSqlConfig as configuration
+ */
+ public void setElSqlConfig(ElSqlConfig elSqlConfig) {
+ this.elSqlConfig = elSqlConfig;
+ }
+
+ public String getResourceUri() {
+ return resourceUri;
+ }
+
+ /**
+ * The eqlsql resource tile which contains the elsql SQL statements to use
+ */
+ public void setResourceUri(String resourceUri) {
+ this.resourceUri = resourceUri;
+ }
+
+ /**
+ * Sets the DataSource to use to communicate with the database.
+ */
+ @Override
+ public void setDataSource(DataSource dataSource) {
+ super.setDataSource(dataSource);
+ }
+
+ /**
+ * Sets whether to use placeholder and replace all placeholder characters with ? sign in the SQL queries.
+ * <p/>
+ * This option is default <tt>true</tt>
+ */
+ @Override
+ public void setUsePlaceholder(boolean usePlaceholder) {
+ super.setUsePlaceholder(usePlaceholder);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
new file mode 100644
index 0000000..0f1c6d7
--- /dev/null
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlConsumer.java
@@ -0,0 +1,32 @@
+/**
+ * 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.elsql;
+
+import org.apache.camel.Processor;
+import org.apache.camel.component.sql.SqlConsumer;
+import org.apache.camel.component.sql.SqlEndpoint;
+import org.apache.camel.component.sql.SqlPrepareStatementStrategy;
+import org.apache.camel.component.sql.SqlProcessingStrategy;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+public class ElsqlConsumer extends SqlConsumer {
+
+ public ElsqlConsumer(SqlEndpoint endpoint, Processor processor, JdbcTemplate jdbcTemplate, String query,
+ SqlPrepareStatementStrategy sqlPrepareStatementStrategy, SqlProcessingStrategy sqlProcessingStrategy) {
+ super(endpoint, processor, jdbcTemplate, query, sqlPrepareStatementStrategy, sqlProcessingStrategy);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
new file mode 100644
index 0000000..8ef9845
--- /dev/null
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
@@ -0,0 +1,121 @@
+/**
+ * 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.elsql;
+
+import java.net.URL;
+
+import com.opengamma.elsql.ElSql;
+import com.opengamma.elsql.ElSqlConfig;
+import org.apache.camel.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.sql.SqlEndpoint;
+import org.apache.camel.component.sql.SqlPrepareStatementStrategy;
+import org.apache.camel.component.sql.SqlProcessingStrategy;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriPath;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.ResourceHelper;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+@UriEndpoint(scheme = "elsql", title = "SQL", syntax = "elsql:elsqlName/resourceUri", consumerClass = ElsqlConsumer.class, label = "database,sql")
+public class ElsqlEndpoint extends SqlEndpoint {
+
+ private volatile ElSql elSql;
+
+ @UriPath
+ @Metadata(required = "true")
+ private String elsqlName;
+ @UriPath
+ private String resourceUri;
+ @UriParam
+ private ElSqlConfig elSqlConfig;
+
+ public ElsqlEndpoint(String uri, Component component, JdbcTemplate jdbcTemplate, String elsqlName, String resourceUri) {
+ super(uri, component, jdbcTemplate, null);
+ this.elsqlName = elsqlName;
+ this.resourceUri = resourceUri;
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ SqlProcessingStrategy proStrategy = new ElsqlSqlProcessingStrategy(elsqlName, elSql);
+ SqlPrepareStatementStrategy preStategy = new ElsqlSqlPrepareStatementStrategy();
+
+ ElsqlConsumer consumer = new ElsqlConsumer(this, processor, getJdbcTemplate(), elsqlName, preStategy, proStrategy);
+ consumer.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
+ consumer.setOnConsume(getOnConsume());
+ consumer.setOnConsumeFailed(getOnConsumeFailed());
+ consumer.setOnConsumeBatchComplete(getOnConsumeBatchComplete());
+ consumer.setBreakBatchOnConsumeFail(isBreakBatchOnConsumeFail());
+ consumer.setExpectedUpdateCount(getExpectedUpdateCount());
+ consumer.setUseIterator(isUseIterator());
+ consumer.setRouteEmptyResultSet(isRouteEmptyResultSet());
+ configureConsumer(consumer);
+ return consumer;
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ ElsqlProducer result = new ElsqlProducer(this, elSql, elsqlName, getJdbcTemplate());
+ return result;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ ObjectHelper.notNull(elSqlConfig, "elSqlConfig", this);
+ ObjectHelper.notNull(resourceUri, "resourceUri", this);
+
+ URL url = ResourceHelper.resolveMandatoryResourceAsUrl(getCamelContext().getClassResolver(), resourceUri);
+
+ elSql = ElSql.parse(elSqlConfig, url);
+ }
+
+ /**
+ * The name of the elsql to use (is @NAMED in the elsql file)
+ */
+ public String getElsqlName() {
+ return elsqlName;
+ }
+
+ public ElSqlConfig getElSqlConfig() {
+ return elSqlConfig;
+ }
+
+ /**
+ * The elsql configuration to use
+ */
+ public void setElSqlConfig(ElSqlConfig elSqlConfig) {
+ this.elSqlConfig = elSqlConfig;
+ }
+
+ public String getResourceUri() {
+ return resourceUri;
+ }
+
+ /**
+ * The eqlsql resource tile which contains the elsql SQL statements to use
+ */
+ public void setResourceUri(String resourceUri) {
+ this.resourceUri = resourceUri;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
new file mode 100644
index 0000000..73925d3
--- /dev/null
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlProducer.java
@@ -0,0 +1,113 @@
+/**
+ * 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.elsql;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+
+import com.opengamma.elsql.ElSql;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.sql.SqlConstants;
+import org.apache.camel.component.sql.SqlEndpoint;
+import org.apache.camel.component.sql.SqlOutputType;
+import org.apache.camel.impl.DefaultProducer;
+import org.springframework.dao.DataAccessException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.PreparedStatementCallback;
+
+import static org.springframework.jdbc.support.JdbcUtils.closeResultSet;
+
+public class ElsqlProducer extends DefaultProducer {
+
+ private final ElSql elSql;
+ private final String elSqlName;
+ private final JdbcTemplate jdbcTemplate;
+
+ public ElsqlProducer(SqlEndpoint endpoint, ElSql elSql, String elSqlName, JdbcTemplate jdbcTemplate) {
+ super(endpoint);
+ this.elSql = elSql;
+ this.elSqlName = elSqlName;
+ this.jdbcTemplate = jdbcTemplate;
+ }
+
+ @Override
+ public ElsqlEndpoint getEndpoint() {
+ return (ElsqlEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ public void process(final Exchange exchange) throws Exception {
+ Object data = exchange.getIn().getBody();
+
+ final String sql = elSql.getSql(elSqlName, new ElsqlSqlParams(exchange, data));
+
+ jdbcTemplate.execute(sql, new PreparedStatementCallback<Object>() {
+ @Override
+ public Object doInPreparedStatement(PreparedStatement ps) throws SQLException, DataAccessException {
+ ResultSet rs = null;
+ try {
+ boolean isResultSet = ps.execute();
+ if (isResultSet) {
+ // preserve headers first, so we can override the SQL_ROW_COUNT header
+ exchange.getOut().getHeaders().putAll(exchange.getIn().getHeaders());
+
+ rs = ps.getResultSet();
+ SqlOutputType outputType = getEndpoint().getOutputType();
+ log.trace("Got result list from query: {}, outputType={}", rs, outputType);
+ if (outputType == SqlOutputType.SelectList) {
+ List<?> data = getEndpoint().queryForList(rs, true);
+ // for noop=true we still want to enrich with the row count header
+ if (getEndpoint().isNoop()) {
+ exchange.getOut().setBody(exchange.getIn().getBody());
+ } else if (getEndpoint().getOutputHeader() != null) {
+ exchange.getOut().setBody(exchange.getIn().getBody());
+ exchange.getOut().setHeader(getEndpoint().getOutputHeader(), data);
+ } else {
+ exchange.getOut().setBody(data);
+ }
+ exchange.getOut().setHeader(SqlConstants.SQL_ROW_COUNT, data.size());
+ } else if (outputType == SqlOutputType.SelectOne) {
+ Object data = getEndpoint().queryForObject(rs);
+ if (data != null) {
+ // for noop=true we still want to enrich with the row count header
+ if (getEndpoint().isNoop()) {
+ exchange.getOut().setBody(exchange.getIn().getBody());
+ } else if (getEndpoint().getOutputHeader() != null) {
+ exchange.getOut().setBody(exchange.getIn().getBody());
+ exchange.getOut().setHeader(getEndpoint().getOutputHeader(), data);
+ } else {
+ exchange.getOut().setBody(data);
+ }
+ exchange.getOut().setHeader(SqlConstants.SQL_ROW_COUNT, 1);
+ }
+ } else {
+ throw new IllegalArgumentException("Invalid outputType=" + outputType);
+ }
+ } else {
+ exchange.getIn().setHeader(SqlConstants.SQL_UPDATE_COUNT, ps.getUpdateCount());
+ }
+ } finally {
+ closeResultSet(rs);
+ }
+
+ return null;
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlParams.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlParams.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlParams.java
new file mode 100644
index 0000000..6822e8b
--- /dev/null
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlParams.java
@@ -0,0 +1,68 @@
+/**
+ * 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.elsql;
+
+import java.util.Collections;
+import java.util.Map;
+
+import com.opengamma.elsql.SqlParams;
+import org.apache.camel.Exchange;
+import org.apache.camel.language.simple.SimpleLanguage;
+
+public class ElsqlSqlParams implements SqlParams {
+
+ private final Exchange exchange;
+ private final Map<?, ?> bodyMap;
+ private final Map<?, ?> headersMap;
+
+ public ElsqlSqlParams(Exchange exchange, Object body) {
+ this.exchange = exchange;
+ this.bodyMap = safeMap(exchange.getContext().getTypeConverter().tryConvertTo(Map.class, body));
+ this.headersMap = safeMap(exchange.getIn().getHeaders());
+ }
+
+ private static Map<?, ?> safeMap(Map<?, ?> map) {
+ return (map == null || map.isEmpty()) ? Collections.emptyMap() : map;
+ }
+
+ @Override
+ public boolean contains(String variable) {
+ if (variable.startsWith("${") && variable.endsWith("}")) {
+ return SimpleLanguage.expression(variable).evaluate(exchange, Object.class) != null;
+ } else if (bodyMap.containsKey(variable)) {
+ return true;
+ } else if (headersMap.containsKey(variable)) {
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ public Object get(String variable) {
+ if (variable.startsWith("${") && variable.endsWith("}")) {
+ return SimpleLanguage.expression(variable).evaluate(exchange, Object.class) != null;
+ } else if (bodyMap.containsKey(variable)) {
+ return bodyMap.get(variable);
+ } else if (headersMap.containsKey(variable)) {
+ return headersMap.get(variable);
+ }
+
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlPrepareStatementStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlPrepareStatementStrategy.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlPrepareStatementStrategy.java
new file mode 100644
index 0000000..d2ef797
--- /dev/null
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlPrepareStatementStrategy.java
@@ -0,0 +1,39 @@
+/**
+ * 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.elsql;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.Iterator;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.sql.SqlPrepareStatementStrategy;
+
+public class ElsqlSqlPrepareStatementStrategy implements SqlPrepareStatementStrategy {
+
+ public String prepareQuery(String query, boolean allowNamedParameters) throws SQLException {
+ return query;
+ }
+
+ public Iterator<?> createPopulateIterator(String query, String preparedQuery, int expectedParams, Exchange exchange, Object value) throws SQLException {
+ return null;
+ }
+
+ public void populateStatement(PreparedStatement ps, Iterator<?> iterator, int expectedParams) throws SQLException {
+ // noop
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
new file mode 100644
index 0000000..d22c33e
--- /dev/null
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlSqlProcessingStrategy.java
@@ -0,0 +1,65 @@
+/**
+ * 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.elsql;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+import com.opengamma.elsql.ElSql;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.sql.SqlEndpoint;
+import org.apache.camel.component.sql.SqlProcessingStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.dao.DataAccessException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.PreparedStatementCallback;
+
+public class ElsqlSqlProcessingStrategy implements SqlProcessingStrategy {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ElsqlSqlProcessingStrategy.class);
+ private final String elSqlName;
+ private final ElSql elSql;
+
+ public ElsqlSqlProcessingStrategy(String elSqlName, ElSql elSql) {
+ this.elSqlName = elSqlName;
+ this.elSql = elSql;
+ }
+
+ @Override
+ public int commit(final SqlEndpoint endpoint, final Exchange exchange, final Object data, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
+ String sql = elSql.getSql(elSqlName, new ElsqlSqlParams(exchange, data));
+
+ return jdbcTemplate.execute(sql, new PreparedStatementCallback<Integer>() {
+ @Override
+ public Integer doInPreparedStatement(PreparedStatement ps) throws SQLException, DataAccessException {
+ ps.execute();
+
+ int updateCount = ps.getUpdateCount();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Update count {}", updateCount);
+ }
+ return updateCount;
+ }
+ });
+ }
+
+ @Override
+ public int commitBatchComplete(final SqlEndpoint endpoint, final JdbcTemplate jdbcTemplate, final String query) throws Exception {
+ return 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-elsql/src/main/resources/META-INF/LICENSE.txt
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/resources/META-INF/LICENSE.txt b/components/camel-elsql/src/main/resources/META-INF/LICENSE.txt
new file mode 100755
index 0000000..6b0b127
--- /dev/null
+++ b/components/camel-elsql/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/6cf99562/components/camel-elsql/src/main/resources/META-INF/NOTICE.txt
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/resources/META-INF/NOTICE.txt b/components/camel-elsql/src/main/resources/META-INF/NOTICE.txt
new file mode 100755
index 0000000..2e215bf
--- /dev/null
+++ b/components/camel-elsql/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/6cf99562/components/camel-elsql/src/main/resources/META-INF/services/org/apache/camel/component/elsql
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/resources/META-INF/services/org/apache/camel/component/elsql b/components/camel-elsql/src/main/resources/META-INF/services/org/apache/camel/component/elsql
new file mode 100644
index 0000000..d27a646
--- /dev/null
+++ b/components/camel-elsql/src/main/resources/META-INF/services/org/apache/camel/component/elsql
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+class=org.apache.camel.component.elsql.ElsqlComponent
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java
index 5cdbed8..1d9ada5 100755
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlComponent.java
@@ -45,6 +45,10 @@ public class SqlComponent extends UriEndpointComponent {
super(context, SqlEndpoint.class);
}
+ public SqlComponent(CamelContext context, Class<? extends Endpoint> endpointClass) {
+ super(context, endpointClass);
+ }
+
@Override
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
DataSource target = null;
http://git-wip-us.apache.org/repos/asf/camel/blob/6cf99562/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 b477531..af0d4ad 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
@@ -421,7 +421,7 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
}
@SuppressWarnings("unchecked")
- protected List<?> queryForList(ResultSet rs, boolean allowMapToClass) throws SQLException {
+ public List<?> queryForList(ResultSet rs, boolean allowMapToClass) throws SQLException {
if (allowMapToClass && outputClass != null) {
Class<?> outputClazz = getCamelContext().getClassResolver().resolveClass(outputClass);
RowMapper rowMapper = new BeanPropertyRowMapper(outputClazz);
@@ -437,7 +437,7 @@ public class SqlEndpoint extends DefaultPollingEndpoint {
}
@SuppressWarnings("unchecked")
- protected Object queryForObject(ResultSet rs) throws SQLException {
+ public Object queryForObject(ResultSet rs) throws SQLException {
Object result = null;
if (outputClass == null) {
RowMapper rowMapper = new ColumnMapRowMapper();
[06/10] camel git commit: CAMEL-9162: camel-elsql component
Posted by da...@apache.org.
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f6cefc74
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f6cefc74
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f6cefc74
Branch: refs/heads/master
Commit: f6cefc74e6455593a9f64f79c9218de0a4f44e1d
Parents: ed0f42a
Author: Claus Ibsen <da...@apache.org>
Authored: Mon Oct 5 07:40:21 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 10:54:54 2015 +0200
----------------------------------------------------------------------
.../camel/component/elsql/ElsqlEndpoint.java | 19 +++--
.../component/elsql/ElSqlConsumerTest.java | 88 ++++++++++++++++++++
.../src/test/resources/elsql/projects.elsql | 4 +
3 files changed, 102 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/f6cefc74/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
index 9025d30..352dc37 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
@@ -25,19 +25,22 @@ import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.component.sql.SqlEndpoint;
+import org.apache.camel.component.sql.SqlPrepareStatementStrategy;
+import org.apache.camel.component.sql.SqlProcessingStrategy;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ResourceHelper;
+import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
@UriEndpoint(scheme = "elsql", title = "SQL", syntax = "elsql:elsqlName:resourceUri", consumerClass = ElsqlConsumer.class, label = "database,sql")
public class ElsqlEndpoint extends SqlEndpoint {
private volatile ElSql elSql;
- private NamedParameterJdbcTemplate jdbcTemplate;
+ private NamedParameterJdbcTemplate namedJdbcTemplate;
@UriPath
@Metadata(required = "true")
@@ -47,21 +50,21 @@ public class ElsqlEndpoint extends SqlEndpoint {
@UriParam
private ElSqlConfig elSqlConfig;
- public ElsqlEndpoint(String uri, Component component, NamedParameterJdbcTemplate jdbcTemplate, String elsqlName, String resourceUri) {
+ public ElsqlEndpoint(String uri, Component component, NamedParameterJdbcTemplate namedJdbcTemplate, String elsqlName, String resourceUri) {
super(uri, component, null, null);
this.elsqlName = elsqlName;
this.resourceUri = resourceUri;
- this.jdbcTemplate = jdbcTemplate;
+ this.namedJdbcTemplate = namedJdbcTemplate;
}
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- // TODO: must be named
- /*
SqlProcessingStrategy proStrategy = new ElsqlSqlProcessingStrategy(elsqlName, elSql);
SqlPrepareStatementStrategy preStategy = new ElsqlSqlPrepareStatementStrategy();
- ElsqlConsumer consumer = new ElsqlConsumer(this, processor, jdbcTemplate, elsqlName, preStategy, proStrategy);
+ JdbcTemplate template = new JdbcTemplate(getDataSource());
+
+ ElsqlConsumer consumer = new ElsqlConsumer(this, processor, template, elsqlName, preStategy, proStrategy);
consumer.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
consumer.setOnConsume(getOnConsume());
consumer.setOnConsumeFailed(getOnConsumeFailed());
@@ -72,13 +75,11 @@ public class ElsqlEndpoint extends SqlEndpoint {
consumer.setRouteEmptyResultSet(isRouteEmptyResultSet());
configureConsumer(consumer);
return consumer;
- */
- return null;
}
@Override
public Producer createProducer() throws Exception {
- ElsqlProducer result = new ElsqlProducer(this, elSql, elsqlName, jdbcTemplate);
+ ElsqlProducer result = new ElsqlProducer(this, elSql, elsqlName, namedJdbcTemplate);
return result;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/f6cefc74/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlConsumerTest.java b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlConsumerTest.java
new file mode 100644
index 0000000..7d34cb3
--- /dev/null
+++ b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlConsumerTest.java
@@ -0,0 +1,88 @@
+/**
+ * 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.elsql;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.After;
+import org.junit.Test;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+
+/**
+ *
+ */
+public class ElSqlConsumerTest extends CamelTestSupport {
+
+ private EmbeddedDatabase db;
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry jndi = super.createRegistry();
+
+ // this is the database we create with some initial data for our unit test
+ db = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.DERBY).addScript("sql/createAndPopulateDatabase.sql").build();
+
+ jndi.bind("dataSource", db);
+
+ return jndi;
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ super.tearDown();
+
+ db.shutdown();
+ }
+
+ @Test
+ public void testConsume() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMinimumMessageCount(3);
+
+ assertMockEndpointsSatisfied();
+
+ List<Exchange> exchanges = mock.getReceivedExchanges();
+ assertTrue(exchanges.size() >= 3);
+
+ assertEquals(1, exchanges.get(0).getIn().getBody(Map.class).get("ID"));
+ assertEquals("Camel", exchanges.get(0).getIn().getBody(Map.class).get("PROJECT"));
+ assertEquals(2, exchanges.get(1).getIn().getBody(Map.class).get("ID"));
+ assertEquals("AMQ", exchanges.get(1).getIn().getBody(Map.class).get("PROJECT"));
+ assertEquals(3, exchanges.get(2).getIn().getBody(Map.class).get("ID"));
+ assertEquals("Linux", exchanges.get(2).getIn().getBody(Map.class).get("PROJECT"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("elsql:allProjects:elsql/projects.elsql?dataSource=dataSource")
+ .to("mock:result");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/f6cefc74/components/camel-elsql/src/test/resources/elsql/projects.elsql
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/resources/elsql/projects.elsql b/components/camel-elsql/src/test/resources/elsql/projects.elsql
index 75dfe51..de60eef 100644
--- a/components/camel-elsql/src/test/resources/elsql/projects.elsql
+++ b/components/camel-elsql/src/test/resources/elsql/projects.elsql
@@ -3,3 +3,7 @@
FROM projects
WHERE license = :body
ORDER BY id
+@NAME(allProjects)
+ SELECT *
+ FROM projects
+ ORDER BY id
[09/10] camel git commit: CAMEL-9162: camel-elsql component
Posted by da...@apache.org.
CAMEL-9162: camel-elsql component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/913efcc1
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/913efcc1
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/913efcc1
Branch: refs/heads/master
Commit: 913efcc1ae852b86000b7709345be28cf6f12e56
Parents: 7241cad
Author: Claus Ibsen <da...@apache.org>
Authored: Mon Oct 5 11:50:34 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Oct 5 11:50:34 2015 +0200
----------------------------------------------------------------------
.../component/elsql/ElSqlDatabaseVendor.java | 42 ++++++++++
.../camel/component/elsql/ElsqlComponent.java | 31 ++++---
.../camel/component/elsql/ElsqlEndpoint.java | 26 +++++-
.../component/elsql/ElSqlComponentTest.java | 88 ++++++++++++++++++++
.../component/elsql/ElSqlDataSourceTest.java | 2 +-
.../src/test/resources/elsql/projects.elsql | 2 +-
6 files changed, 169 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/913efcc1/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElSqlDatabaseVendor.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElSqlDatabaseVendor.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElSqlDatabaseVendor.java
new file mode 100644
index 0000000..28ce85a
--- /dev/null
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElSqlDatabaseVendor.java
@@ -0,0 +1,42 @@
+package org.apache.camel.component.elsql;
+
+import com.opengamma.elsql.ElSqlConfig;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+public enum ElSqlDatabaseVendor {
+
+ Default, Postgres, HSql, MySql, Oracle, SqlServer2008, Veritca;
+
+ ElSqlConfig asElSqlConfig() {
+ if (Postgres.equals(this)) {
+ return ElSqlConfig.POSTGRES;
+ } else if (HSql.equals(this)) {
+ return ElSqlConfig.HSQL;
+ } else if (MySql.equals(this)) {
+ return ElSqlConfig.MYSQL;
+ } else if (Oracle.equals(this)) {
+ return ElSqlConfig.ORACLE;
+ } else if (SqlServer2008.equals(this)) {
+ return ElSqlConfig.SQL_SERVER_2008;
+ } else if (Veritca.equals(this)) {
+ return ElSqlConfig.VERTICA;
+ } else {
+ return ElSqlConfig.DEFAULT;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/913efcc1/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
index 51142e8..1c68612 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlComponent.java
@@ -28,6 +28,7 @@ import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
public class ElsqlComponent extends UriEndpointComponent {
+ private ElSqlDatabaseVendor databaseVendor;
private DataSource dataSource;
private ElSqlConfig elSqlConfig;
private String resourceUri;
@@ -67,7 +68,7 @@ public class ElsqlComponent extends UriEndpointComponent {
elsqlName = part[0];
resUri = part[1];
} else if (part.length > 2) {
- throw new IllegalArgumentException("Invalid uri. Must by elsql:elsqlName/resourceUri, was: " + uri);
+ throw new IllegalArgumentException("Invalid uri. Must by elsql:elsqlName:resourceUri, was: " + uri);
}
String onConsume = getAndRemoveParameter(parameters, "consumer.onConsume", String.class);
@@ -84,27 +85,25 @@ public class ElsqlComponent extends UriEndpointComponent {
}
ElsqlEndpoint endpoint = new ElsqlEndpoint(uri, this, jdbcTemplate, elsqlName, resUri);
+ endpoint.setElSqlConfig(elSqlConfig);
+ endpoint.setDatabaseVendor(databaseVendor);
+ endpoint.setDataSource(ds);
+ endpoint.setDataSourceRef(dataSourceRef);
endpoint.setOnConsume(onConsume);
endpoint.setOnConsumeFailed(onConsumeFailed);
endpoint.setOnConsumeBatchComplete(onConsumeBatchComplete);
- endpoint.setDataSource(ds);
- endpoint.setDataSourceRef(dataSourceRef);
- endpoint.setElSqlConfig(elSqlConfig);
return endpoint;
}
- @Override
- protected void doStart() throws Exception {
- super.doStart();
-
- if (elSqlConfig == null) {
- elSqlConfig = ElSqlConfig.DEFAULT;
- }
+ public ElSqlDatabaseVendor getDatabaseVendor() {
+ return databaseVendor;
}
- @Override
- protected void doStop() throws Exception {
- super.doStop();
+ /**
+ * To use a vendor specific {@link com.opengamma.elsql.ElSqlConfig}
+ */
+ public void setDatabaseVendor(ElSqlDatabaseVendor databaseVendor) {
+ this.databaseVendor = databaseVendor;
}
/**
@@ -126,7 +125,7 @@ public class ElsqlComponent extends UriEndpointComponent {
}
/**
- * To use the given ElSqlConfig as configuration
+ * To use a specific configured ElSqlConfig. It may be better to use the <tt>databaseVendor</tt> option instead.
*/
public void setElSqlConfig(ElSqlConfig elSqlConfig) {
this.elSqlConfig = elSqlConfig;
@@ -137,7 +136,7 @@ public class ElsqlComponent extends UriEndpointComponent {
}
/**
- * The eqlsql resource tile which contains the elsql SQL statements to use
+ * The resource file which contains the elsql SQL statements to use
*/
public void setResourceUri(String resourceUri) {
this.resourceUri = resourceUri;
http://git-wip-us.apache.org/repos/asf/camel/blob/913efcc1/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
index d2b2cbf..62a00e7 100644
--- a/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
+++ b/components/camel-elsql/src/main/java/org/apache/camel/component/elsql/ElsqlEndpoint.java
@@ -45,7 +45,7 @@ public class ElsqlEndpoint extends DefaultSqlEndpoint {
private static final Logger LOG = LoggerFactory.getLogger(ElsqlEndpoint.class);
- private volatile ElSql elSql;
+ private ElSql elSql;
private NamedParameterJdbcTemplate namedJdbcTemplate;
@UriPath
@@ -54,6 +54,8 @@ public class ElsqlEndpoint extends DefaultSqlEndpoint {
@UriPath
private String resourceUri;
@UriParam
+ private ElSqlDatabaseVendor databaseVendor;
+ @UriParam
private ElSqlConfig elSqlConfig;
public ElsqlEndpoint(String uri, Component component, NamedParameterJdbcTemplate namedJdbcTemplate, String elsqlName, String resourceUri) {
@@ -95,9 +97,14 @@ public class ElsqlEndpoint extends DefaultSqlEndpoint {
protected void doStart() throws Exception {
super.doStart();
- ObjectHelper.notNull(elSqlConfig, "elSqlConfig", this);
ObjectHelper.notNull(resourceUri, "resourceUri", this);
+ if (elSqlConfig == null && databaseVendor != null) {
+ elSqlConfig = databaseVendor.asElSqlConfig();
+ } else if (elSqlConfig == null) {
+ elSqlConfig = ElSqlDatabaseVendor.Default.asElSqlConfig();
+ }
+
URL url = ResourceHelper.resolveMandatoryResourceAsUrl(getCamelContext().getClassResolver(), resourceUri);
elSql = ElSql.parse(elSqlConfig, url);
}
@@ -109,12 +116,23 @@ public class ElsqlEndpoint extends DefaultSqlEndpoint {
return elsqlName;
}
+ public ElSqlDatabaseVendor getDatabaseVendor() {
+ return databaseVendor;
+ }
+
+ /**
+ * To use a vendor specific {@link com.opengamma.elsql.ElSqlConfig}
+ */
+ public void setDatabaseVendor(ElSqlDatabaseVendor databaseVendor) {
+ this.databaseVendor = databaseVendor;
+ }
+
public ElSqlConfig getElSqlConfig() {
return elSqlConfig;
}
/**
- * The elsql configuration to use
+ * To use a specific configured ElSqlConfig. It may be better to use the <tt>databaseVendor</tt> option instead.
*/
public void setElSqlConfig(ElSqlConfig elSqlConfig) {
this.elSqlConfig = elSqlConfig;
@@ -125,7 +143,7 @@ public class ElsqlEndpoint extends DefaultSqlEndpoint {
}
/**
- * The eqlsql resource tile which contains the elsql SQL statements to use
+ * The resource file which contains the elsql SQL statements to use
*/
public void setResourceUri(String resourceUri) {
this.resourceUri = resourceUri;
http://git-wip-us.apache.org/repos/asf/camel/blob/913efcc1/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlComponentTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlComponentTest.java b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlComponentTest.java
new file mode 100644
index 0000000..838f5a2
--- /dev/null
+++ b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlComponentTest.java
@@ -0,0 +1,88 @@
+/**
+ * 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.elsql;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.After;
+import org.junit.Test;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+
+public class ElSqlComponentTest extends CamelTestSupport {
+
+ private EmbeddedDatabase db;
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry jndi = super.createRegistry();
+
+ // this is the database we create with some initial data for our unit test
+ db = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.DERBY).addScript("sql/createAndPopulateDatabase.sql").build();
+
+ jndi.bind("dataSource", db);
+
+ return jndi;
+ }
+
+ @Test
+ public void testSimpleBody() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ template.sendBody("direct:simple", "XXX");
+
+ mock.assertIsSatisfied();
+
+ // the result is a List
+ List<?> received = assertIsInstanceOf(List.class, mock.getReceivedExchanges().get(0).getIn().getBody());
+
+ // and each row in the list is a Map
+ Map<?, ?> row = assertIsInstanceOf(Map.class, received.get(0));
+
+ // and we should be able the get the project from the map that should be Linux
+ assertEquals("Linux", row.get("PROJECT"));
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ super.tearDown();
+
+ db.shutdown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+ getContext().getComponent("elsql", ElsqlComponent.class).setDataSource(db);
+ getContext().getComponent("elsql", ElsqlComponent.class).setResourceUri("elsql/projects.elsql");
+
+ from("direct:simple")
+ .to("elsql:projectsById")
+ .to("mock:result");
+ }
+ };
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/913efcc1/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlDataSourceTest.java
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlDataSourceTest.java b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlDataSourceTest.java
index 858daf4..e555808 100644
--- a/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlDataSourceTest.java
+++ b/components/camel-elsql/src/test/java/org/apache/camel/component/elsql/ElSqlDataSourceTest.java
@@ -77,7 +77,7 @@ public class ElSqlDataSourceTest extends CamelTestSupport {
return new RouteBuilder() {
public void configure() {
from("direct:simple")
- .to("elsql:projects:elsql/projects.elsql?dataSource=dataSource")
+ .to("elsql:projectsById:elsql/projects.elsql?dataSource=dataSource")
.to("mock:result");
}
};
http://git-wip-us.apache.org/repos/asf/camel/blob/913efcc1/components/camel-elsql/src/test/resources/elsql/projects.elsql
----------------------------------------------------------------------
diff --git a/components/camel-elsql/src/test/resources/elsql/projects.elsql b/components/camel-elsql/src/test/resources/elsql/projects.elsql
index 1d957ef..8cc1a8b 100644
--- a/components/camel-elsql/src/test/resources/elsql/projects.elsql
+++ b/components/camel-elsql/src/test/resources/elsql/projects.elsql
@@ -1,4 +1,4 @@
-@NAME(projects)
+@NAME(projectsById)
SELECT *
FROM projects
WHERE license = :body