You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by nf...@apache.org on 2019/10/28 08:48:01 UTC
[camel-k-runtime] branch master updated: CamelCloudEventXXX not
converted to CE header #177
This is an automated email from the ASF dual-hosted git repository.
nferraro pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel-k-runtime.git
The following commit(s) were added to refs/heads/master by this push:
new 177022c CamelCloudEventXXX not converted to CE header #177
new f4c290e Merge pull request #178 from lburgazzoli/github-177
177022c is described below
commit 177022c3f2935f80d3d2ab59bdcf50407dc7faef
Author: lburgazzoli <lb...@gmail.com>
AuthorDate: Sun Oct 27 17:55:01 2019 +0100
CamelCloudEventXXX not converted to CE header #177
---
.../camel/component/knative/spi/CloudEvents.java | 2 +-
.../component/knative/http/KnativeHttpTest.java | 120 ++++++++++++++++++++-
.../knative/ce/AbstractCloudEventProcessor.java | 41 ++++---
3 files changed, 143 insertions(+), 20 deletions(-)
diff --git a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/CloudEvents.java b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/CloudEvents.java
index dd8ab96..6118ef0 100644
--- a/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/CloudEvents.java
+++ b/camel-knative/camel-knative-api/src/main/java/org/apache/camel/component/knative/spi/CloudEvents.java
@@ -34,7 +34,7 @@ public enum CloudEvents implements CloudEvent {
Attribute.simple(CloudEvent.CAMEL_CLOUD_EVENT_ID, "CE-EventID", "eventID"),
Attribute.simple(CloudEvent.CAMEL_CLOUD_EVENT_TIME, "CE-EventTime", "eventTime"),
Attribute.simple(CloudEvent.CAMEL_CLOUD_EVENT_SCHEMA_URL, "CE-SchemaURL", "schemaURL"),
- Attribute.simple(CloudEvent.CAMEL_CLOUD_EVENT_CONTENT_TYPE, "ContentType", "contentType"),
+ Attribute.simple(CloudEvent.CAMEL_CLOUD_EVENT_CONTENT_TYPE, "Content-Type", "contentType"),
Attribute.simple(CloudEvent.CAMEL_CLOUD_EVENT_EXTENSIONS, "CE-Extensions", "extensions")
)
)),
diff --git a/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java b/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
index 9628838..390efd4 100644
--- a/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
+++ b/camel-knative/camel-knative-http/src/test/java/org/apache/camel/component/knative/http/KnativeHttpTest.java
@@ -262,7 +262,6 @@ public class KnativeHttpTest {
public void configure() throws Exception {
from("knative:endpoint/myEndpoint")
.to("mock:ce");
-
from("direct:source")
.toF("undertow:http://localhost:%d/a/path", port);
}
@@ -272,11 +271,16 @@ public class KnativeHttpTest {
MockEndpoint mock = context.getEndpoint("mock:ce", MockEndpoint.class);
mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_VERSION).id(), ce.version());
+ mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_VERSION).http(), ce.version());
mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).id(), "org.apache.camel.event");
+ mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).http(), "org.apache.camel.event");
mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_ID).id(), "myEventID");
+ mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_ID).http(), "myEventID");
mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE).id(), "/somewhere");
+ mock.expectedHeaderReceived(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE).http(), "/somewhere");
mock.expectedHeaderReceived(Exchange.CONTENT_TYPE, "text/plain");
mock.expectedMessagesMatches(e -> e.getMessage().getHeaders().containsKey(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TIME).id()));
+ mock.expectedMessagesMatches(e -> e.getMessage().getHeaders().containsKey(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TIME).http()));
mock.expectedBodiesReceived("test");
mock.expectedMessageCount(1);
@@ -918,8 +922,7 @@ public class KnativeHttpTest {
public void configure() throws Exception {
from("direct:source")
.to("knative:event/myEvent?kind=MyObject&apiVersion=v1");
-
- fromF("knative:event/myEvent?kind=MyOtherObject&apiVersion=v2")
+ from("knative:event/myEvent?kind=MyOtherObject&apiVersion=v2")
.to("mock:ce");
}
});
@@ -1370,8 +1373,6 @@ public class KnativeHttpTest {
}
}
-
-
@ParameterizedTest
@MethodSource("provideCloudEventsImplementations")
void testHeadersOverrideFromConf(CloudEvent ce) throws Exception {
@@ -1434,5 +1435,114 @@ public class KnativeHttpTest {
server.stop();
}
}
+
+ @ParameterizedTest
+ @MethodSource("provideCloudEventsImplementations")
+ void testHeadersOverrideFromRouteWithCamelHeader(CloudEvent ce) throws Exception {
+ configureKnativeComponent(
+ context,
+ ce,
+ endpoint(
+ Knative.EndpointKind.sink,
+ "ep",
+ "localhost",
+ port,
+ KnativeSupport.mapOf(
+ Knative.KNATIVE_EVENT_TYPE, "org.apache.camel.event",
+ Knative.CONTENT_TYPE, "text/plain"
+ )
+ )
+ );
+
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicReference<HttpServerExchange> exchange = new AtomicReference<>();
+
+ Undertow server = Undertow.builder()
+ .addHttpListener(port, "localhost")
+ .setHandler(se -> {
+ exchange.set(se);
+ latch.countDown();
+ })
+ .build();
+
+ RouteBuilder.addRoutes(context, b -> {
+ b.from("direct:start")
+ .setHeader(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).constant("myType")
+ .to("knative:endpoint/ep");
+ });
+
+ context.start();
+ try {
+ server.start();
+ template.sendBody("direct:start", "");
+
+ latch.await();
+
+ HeaderMap headers = exchange.get().getRequestHeaders();
+
+ assertThat(headers.getFirst(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_VERSION).http())).isEqualTo(ce.version());
+ assertThat(headers.getFirst(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).http())).isEqualTo("myType");
+ assertThat(headers.getFirst(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_ID).http())).isNotNull();
+ assertThat(headers.getFirst(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE).http())).isEqualTo("knative://endpoint/ep");
+ assertThat(headers.getFirst(Exchange.CONTENT_TYPE)).isEqualTo("text/plain");
+ } finally {
+ server.stop();
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("provideCloudEventsImplementations")
+ void testHeadersOverrideFromRouteWithCEHeader(CloudEvent ce) throws Exception {
+ configureKnativeComponent(
+ context,
+ ce,
+ endpoint(
+ Knative.EndpointKind.sink,
+ "ep",
+ "localhost",
+ port,
+ KnativeSupport.mapOf(
+ Knative.KNATIVE_EVENT_TYPE, "org.apache.camel.event",
+ Knative.CONTENT_TYPE, "text/plain"
+ )
+ )
+ );
+
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicReference<HttpServerExchange> exchange = new AtomicReference<>();
+
+ Undertow server = Undertow.builder()
+ .addHttpListener(port, "localhost")
+ .setHandler(se -> {
+ exchange.set(se);
+ latch.countDown();
+ })
+ .build();
+
+ RouteBuilder.addRoutes(context, b -> {
+ b.from("direct:start")
+ .setHeader(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).http()).constant("fromCEHeader")
+ .setHeader(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).constant("fromCamelHeader")
+ .to("knative:endpoint/ep");
+ });
+
+ context.start();
+ try {
+ server.start();
+ template.sendBody("direct:start", "");
+
+ latch.await();
+
+ HeaderMap headers = exchange.get().getRequestHeaders();
+
+ assertThat(headers.getFirst(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_VERSION).http())).isEqualTo(ce.version());
+ assertThat(headers.getFirst(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).http())).isEqualTo("fromCEHeader");
+ assertThat(headers.getFirst(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_ID).http())).isNotNull();
+ assertThat(headers.getFirst(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE).http())).isEqualTo("knative://endpoint/ep");
+ assertThat(headers.getFirst(Exchange.CONTENT_TYPE)).isEqualTo("text/plain");
+ } finally {
+ server.stop();
+ }
+ }
}
diff --git a/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/ce/AbstractCloudEventProcessor.java b/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/ce/AbstractCloudEventProcessor.java
index 4cd638a..9a4e550 100644
--- a/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/ce/AbstractCloudEventProcessor.java
+++ b/camel-knative/camel-knative/src/main/java/org/apache/camel/component/knative/ce/AbstractCloudEventProcessor.java
@@ -22,6 +22,7 @@ import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;
import java.util.Objects;
+import java.util.function.Supplier;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -55,7 +56,7 @@ abstract class AbstractCloudEventProcessor implements CloudEventProcessor {
final Map<String, Object> headers = exchange.getIn().getHeaders();
for (CloudEvent.Attribute attribute: ce.attributes()) {
- Object val = headers.remove(attribute.http());
+ Object val = headers.get(attribute.http());
if (val != null) {
headers.put(attribute.id(), val);
}
@@ -75,23 +76,31 @@ abstract class AbstractCloudEventProcessor implements CloudEventProcessor {
final CloudEvent ce = cloudEvent();
return exchange -> {
- String eventType = service.getMetadata().get(Knative.KNATIVE_EVENT_TYPE);
- if (eventType == null) {
- eventType = endpoint.getConfiguration().getCloudEventsType();
+ final String contentType = service.getMetadata().get(Knative.CONTENT_TYPE);
+ final Map<String, Object> headers = exchange.getMessage().getHeaders();
+
+ for (CloudEvent.Attribute attribute: ce.attributes()) {
+ Object value = headers.get(attribute.id());
+ if (value != null) {
+ headers.putIfAbsent(attribute.http(), value);
+ }
}
- final String contentType = service.getMetadata().get(Knative.CONTENT_TYPE);
- final ZonedDateTime created = exchange.getCreated().toInstant().atZone(ZoneId.systemDefault());
- final String eventTime = DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(created);
- final Map<String, Object> headers = exchange.getIn().getHeaders();
-
- headers.putIfAbsent(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_ID).http(), exchange.getExchangeId());
- headers.putIfAbsent(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_SOURCE).http(), endpoint.getEndpointUri());
- headers.putIfAbsent(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_VERSION).http(), ce.version());
- headers.putIfAbsent(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TYPE).http(), eventType);
- headers.putIfAbsent(ce.mandatoryAttribute(CloudEvent.CAMEL_CLOUD_EVENT_TIME).http(), eventTime);
headers.putIfAbsent(Exchange.CONTENT_TYPE, contentType);
+ setCloudEventHeader(headers, CloudEvent.CAMEL_CLOUD_EVENT_ID, exchange::getExchangeId);
+ setCloudEventHeader(headers, CloudEvent.CAMEL_CLOUD_EVENT_SOURCE, endpoint::getEndpointUri);
+ setCloudEventHeader(headers, CloudEvent.CAMEL_CLOUD_EVENT_VERSION, ce::version);
+ setCloudEventHeader(headers, CloudEvent.CAMEL_CLOUD_EVENT_TYPE, () -> {
+ return service.getMetadata().getOrDefault(Knative.KNATIVE_EVENT_TYPE, endpoint.getConfiguration().getCloudEventsType());
+ });
+ setCloudEventHeader(headers, CloudEvent.CAMEL_CLOUD_EVENT_TIME, () -> {
+ final ZonedDateTime created = exchange.getCreated().toInstant().atZone(ZoneId.systemDefault());
+ final String eventTime = DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(created);
+
+ return eventTime;
+ });
+
for (Map.Entry<String, String> entry: service.getMetadata().entrySet()) {
if (entry.getKey().startsWith(Knative.KNATIVE_CE_OVERRIDE_PREFIX)) {
final String key = entry.getKey().substring(Knative.KNATIVE_CE_OVERRIDE_PREFIX.length());
@@ -102,4 +111,8 @@ abstract class AbstractCloudEventProcessor implements CloudEventProcessor {
}
};
}
+
+ protected void setCloudEventHeader(Map<String, Object> headers, String id, Supplier<Object> supplier) {
+ headers.putIfAbsent(cloudEvent().mandatoryAttribute(id).http(), supplier.get());
+ }
}