You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by he...@apache.org on 2015/12/11 18:25:51 UTC
camel git commit: Added support for input conversions for
@RddCallback.
Repository: camel
Updated Branches:
refs/heads/master c4bdb4463 -> a68c79d68
Added support for input conversions for @RddCallback.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a68c79d6
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a68c79d6
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a68c79d6
Branch: refs/heads/master
Commit: a68c79d68a5f256006da0287ed32403b808ed984
Parents: c4bdb44
Author: Henryk Konsek <he...@gmail.com>
Authored: Fri Dec 11 18:15:55 2015 +0100
Committer: Henryk Konsek <he...@gmail.com>
Committed: Fri Dec 11 18:15:55 2015 +0100
----------------------------------------------------------------------
.../spark/annotations/AnnotatedRddCallback.java | 9 +++++++
.../annotations/AnnotatedRddCallbackProxy.java | 17 +++++++++++-
.../component/spark/SparkProducerTest.java | 28 ++++++++++++++------
3 files changed, 45 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a68c79d6/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallback.java
----------------------------------------------------------------------
diff --git a/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallback.java b/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallback.java
index dfd8e62..ee41697 100644
--- a/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallback.java
+++ b/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallback.java
@@ -16,8 +16,13 @@
*/
package org.apache.camel.component.spark.annotations;
+import org.apache.camel.CamelContext;
import org.apache.camel.component.spark.RddCallback;
+/**
+ * Provides facade for working with annotated RDD callbacks i.e. POJO classes with an appropriate annotations on
+ * selected methods.
+ */
public final class AnnotatedRddCallback {
private AnnotatedRddCallback() {
@@ -27,4 +32,8 @@ public final class AnnotatedRddCallback {
return new AnnotatedRddCallbackProxy(objectWithCallback);
}
+ public static RddCallback annotatedRddCallback(Object objectWithCallback, CamelContext camelContext) {
+ return new AnnotatedRddCallbackProxy(objectWithCallback, camelContext);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/a68c79d6/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallbackProxy.java
----------------------------------------------------------------------
diff --git a/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallbackProxy.java b/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallbackProxy.java
index ff5337f..5d2e9c9 100644
--- a/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallbackProxy.java
+++ b/components/camel-spark/src/main/java/org/apache/camel/component/spark/annotations/AnnotatedRddCallbackProxy.java
@@ -23,6 +23,7 @@ import java.util.List;
import static java.util.Arrays.asList;
+import org.apache.camel.CamelContext;
import org.apache.camel.component.spark.RddCallback;
import org.apache.spark.api.java.AbstractJavaRDDLike;
@@ -34,14 +35,21 @@ class AnnotatedRddCallbackProxy implements RddCallback {
private final List<Method> rddCallbacks;
- public AnnotatedRddCallbackProxy(Object objectWithCallback) {
+ private final CamelContext camelContext;
+
+ public AnnotatedRddCallbackProxy(Object objectWithCallback, CamelContext camelContext) {
this.objectWithCallback = objectWithCallback;
+ this.camelContext = camelContext;
this.rddCallbacks = findMethodsWithAnnotation(objectWithCallback.getClass(), org.apache.camel.component.spark.annotations.RddCallback.class);
if (rddCallbacks.size() == 0) {
throw new UnsupportedOperationException("Can't find methods annotated with @RddCallback.");
}
}
+ public AnnotatedRddCallbackProxy(Object objectWithCallback) {
+ this(objectWithCallback, null);
+ }
+
@Override
public Object onRdd(AbstractJavaRDDLike rdd, Object... payloads) {
try {
@@ -54,6 +62,13 @@ class AnnotatedRddCallbackProxy implements RddCallback {
Method callbackMethod = rddCallbacks.get(0);
callbackMethod.setAccessible(true);
+
+ if (camelContext != null) {
+ for (int i = 1; i < arguments.size(); i++) {
+ arguments.set(i, camelContext.getTypeConverter().convertTo(callbackMethod.getParameterTypes()[i], arguments.get(i)));
+ }
+ }
+
return callbackMethod.invoke(objectWithCallback, arguments.toArray(new Object[arguments.size()]));
} catch (IllegalAccessException | InvocationTargetException e) {
throw new RuntimeException(e);
http://git-wip-us.apache.org/repos/asf/camel/blob/a68c79d6/components/camel-spark/src/test/java/org/apache/camel/component/spark/SparkProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-spark/src/test/java/org/apache/camel/component/spark/SparkProducerTest.java b/components/camel-spark/src/test/java/org/apache/camel/component/spark/SparkProducerTest.java
index ae34cc8..44fc203 100644
--- a/components/camel-spark/src/test/java/org/apache/camel/component/spark/SparkProducerTest.java
+++ b/components/camel-spark/src/test/java/org/apache/camel/component/spark/SparkProducerTest.java
@@ -104,24 +104,24 @@ public class SparkProducerTest extends CamelTestSupport {
@Test
public void shouldExecuteRddCallbackWithSinglePayload() {
- long pomLinesCount = template.requestBodyAndHeader(sparkUri, 10, SPARK_RDD_CALLBACK_HEADER, new org.apache.camel.component.spark.RddCallback() {
+ long linesCount = template.requestBodyAndHeader(sparkUri, 10, SPARK_RDD_CALLBACK_HEADER, new org.apache.camel.component.spark.RddCallback() {
@Override
public Long onRdd(AbstractJavaRDDLike rdd, Object... payloads) {
return rdd.count() * (int) payloads[0];
}
}, Long.class);
- Truth.assertThat(pomLinesCount).isEqualTo(numberOfLinesInTestFile * 10);
+ Truth.assertThat(linesCount).isEqualTo(numberOfLinesInTestFile * 10);
}
@Test
public void shouldExecuteRddCallbackWithPayloads() {
- long pomLinesCount = template.requestBodyAndHeader(sparkUri, asList(10, 10), SPARK_RDD_CALLBACK_HEADER, new org.apache.camel.component.spark.RddCallback() {
+ long linesCount = template.requestBodyAndHeader(sparkUri, asList(10, 10), SPARK_RDD_CALLBACK_HEADER, new org.apache.camel.component.spark.RddCallback() {
@Override
public Long onRdd(AbstractJavaRDDLike rdd, Object... payloads) {
return rdd.count() * (int) payloads[0] * (int) payloads[1];
}
}, Long.class);
- Truth.assertThat(pomLinesCount).isEqualTo(numberOfLinesInTestFile * 10 * 10);
+ Truth.assertThat(linesCount).isEqualTo(numberOfLinesInTestFile * 10 * 10);
}
@Test
@@ -132,14 +132,14 @@ public class SparkProducerTest extends CamelTestSupport {
return rdd.count() * (int) payloads[0] * (int) payloads[1];
}
};
- long pomLinesCount = template.requestBodyAndHeader(sparkUri, asList("10", "10"), SPARK_RDD_CALLBACK_HEADER, rddCallback, Long.class);
- Truth.assertThat(pomLinesCount).isEqualTo(1900);
+ long linesCount = template.requestBodyAndHeader(sparkUri, asList("10", "10"), SPARK_RDD_CALLBACK_HEADER, rddCallback, Long.class);
+ Truth.assertThat(linesCount).isEqualTo(1900);
}
@Test
public void shouldUseTransformationFromRegistry() {
- long pomLinesCount = template.requestBody(sparkUri + "&rddCallback=#countLinesTransformation", null, Long.class);
- Truth.assertThat(pomLinesCount).isGreaterThan(0L);
+ long linesCount = template.requestBody(sparkUri + "&rddCallback=#countLinesTransformation", null, Long.class);
+ Truth.assertThat(linesCount).isGreaterThan(0L);
}
@Test
@@ -204,6 +204,18 @@ public class SparkProducerTest extends CamelTestSupport {
Truth.assertThat(pomLinesCount).isEqualTo(numberOfLinesInTestFile * 10 * 10);
}
+ @Test
+ public void shouldExecuteAnnotatedCallbackWithConversions() {
+ org.apache.camel.component.spark.RddCallback rddCallback = annotatedRddCallback(new Object() {
+ @RddCallback
+ long countLines(JavaRDD<String> textFile, int first, int second) {
+ return textFile.count() * first * second;
+ }
+ }, context);
+ long pomLinesCount = template.requestBodyAndHeader(sparkUri, asList(10, "10"), SPARK_RDD_CALLBACK_HEADER, rddCallback, Long.class);
+ Truth.assertThat(pomLinesCount).isEqualTo(numberOfLinesInTestFile * 10 * 10);
+ }
+
// Hive tests
@Test