You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2014/06/27 18:29:23 UTC
[6/8] git commit: CAMEL-7354: camel-spark component.
CAMEL-7354: camel-spark component.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/183bbef2
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/183bbef2
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/183bbef2
Branch: refs/heads/master
Commit: 183bbef21c8b4db6b1b0f83052df395c7adc8537
Parents: 3ff1072
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Jun 27 17:34:29 2014 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jun 27 17:34:29 2014 +0200
----------------------------------------------------------------------
components/camel-spark/pom.xml | 32 +++++++
.../camel/component/spark/CamelSparkRoute.java | 9 +-
.../component/spark/DefaultSparkBinding.java | 89 +++++++++++++++++++-
.../camel/component/spark/SparkBinding.java | 22 ++++-
.../component/spark/SparkConfiguration.java | 10 +++
.../camel/component/spark/SparkConstants.java | 26 ++++++
.../spark/SparkHeaderFilterStrategy.java | 4 +
7 files changed, 188 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/183bbef2/components/camel-spark/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-spark/pom.xml b/components/camel-spark/pom.xml
index d3f48f2..16f7c81 100644
--- a/components/camel-spark/pom.xml
+++ b/components/camel-spark/pom.xml
@@ -101,4 +101,36 @@
</dependency>
</dependencies>
+ <!-- unit testing requires java 8 -->
+ <profiles>
+ <profile>
+ <id>jdk8-test</id>
+ <activation>
+ <jdk>1.8</jdk>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>false</skipTests>
+ <forkedProcessTimeoutInSeconds>300</forkedProcessTimeoutInSeconds>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>true</skipTests>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/183bbef2/components/camel-spark/src/main/java/org/apache/camel/component/spark/CamelSparkRoute.java
----------------------------------------------------------------------
diff --git a/components/camel-spark/src/main/java/org/apache/camel/component/spark/CamelSparkRoute.java b/components/camel-spark/src/main/java/org/apache/camel/component/spark/CamelSparkRoute.java
index c418f88..76f1fa0 100644
--- a/components/camel-spark/src/main/java/org/apache/camel/component/spark/CamelSparkRoute.java
+++ b/components/camel-spark/src/main/java/org/apache/camel/component/spark/CamelSparkRoute.java
@@ -20,6 +20,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
import org.apache.camel.Processor;
+import org.apache.camel.util.ObjectHelper;
import spark.Request;
import spark.Response;
import spark.Route;
@@ -47,8 +48,12 @@ public class CamelSparkRoute implements Route {
exchange.setException(e);
}
- if (exchange.getException() != null) {
- // TODO: how to handle exchange failures
+ Message msg = exchange.hasOut() ? exchange.getOut() : exchange.getIn();
+
+ try {
+ endpoint.getSparkBinding().toSparkResponse(msg, response, endpoint.getSparkConfiguration());
+ } catch (Exception e) {
+ throw ObjectHelper.wrapRuntimeCamelException(e);
}
if (exchange.hasOut()) {
http://git-wip-us.apache.org/repos/asf/camel/blob/183bbef2/components/camel-spark/src/main/java/org/apache/camel/component/spark/DefaultSparkBinding.java
----------------------------------------------------------------------
diff --git a/components/camel-spark/src/main/java/org/apache/camel/component/spark/DefaultSparkBinding.java b/components/camel-spark/src/main/java/org/apache/camel/component/spark/DefaultSparkBinding.java
index f40b449..854f94b 100644
--- a/components/camel-spark/src/main/java/org/apache/camel/component/spark/DefaultSparkBinding.java
+++ b/components/camel-spark/src/main/java/org/apache/camel/component/spark/DefaultSparkBinding.java
@@ -16,16 +16,27 @@
*/
package org.apache.camel.component.spark;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectOutputStream;
+import java.io.PrintWriter;
+import java.io.StringWriter;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
+import java.util.Iterator;
import java.util.Map;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
+import org.apache.camel.TypeConverter;
import org.apache.camel.spi.HeaderFilterStrategy;
+import org.apache.camel.util.ExchangeHelper;
+import org.apache.camel.util.IOHelper;
+import org.apache.camel.util.MessageHelper;
+import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spark.Request;
+import spark.Response;
public class DefaultSparkBinding implements SparkBinding {
@@ -90,12 +101,88 @@ public class DefaultSparkBinding implements SparkBinding {
}
String[] splat = request.splat();
- String key = "splat";
+ String key = SparkConstants.SPLAT;
if (headerFilterStrategy != null
&& !headerFilterStrategy.applyFilterToExternalHeaders(key, splat, exchange)) {
SparkHelper.appendHeader(headers, key, splat);
}
+ }
+
+ @Override
+ public void toSparkResponse(Message message, Response response, SparkConfiguration configuration) throws Exception {
+ LOG.trace("toSparkResponse: {}", message);
+
+ // the response code is 200 for OK and 500 for failed
+ boolean failed = message.getExchange().isFailed();
+ int defaultCode = failed ? 500 : 200;
+
+ int code = message.getHeader(Exchange.HTTP_RESPONSE_CODE, defaultCode, int.class);
+ response.status(code);
+ LOG.trace("HTTP Status Code: {}", code);
+
+ TypeConverter tc = message.getExchange().getContext().getTypeConverter();
+
+ // append headers
+ // must use entrySet to ensure case of keys is preserved
+ for (Map.Entry<String, Object> entry : message.getHeaders().entrySet()) {
+ String key = entry.getKey();
+ Object value = entry.getValue();
+ // use an iterator as there can be multiple values. (must not use a delimiter)
+ final Iterator<?> it = ObjectHelper.createIterator(value, null);
+ while (it.hasNext()) {
+ String headerValue = tc.convertTo(String.class, it.next());
+ if (headerValue != null && headerFilterStrategy != null
+ && !headerFilterStrategy.applyFilterToCamelHeaders(key, headerValue, message.getExchange())) {
+ LOG.trace("HTTP-Header: {}={}", key, headerValue);
+ response.header(key, headerValue);
+ }
+ }
+ }
+
+ // set the content type in the response.
+ String contentType = MessageHelper.getContentType(message);
+ if (contentType != null) {
+ // set content-type
+ response.header(Exchange.CONTENT_TYPE, contentType);
+ LOG.trace("Content-Type: {}", contentType);
+ }
+
+ Object body = message.getBody();
+ Exception cause = message.getExchange().getException();
+
+ // if there was an exception then use that as body
+ if (cause != null) {
+ if (configuration.isTransferException()) {
+ // we failed due an exception, and transfer it as java serialized object
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(bos);
+ oos.writeObject(cause);
+ oos.flush();
+ IOHelper.close(oos, bos);
+
+ body = bos.toByteArray();
+ // force content type to be serialized java object
+ message.setHeader(Exchange.CONTENT_TYPE, SparkConstants.CONTENT_TYPE_JAVA_SERIALIZED_OBJECT);
+ } else {
+ // we failed due an exception so print it as plain text
+ StringWriter sw = new StringWriter();
+ PrintWriter pw = new PrintWriter(sw);
+ cause.printStackTrace(pw);
+
+ // the body should then be the stacktrace
+ body = sw.toString().getBytes();
+ // force content type to be text/plain as that is what the stacktrace is
+ message.setHeader(Exchange.CONTENT_TYPE, "text/plain");
+ }
+ // and mark the exception as failure handled, as we handled it by returning it as the response
+ ExchangeHelper.setFailureHandled(message.getExchange());
+ }
+
+ if (body != null) {
+ String str = tc.mandatoryConvertTo(String.class, message.getExchange(), body);
+ response.body(str);
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/camel/blob/183bbef2/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkBinding.java
----------------------------------------------------------------------
diff --git a/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkBinding.java b/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkBinding.java
index 61d8798..21053ae 100644
--- a/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkBinding.java
+++ b/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkBinding.java
@@ -21,13 +21,14 @@ import java.util.Map;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import spark.Request;
+import spark.Response;
public interface SparkBinding {
/**
* Binds from Spark {@link Request} to Camel {@link org.apache.camel.Message}.
*
- * @param request the netty http request
+ * @param request the Spark request
* @param exchange the exchange that should contain the returned message.
* @param configuration configuration
* @return the message to store on the given exchange
@@ -35,6 +36,25 @@ public interface SparkBinding {
*/
Message toCamelMessage(Request request, Exchange exchange, SparkConfiguration configuration) throws Exception;
+ /**
+ * Binds from Spark {@link Request} to Camel headers as a {@link Map}.
+ *
+ * @param request the Spark request
+ * @param headers the Camel headers that should be populated
+ * @param exchange the exchange that should contain the returned message.
+ * @param configuration the endpoint configuration
+ * @throws Exception is thrown if error during binding
+ */
void populateCamelHeaders(Request request, Map<String, Object> headers, Exchange exchange, SparkConfiguration configuration) throws Exception;
+ /**
+ * Binds from Camel {@link Message} to Spark {@link Response}.
+ *
+ * @param message the Camel message
+ * @param response the Spark response to bind to
+ * @param configuration the endpoint configuration
+ * @throws Exception is thrown if error during binding
+ */
+ void toSparkResponse(Message message, Response response, SparkConfiguration configuration) throws Exception;
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/183bbef2/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkConfiguration.java b/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkConfiguration.java
index 0c6d2b7..a24245a 100644
--- a/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkConfiguration.java
+++ b/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkConfiguration.java
@@ -28,6 +28,8 @@ public class SparkConfiguration {
private boolean disableStreamCache;
@UriParam
private boolean urlDecodeHeaders;
+ @UriParam
+ private boolean transferException;
public boolean isMapHeaders() {
return mapHeaders;
@@ -52,4 +54,12 @@ public class SparkConfiguration {
public void setUrlDecodeHeaders(boolean urlDecodeHeaders) {
this.urlDecodeHeaders = urlDecodeHeaders;
}
+
+ public boolean isTransferException() {
+ return transferException;
+ }
+
+ public void setTransferException(boolean transferException) {
+ this.transferException = transferException;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/183bbef2/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkConstants.java b/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkConstants.java
new file mode 100644
index 0000000..8a026fa
--- /dev/null
+++ b/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkConstants.java
@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.spark;
+
+public final class SparkConstants {
+
+ public static final String CONTENT_TYPE_JAVA_SERIALIZED_OBJECT = "application/x-java-serialized-object";
+ public static final String SPLAT = "splat";
+
+ private SparkConstants() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/183bbef2/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkHeaderFilterStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkHeaderFilterStrategy.java b/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkHeaderFilterStrategy.java
index 33e0eb0..f729bde 100644
--- a/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkHeaderFilterStrategy.java
+++ b/components/camel-spark/src/main/java/org/apache/camel/component/spark/SparkHeaderFilterStrategy.java
@@ -31,6 +31,7 @@ public class SparkHeaderFilterStrategy extends DefaultHeaderFilterStrategy {
getOutFilter().add("content-length");
getOutFilter().add("content-type");
getOutFilter().add("host");
+ getOutFilter().add("user-agent");
// Add the filter for the Generic Message header
// http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.5
getOutFilter().add("cache-control");
@@ -48,6 +49,9 @@ public class SparkHeaderFilterStrategy extends DefaultHeaderFilterStrategy {
// filter headers begin with "Camel" or "org.apache.camel"
// must ignore case for Http based transports
setOutFilterPattern("(?i)(Camel|org\\.apache\\.camel)[\\.|a-z|A-z|0-9]*");
+
+ // filter out splat as its an internal header
+ getOutFilter().add(SparkConstants.SPLAT);
}
}