Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -19,23 +19,31 @@
import java.io.InputStream;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.function.Supplier;

import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.cloudevents.CloudEvent;
import org.apache.camel.component.knative.KnativeEndpoint;
import org.apache.camel.component.knative.spi.Knative;
import org.apache.camel.component.knative.spi.KnativeResource;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.support.DefaultHeaderFilterStrategy;
import org.apache.camel.util.StringHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

abstract class AbstractCloudEventProcessor implements CloudEventProcessor {
private final CloudEvent cloudEvent;

// Filters internal Camel headers (Camel*/camel*) from structured-mode CloudEvent extensions, keeping the
// extension-to-header mapping consistent with the binary content-mode path (see KnativeHttpHeaderFilterStrategy).
private final HeaderFilterStrategy headerFilterStrategy = new DefaultHeaderFilterStrategy();

protected AbstractCloudEventProcessor(CloudEvent cloudEvent) {
this.cloudEvent = cloudEvent;
}
Expand Down Expand Up @@ -72,6 +80,17 @@ public Processor consumer(KnativeEndpoint endpoint, KnativeResource service) {

protected abstract void decodeStructuredContent(Exchange exchange, Map<String, Object> content);

/**
* Maps a structured-mode CloudEvent extension field to a message header, applying the header filter strategy so
* that internal Camel ({@code Camel*}) headers are filtered consistently with the binary content-mode path.
*/
protected void mapExtensionAsHeader(Message message, Exchange exchange, String key, Object value) {
final String headerName = key.toLowerCase(Locale.US);
if (!headerFilterStrategy.applyFilterToExternalHeaders(headerName, value, exchange)) {
message.setHeader(headerName, value);
}
}

@Override
public Processor producer(KnativeEndpoint endpoint, KnativeResource service) {
final CloudEvent ce = cloudEvent();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
*/
package org.apache.camel.component.knative.ce;

import java.util.Locale;
import java.util.Map;
import java.util.Objects;

Expand Down Expand Up @@ -52,11 +51,10 @@ protected void decodeStructuredContent(Exchange exchange, Map<String, Object> co
}

//
// Map every remaining field as it is (extensions).
// Map every remaining field as it is (extensions), applying the header filter
// strategy so internal Camel headers are filtered consistently with binary mode.
//
content.forEach((key, val) -> {
message.setHeader(key.toLowerCase(Locale.US), val);
});
content.forEach((key, val) -> mapExtensionAsHeader(message, exchange, key, val));
}
}),
v1_0_1(new AbstractCloudEventProcessor(CloudEvents.v1_0_1) {
Expand All @@ -79,11 +77,10 @@ protected void decodeStructuredContent(Exchange exchange, Map<String, Object> co
}

//
// Map every remaining field as it is (extensions).
// Map every remaining field as it is (extensions), applying the header filter
// strategy so internal Camel headers are filtered consistently with binary mode.
//
content.forEach((key, val) -> {
message.setHeader(key.toLowerCase(Locale.US), val);
});
content.forEach((key, val) -> mapExtensionAsHeader(message, exchange, key, val));
}
}),
v1_0_2(new AbstractCloudEventProcessor(CloudEvents.v1_0_2) {
Expand All @@ -106,11 +103,10 @@ protected void decodeStructuredContent(Exchange exchange, Map<String, Object> co
}

//
// Map every remaining field as it is (extensions).
// Map every remaining field as it is (extensions), applying the header filter
// strategy so internal Camel headers are filtered consistently with binary mode.
//
content.forEach((key, val) -> {
message.setHeader(key.toLowerCase(Locale.US), val);
});
content.forEach((key, val) -> mapExtensionAsHeader(message, exchange, key, val));
}
});

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -379,6 +379,61 @@ void testConsumeStructuredContent(CloudEvent ce) throws Exception {
mock.assertIsSatisfied();
}

@ParameterizedTest
@EnumSource(CloudEvents.class)
void testConsumeStructuredContentFiltersInternalHeaders(CloudEvent ce) throws Exception {
configureKnativeComponent(
context,
ce,
sourceEndpoint(
"myEndpoint",
Map.of(
Knative.KNATIVE_CLOUD_EVENT_TYPE, "org.apache.camel.event",
Knative.CONTENT_TYPE, "text/plain")));

RouteBuilder.addRoutes(context, b -> {
b.from("knative:endpoint/myEndpoint")
.to("mock:ce");
});

context.start();

MockEndpoint mock = context.getEndpoint("mock:ce", MockEndpoint.class);
mock.expectedBodiesReceived("test");
mock.expectedMessageCount(1);
// a regular CloudEvent extension must still be propagated as a message header
mock.expectedHeaderReceived("myextension", "myvalue");
// an extension that resolves to an internal Camel header must be filtered out and must not
// be able to inject a framework control header (headers are matched case-insensitively)
mock.expectedMessagesMatches(e -> e.getMessage().getHeader(Exchange.FILE_NAME) == null);

if (Objects.equals(CloudEvents.v1_0.version(), ce.version())
|| Objects.equals(CloudEvents.v1_0_1.version(), ce.version())
|| Objects.equals(CloudEvents.v1_0_2.version(), ce.version())) {
given()
.contentType(Knative.MIME_STRUCTURED_CONTENT_MODE)
.body(
Map.of(
"specversion", ce.version(),
"type", "org.apache.camel.event",
"id", "myEventID",
"source", "/somewhere",
"datacontenttype", "text/plain",
"myextension", "myvalue",
"camelfilename", "injected.txt",
"data", "test"),
ObjectMapperType.JACKSON_2)
.when()
.post()
.then()
.statusCode(200);
} else {
throw new IllegalArgumentException("Unknown CloudEvent spec: " + ce.version());
}

mock.assertIsSatisfied();
}

@ParameterizedTest
@EnumSource(CloudEvents.class)
void testConsumeContent(CloudEvent ce) throws Exception {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,17 @@ the message, consistent with the inbound header filtering already performed by t
Ordinary application headers are unaffected. If a route relied on `Camel*` headers being propagated from the MIME
content, set them explicitly after unmarshalling.

=== camel-knative - structured-mode CloudEvent header filtering

When consuming a CloudEvent in structured content mode (`application/cloudevents+json`), the Knative component now
applies a `HeaderFilterStrategy` to the event fields (extensions) mapped from the payload onto the Camel message.
Camel-internal headers (the `Camel*` namespace, matched case-insensitively) present as structured-event fields are
no longer mapped onto the message, consistent with the inbound header filtering already performed on the binary
content-mode / HTTP header path.

Ordinary CloudEvent extension attributes are unaffected. If a route relied on `Camel*`-named fields being
propagated from the structured payload, set them explicitly after consuming the event.

=== camel-pinecone

The `tls` endpoint option now correctly documents its default as `false` (previously the catalog
Expand Down