This is an automated email from the ASF dual-hosted git repository.
Croway pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a8edb80b47f4 CAMEL-25888: camel-pubnub - the producer uses its
configured channel unless an override channel header is set
a8edb80b47f4 is described below
commit a8edb80b47f4d8adbe06361d357efd7ca626edb3
Author: croway <[email protected]>
AuthorDate: Tue Oct 20 14:19:24 2026 +0200
CAMEL-25888: camel-pubnub - the producer uses its configured channel unless
an override channel header is set
The pubnub consumer sets CamelPubNubChannel to the channel each message or
presence event was received on, and the producer preferred that header over
the channel configured on its endpoint, so a route forwarding messages from
a
pubnub consumer to another pubnub endpoint published them back to the
channel
they came from.
The producer now uses its configured channel. The new
CamelPubNubOverrideChannel
header selects a different channel explicitly; CamelPubNubChannel stays
consumer
metadata. The 4.24 upgrade guide lists the change.
---
.../apache/camel/catalog/components/pubnub.json | 3 +-
.../org/apache/camel/component/pubnub/pubnub.json | 3 +-
.../camel/component/pubnub/PubNubConstants.java | 3 +
.../camel/component/pubnub/PubNubProducer.java | 2 +-
.../camel/component/pubnub/PubNubPublishTest.java | 32 +++++++++
.../component/pubnub/PubNubRouteChannelTest.java | 79 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 +++
.../endpoint/dsl/PubNubEndpointBuilderFactory.java | 12 ++++
8 files changed, 140 insertions(+), 3 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pubnub.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pubnub.json
index 7b2f79801dc3..dd9a6e85c213 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pubnub.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pubnub.json
@@ -41,7 +41,8 @@
"CamelPubNubOperation": { "index": 0, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The operation to perform.", "constantName":
"org.apache.camel.component.pubnub.PubNubConstants#OPERATION" },
"CamelPubNubTimeToken": { "index": 1, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The Timestamp for the event.", "constantName":
"org.apache.camel.component.pubnub.PubNubConstants#TIMETOKEN" },
"CamelPubNubChannel": { "index": 2, "kind": "header", "displayName": "",
"group": "consumer", "label": "consumer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The channel for which the message belongs.",
"constantName": "org.apache.camel.component.pubnub.PubNubConstants#CHANNEL" },
- "CamelPubNubUUID": { "index": 3, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "UUID to be used as a device identifier.",
"constantName": "org.apache.camel.component.pubnub.PubNubConstants#UUID" }
+ "CamelPubNubOverrideChannel": { "index": 3, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The channel to use instead of the
channel configured on the endpoint.", "constantName":
"org.apache.camel.component.pubnub.PubNubConstants#OVERRIDE_CHANNEL" },
+ "CamelPubNubUUID": { "index": 4, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "UUID to be used as a device identifier.",
"constantName": "org.apache.camel.component.pubnub.PubNubConstants#UUID" }
},
"properties": {
"channel": { "index": 0, "kind": "path", "displayName": "Channel",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.pubnub.PubNubConfiguration", "configurationField":
"configuration", "description": "The channel used for subscribing\/publishing
events" },
diff --git
a/components/camel-pubnub/src/generated/resources/META-INF/org/apache/camel/component/pubnub/pubnub.json
b/components/camel-pubnub/src/generated/resources/META-INF/org/apache/camel/component/pubnub/pubnub.json
index 7b2f79801dc3..dd9a6e85c213 100644
---
a/components/camel-pubnub/src/generated/resources/META-INF/org/apache/camel/component/pubnub/pubnub.json
+++
b/components/camel-pubnub/src/generated/resources/META-INF/org/apache/camel/component/pubnub/pubnub.json
@@ -41,7 +41,8 @@
"CamelPubNubOperation": { "index": 0, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The operation to perform.", "constantName":
"org.apache.camel.component.pubnub.PubNubConstants#OPERATION" },
"CamelPubNubTimeToken": { "index": 1, "kind": "header", "displayName": "",
"group": "common", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The Timestamp for the event.", "constantName":
"org.apache.camel.component.pubnub.PubNubConstants#TIMETOKEN" },
"CamelPubNubChannel": { "index": 2, "kind": "header", "displayName": "",
"group": "consumer", "label": "consumer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The channel for which the message belongs.",
"constantName": "org.apache.camel.component.pubnub.PubNubConstants#CHANNEL" },
- "CamelPubNubUUID": { "index": 3, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "UUID to be used as a device identifier.",
"constantName": "org.apache.camel.component.pubnub.PubNubConstants#UUID" }
+ "CamelPubNubOverrideChannel": { "index": 3, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The channel to use instead of the
channel configured on the endpoint.", "constantName":
"org.apache.camel.component.pubnub.PubNubConstants#OVERRIDE_CHANNEL" },
+ "CamelPubNubUUID": { "index": 4, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "UUID to be used as a device identifier.",
"constantName": "org.apache.camel.component.pubnub.PubNubConstants#UUID" }
},
"properties": {
"channel": { "index": 0, "kind": "path", "displayName": "Channel",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.pubnub.PubNubConfiguration", "configurationField":
"configuration", "description": "The channel used for subscribing\/publishing
events" },
diff --git
a/components/camel-pubnub/src/main/java/org/apache/camel/component/pubnub/PubNubConstants.java
b/components/camel-pubnub/src/main/java/org/apache/camel/component/pubnub/PubNubConstants.java
index 8e26e70c25c4..af3bf330424a 100644
---
a/components/camel-pubnub/src/main/java/org/apache/camel/component/pubnub/PubNubConstants.java
+++
b/components/camel-pubnub/src/main/java/org/apache/camel/component/pubnub/PubNubConstants.java
@@ -25,6 +25,9 @@ public final class PubNubConstants {
public static final String TIMETOKEN = "CamelPubNubTimeToken";
@Metadata(label = "consumer", description = "The channel for which the
message belongs.", javaType = "String")
public static final String CHANNEL = "CamelPubNubChannel";
+ @Metadata(label = "producer", description = "The channel to use instead of
the channel configured on the endpoint.",
+ javaType = "String")
+ public static final String OVERRIDE_CHANNEL = "CamelPubNubOverrideChannel";
@Metadata(label = "producer", description = "UUID to be used as a device
identifier.", javaType = "String")
public static final String UUID = "CamelPubNubUUID";
diff --git
a/components/camel-pubnub/src/main/java/org/apache/camel/component/pubnub/PubNubProducer.java
b/components/camel-pubnub/src/main/java/org/apache/camel/component/pubnub/PubNubProducer.java
index 448771cc4868..54c0cb402169 100644
---
a/components/camel-pubnub/src/main/java/org/apache/camel/component/pubnub/PubNubProducer.java
+++
b/components/camel-pubnub/src/main/java/org/apache/camel/component/pubnub/PubNubProducer.java
@@ -257,7 +257,7 @@ public class PubNubProducer extends DefaultAsyncProducer {
}
private String getChannel(Exchange exchange) {
- String channel = exchange.getIn().getHeader(PubNubConstants.CHANNEL,
String.class);
+ String channel =
exchange.getIn().getHeader(PubNubConstants.OVERRIDE_CHANNEL, String.class);
return channel != null ? channel : pubnubConfiguration.getChannel();
}
diff --git
a/components/camel-pubnub/src/test/java/org/apache/camel/component/pubnub/PubNubPublishTest.java
b/components/camel-pubnub/src/test/java/org/apache/camel/component/pubnub/PubNubPublishTest.java
index 6e57d1c06423..110339c27dec 100644
---
a/components/camel-pubnub/src/test/java/org/apache/camel/component/pubnub/PubNubPublishTest.java
+++
b/components/camel-pubnub/src/test/java/org/apache/camel/component/pubnub/PubNubPublishTest.java
@@ -25,8 +25,10 @@ import org.junit.jupiter.api.Test;
import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
import static com.github.tomakehurst.wiremock.client.WireMock.get;
import static com.github.tomakehurst.wiremock.client.WireMock.post;
+import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.verify;
import static org.apache.camel.component.pubnub.PubNubConstants.TIMETOKEN;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -63,6 +65,36 @@ public class PubNubPublishTest extends PubNubTestBase {
MockEndpoint.assertIsSatisfied(context);
}
+ @Test
+ public void publishUsesConfiguredChannelWhenChannelHeaderIsPresent()
throws Exception {
+
stubFor(post(urlPathEqualTo("/publish/myPublishKey/mySubscribeKey/0/someChannel/0"))
+
.willReturn(aResponse().withStatus(200).withBody("[1,\"Sent\",\"14598111595318003\"]")));
+
stubFor(post(urlPathEqualTo("/publish/myPublishKey/mySubscribeKey/0/otherChannel/0"))
+
.willReturn(aResponse().withStatus(200).withBody("[1,\"Sent\",\"14598111595318004\"]")));
+ mockResult.expectedMessageCount(1);
+
+ template.sendBodyAndHeader("direct:publish", new Hello("Hi"),
PubNubConstants.CHANNEL, "otherChannel");
+ MockEndpoint.assertIsSatisfied(context);
+
+ verify(1,
postRequestedFor(urlPathEqualTo("/publish/myPublishKey/mySubscribeKey/0/someChannel/0")));
+ verify(0,
postRequestedFor(urlPathEqualTo("/publish/myPublishKey/mySubscribeKey/0/otherChannel/0")));
+ }
+
+ @Test
+ public void publishUsesOverrideChannelHeader() throws Exception {
+
stubFor(post(urlPathEqualTo("/publish/myPublishKey/mySubscribeKey/0/someChannel/0"))
+
.willReturn(aResponse().withStatus(200).withBody("[1,\"Sent\",\"14598111595318003\"]")));
+
stubFor(post(urlPathEqualTo("/publish/myPublishKey/mySubscribeKey/0/otherChannel/0"))
+
.willReturn(aResponse().withStatus(200).withBody("[1,\"Sent\",\"14598111595318004\"]")));
+ mockResult.expectedMessageCount(1);
+
+ template.sendBodyAndHeader("direct:publish", new Hello("Hi"),
PubNubConstants.OVERRIDE_CHANNEL, "otherChannel");
+ MockEndpoint.assertIsSatisfied(context);
+
+ verify(1,
postRequestedFor(urlPathEqualTo("/publish/myPublishKey/mySubscribeKey/0/otherChannel/0")));
+ verify(0,
postRequestedFor(urlPathEqualTo("/publish/myPublishKey/mySubscribeKey/0/someChannel/0")));
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
diff --git
a/components/camel-pubnub/src/test/java/org/apache/camel/component/pubnub/PubNubRouteChannelTest.java
b/components/camel-pubnub/src/test/java/org/apache/camel/component/pubnub/PubNubRouteChannelTest.java
new file mode 100644
index 000000000000..f44191b9016c
--- /dev/null
+++
b/components/camel-pubnub/src/test/java/org/apache/camel/component/pubnub/PubNubRouteChannelTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.pubnub;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.findAll;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.post;
+import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class PubNubRouteChannelTest extends PubNubTestBase {
+
+ private static final String INBOUND_PUBLISH =
"/publish/myPublishKey/mySubscribeKey/0/inbound/0";
+ private static final String OUTBOUND_PUBLISH =
"/publish/myPublishKey/mySubscribeKey/0/outbound/0";
+
+ @EndpointInject("mock:result")
+ private MockEndpoint mockResult;
+
+ @Test
+ public void forwardedEventIsPublishedToProducerChannel() throws Exception {
+ stubFor(get(urlPathEqualTo("/v2/subscribe/mySubscribeKey/inbound/0"))
+ .willReturn(aResponse()
+ .withBody(
+
"{\"t\":{\"t\":\"14607577960932487\",\"r\":1},\"m\":[{\"a\":\"4\",\"f\":0,\"i\":\"sender\",\"p\":{\"t\":\"14607577960925503\",\"r\":1},\"o\":"
+ +
"{\"t\":\"14737141991877032\",\"r\":2},\"k\":\"sub-c-mykey\",\"c\":\"inbound\",\"d\":{\"text\":\"Message\"},\"b\":\"inbound\"}]}")));
+
stubFor(get(urlPathEqualTo("/v2/presence/sub-key/mySubscribeKey/channel/inbound/heartbeat"))
+ .willReturn(aResponse().withBody("{\"status\": 200,
\"message\": \"OK\", \"service\": \"Presence\"}")));
+ stubFor(post(urlPathEqualTo(OUTBOUND_PUBLISH))
+
.willReturn(aResponse().withStatus(200).withBody("[1,\"Sent\",\"14598111595318003\"]")));
+ stubFor(post(urlPathEqualTo(INBOUND_PUBLISH))
+
.willReturn(aResponse().withStatus(200).withBody("[1,\"Sent\",\"14598111595318004\"]")));
+
+ mockResult.expectedMinimumMessageCount(1);
+ mockResult.expectedHeaderReceived(PubNubConstants.CHANNEL, "inbound");
+ context.getRouteController().startRoute("forward");
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ context.getRouteController().stopRoute("forward");
+
+
assertTrue(findAll(postRequestedFor(urlPathEqualTo(OUTBOUND_PUBLISH))).size()
>= 1,
+ "The forwarded event should be published to the producer's
channel");
+
assertTrue(findAll(postRequestedFor(urlPathEqualTo(INBOUND_PUBLISH))).isEmpty(),
+ "The forwarded event should not be published to the channel it
was received on");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+
from("pubnub:inbound?pubnub=#pubnub").id("forward").autoStartup(false)
+ .to("pubnub:outbound?pubnub=#pubnub")
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 4f5215ef30ed..d403cecd0527 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1383,6 +1383,15 @@ nothing was stored and the exchange completed without an
error. The requests are
the answer like the other state operations: the states are stored, and a
request that Dapr rejects (or a sidecar that
cannot be reached) fails the exchange.
+=== camel-pubnub - the producer uses its configured channel
+
+The pubnub producer no longer reads the `CamelPubNubChannel` header. This
header is set by the pubnub consumer to the
+channel a message was received on, and the producer used it instead of the
channel configured on its endpoint, so a
+route from one pubnub endpoint to another sent each message back to the
channel it came from. The producer now uses the
+channel of its endpoint, for every operation. To choose a different channel
for a message, set the new
+`CamelPubNubOverrideChannel` header (`PubNubConstants.OVERRIDE_CHANNEL`).
Routes that set `CamelPubNubChannel` to pick
+the channel of a pubnub producer must set `CamelPubNubOverrideChannel` instead.
+
=== camel-docker - the events consumer subscribes again when its event stream
ends
The `docker:events` consumer subscribes again when its event stream fails or
ends (for example when the Docker daemon
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PubNubEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PubNubEndpointBuilderFactory.java
index 7172698c98a3..f38b6532f890 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PubNubEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PubNubEndpointBuilderFactory.java
@@ -840,6 +840,18 @@ public interface PubNubEndpointBuilderFactory {
public String pubNubChannel() {
return "CamelPubNubChannel";
}
+ /**
+ * The channel to use instead of the channel configured on the
endpoint.
+ *
+ * The option is a: {@code String} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code PubNubOverrideChannel}.
+ */
+ public String pubNubOverrideChannel() {
+ return "CamelPubNubOverrideChannel";
+ }
/**
* UUID to be used as a device identifier.
*
