This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 ba7215a7e99d [camel-cometd] Automate
CometdProducerConsumerInteractiveAuthenticatedManualTest (#21938)
ba7215a7e99d is described below
commit ba7215a7e99df568f4b4fd0c9637c718b8e87158
Author: JinyuChen97 <[email protected]>
AuthorDate: Wed Mar 11 18:52:47 2026 +0000
[camel-cometd] Automate
CometdProducerConsumerInteractiveAuthenticatedManualTest (#21938)
https://issues.apache.org/jira/browse/CAMEL-22515
---
components/camel-cometd/pom.xml | 12 ++
.../CometdProducerConsumerAuthenticatedTest.java | 219 +++++++++++++++++++++
.../CometdProducerConsumerExtensionTest.java | 202 +++++++++++++++++++
...ProducerConsumerInOutInteractiveManualTest.java | 84 --------
.../cometd/CometdProducerConsumerInOutTest.java | 151 ++++++++++++++
...ConsumerInteractiveAuthenticatedManualTest.java | 166 ----------------
...ucerConsumerInteractiveExtensionManualTest.java | 109 ----------
...ometdProducerConsumerInteractiveManualTest.java | 65 ------
.../CometdProducerConsumerInteractiveTest.java | 74 +++++++
9 files changed, 658 insertions(+), 424 deletions(-)
diff --git a/components/camel-cometd/pom.xml b/components/camel-cometd/pom.xml
index d69a9b496be4..548db4fd7eab 100644
--- a/components/camel-cometd/pom.xml
+++ b/components/camel-cometd/pom.xml
@@ -92,6 +92,18 @@
<version>${mockito-version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.cometd.java</groupId>
+ <artifactId>cometd-java-client-http-jetty</artifactId>
+ <version>${cometd-java-client-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-client</artifactId>
+ <version>${jetty-version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerAuthenticatedTest.java
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerAuthenticatedTest.java
new file mode 100644
index 000000000000..741fc5b70c7a
--- /dev/null
+++
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerAuthenticatedTest.java
@@ -0,0 +1,219 @@
+/*
+ * 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.cometd;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.cometd.bayeux.Channel;
+import org.cometd.bayeux.Message;
+import org.cometd.bayeux.client.ClientSession;
+import org.cometd.bayeux.server.BayeuxServer;
+import org.cometd.bayeux.server.ServerMessage;
+import org.cometd.bayeux.server.ServerSession;
+import org.cometd.client.BayeuxClient;
+import org.cometd.client.http.jetty.JettyHttpClientTransport;
+import org.cometd.server.DefaultSecurityPolicy;
+import org.eclipse.jetty.client.HttpClient;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class CometdProducerConsumerAuthenticatedTest extends CamelTestSupport {
+
+ private int port;
+ private String uri;
+
+ @Test
+ void testAuthenticatedProducerConsumer() throws Exception {
+ getMockEndpoint("mock:test").expectedBodiesReceived("hello");
+ template.sendBody("direct:input", "hello");
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Test
+ void testHandshakeWithValidCredentials() throws Exception {
+ HttpClient httpClient = new HttpClient();
+ httpClient.start();
+ try {
+ BayeuxClient client = createRemoteClient(httpClient, "changeit",
"changeit");
+ client.handshake();
+ try {
+ assertTrue(client.waitFor(5000, BayeuxClient.State.CONNECTED),
+ "Handshake with valid credentials should reach
CONNECTED state");
+ } finally {
+ client.disconnect();
+ client.waitFor(5000, BayeuxClient.State.DISCONNECTED);
+ }
+ } finally {
+ httpClient.stop();
+ }
+ }
+
+ @Test
+ void testHandshakeWithInvalidCredentials() throws Exception {
+ HttpClient httpClient = new HttpClient();
+ httpClient.start();
+ try {
+ BayeuxClient client = createRemoteClient(httpClient, "changeit",
"wrongpassword");
+ client.handshake();
+ try {
+ // Server rejects invalid credentials. client never reaches
CONNECTED within timeout
+ assertFalse(client.waitFor(3000, BayeuxClient.State.CONNECTED),
+ "Handshake with invalid credentials should not reach
CONNECTED state");
+ } finally {
+ client.disconnect();
+ client.waitFor(5000, BayeuxClient.State.DISCONNECTED);
+ }
+ } finally {
+ httpClient.stop();
+ }
+ }
+
+ private BayeuxClient createRemoteClient(HttpClient httpClient, String
user, String credentials) {
+ String url = "http://127.0.0.1:"; + port + "/cometd";
+ BayeuxClient client = new BayeuxClient(url, new
JettyHttpClientTransport(null, httpClient));
+ client.addExtension(new ClientSession.Extension() {
+ @Override
+ public boolean sendMeta(ClientSession session,
org.cometd.bayeux.Message.Mutable message) {
+ if (Channel.META_HANDSHAKE.equals(message.getChannel())) {
+ Map<String, Object> authentication = new HashMap<>();
+ authentication.put("user", user);
+ authentication.put("credentials", credentials);
+ Map<String, Object> ext = message.getExt(true);
+ ext.put("authentication", authentication);
+ }
+ return true;
+ }
+ });
+ return client;
+ }
+
+ @Override
+ public void setupResources() {
+ port = AvailablePortFinder.getNextAvailable();
+ uri = "cometd://127.0.0.1:" + port +
"/service/test?baseResource=file:./target/test-classes/webapp&"
+ +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ CometdComponent component = context.getComponent("cometd",
CometdComponent.class);
+ CometdProducerConsumerAuthenticatedTest.BayeuxAuthenticator
bayeuxAuthenticator
+ = new
CometdProducerConsumerAuthenticatedTest.BayeuxAuthenticator();
+ component.setSecurityPolicy(bayeuxAuthenticator);
+ component.addExtension(bayeuxAuthenticator);
+
+ from("direct:input").to(uri);
+ from(uri).to("mock:test");
+ }
+ };
+ }
+
+ /**
+ * Custom SecurityPolicy, see
http://cometd.org/documentation/howtos/authentication for details
+ */
+ public static final class BayeuxAuthenticator extends DefaultSecurityPolicy
+ implements BayeuxServer.Extension, ServerSession.RemovedListener {
+
+ private String user = "changeit";
+ private String pwd = "changeit";
+
+ @Override
+ public boolean canHandshake(BayeuxServer server, ServerSession
session, ServerMessage message) {
+ if (session.isLocalSession()) {
+ return true;
+ }
+
+ Map<String, Object> ext = message.getExt();
+ if (ext == null) {
+ return false;
+ }
+
+ @SuppressWarnings("unchecked")
+ Map<String, Object> authentication = (Map<String, Object>)
ext.get("authentication");
+ if (authentication == null) {
+ return false;
+ }
+
+ Object authenticationData = verify(authentication);
+ if (authenticationData == null) {
+ return false;
+ }
+
+ session.addListener(this);
+
+ return true;
+ }
+
+ private Object verify(Map<String, Object> authentication) {
+ if (!user.equals(authentication.get("user"))) {
+ return null;
+ }
+ if (!pwd.equals(authentication.get("credentials"))) {
+ return null;
+ }
+ return "OK";
+ }
+
+ @Override
+ public boolean sendMeta(ServerSession to, ServerMessage.Mutable
message) {
+ if (Channel.META_HANDSHAKE.equals(message.getChannel())) {
+ if (!message.isSuccessful()) {
+ Map<String, Object> advice = message.getAdvice(true);
+ advice.put(Message.RECONNECT_FIELD,
Message.RECONNECT_HANDSHAKE_VALUE);
+
+ Map<String, Object> ext = message.getExt(true);
+ Map<String, Object> authentication = new HashMap<>();
+ ext.put("authentication", authentication);
+ authentication.put("failed", true);
+ authentication.put("failureReason", "invalid_credentials");
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public void removed(ServerSession session, ServerMessage message,
boolean timeout) {
+ // Remove authentication data
+ }
+
+ @Override
+ public boolean rcv(ServerSession from, ServerMessage.Mutable message) {
+ return true;
+ }
+
+ @Override
+ public boolean rcvMeta(ServerSession from, ServerMessage.Mutable
message) {
+ return true;
+ }
+
+ @Override
+ public boolean send(ServerSession from, ServerSession to,
ServerMessage.Mutable message) {
+ return true;
+ }
+ }
+
+}
diff --git
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerExtensionTest.java
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerExtensionTest.java
new file mode 100644
index 000000000000..dae16aab5d14
--- /dev/null
+++
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerExtensionTest.java
@@ -0,0 +1,202 @@
+/*
+ * 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.cometd;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.cometd.bayeux.server.BayeuxServer;
+import org.cometd.bayeux.server.ServerMessage;
+import org.cometd.bayeux.server.ServerSession;
+import org.cometd.client.BayeuxClient;
+import org.cometd.client.http.jetty.JettyHttpClientTransport;
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class CometdProducerConsumerExtensionTest extends CamelTestSupport {
+
+ private int port;
+ private String uri;
+
+ private int portSSL;
+ private String uriSSL;
+
+ private String pwd = "changeit";
+
+ @Test
+ void testCensorExtensionReplaceForbiddenWords() throws Exception {
+ List<Object> received = new CopyOnWriteArrayList<>();
+ CountDownLatch latch = new CountDownLatch(3);
+
+ HttpClient httpClient = new HttpClient();
+ httpClient.start();
+ try {
+ BayeuxClient client = new BayeuxClient(
+ "http://127.0.0.1:"; + port + "/cometd", new
JettyHttpClientTransport(null, httpClient));
+ client.handshake();
+ assertTrue(client.waitFor(5000, BayeuxClient.State.CONNECTED));
+
+ // Wait for subscription to be confirmed before publishing
+ CountDownLatch subscribed = new CountDownLatch(1);
+ client.getChannel("/channel/test").subscribe(
+ (channel, message) -> {
+ received.add(message.getData());
+ latch.countDown();
+ },
+ message -> subscribed.countDown());
+ assertTrue(subscribed.await(5, TimeUnit.SECONDS));
+
+ template.sendBody("direct:input", "one");
+ template.sendBody("direct:input", "two");
+ template.sendBody("direct:input", "hello");
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS));
+ assertEquals("***", received.get(0));
+ assertEquals("***", received.get(1));
+ assertEquals("hello", received.get(2));
+
+ client.disconnect();
+ client.waitFor(5000, BayeuxClient.State.DISCONNECTED);
+ } finally {
+ httpClient.stop();
+ }
+ }
+
+ @Test
+ void testCensorExtensionReplaceForbiddenWordsThroughSSL() throws Exception
{
+ List<Object> received = new CopyOnWriteArrayList<>();
+ CountDownLatch latch = new CountDownLatch(2);
+
+ URL keyStoreUrl =
CometdProducerConsumerInOutTest.class.getResource("/jsse/localhost.p12");
+ SslContextFactory.Client sslContextFactory = new
SslContextFactory.Client();
+ sslContextFactory.setTrustStorePath(keyStoreUrl.getPath());
+ sslContextFactory.setTrustStorePassword("changeit");
+ sslContextFactory.setTrustStoreType("PKCS12");
+ HttpClient httpClient = new HttpClient();
+ httpClient.setSslContextFactory(sslContextFactory);
+ httpClient.start();
+ try {
+ BayeuxClient client = new BayeuxClient(
+ "https://localhost:"; + portSSL + "/cometd", new
JettyHttpClientTransport(null, httpClient));
+ client.handshake();
+ assertTrue(client.waitFor(5000, BayeuxClient.State.CONNECTED));
+
+ // Wait for subscription to be confirmed before publishing
+ CountDownLatch subscribed = new CountDownLatch(1);
+ client.getChannel("/channel/test").subscribe(
+ (channel, message) -> {
+ received.add(message.getData());
+ latch.countDown();
+ },
+ message -> subscribed.countDown());
+ assertTrue(subscribed.await(5, TimeUnit.SECONDS));
+
+ template.sendBody("direct:input-ssl", "one");
+ template.sendBody("direct:input-ssl", "hello again");
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS));
+ assertEquals("***", received.get(0));
+ assertEquals("hello again", received.get(1));
+
+ client.disconnect();
+ client.waitFor(5000, BayeuxClient.State.DISCONNECTED);
+ } finally {
+ httpClient.stop();
+ }
+ }
+
+ @Override
+ public void setupResources() {
+ port = AvailablePortFinder.getNextAvailable();
+ uri = "cometd://127.0.0.1:" + port +
"/channel/test?baseResource=file:./target/test-classes/webapp&"
+ +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
+
+ portSSL = AvailablePortFinder.getNextAvailable();
+ uriSSL = "cometds://127.0.0.1:" + portSSL +
"/channel/test?baseResource=file:./target/test-classes/webapp&"
+ +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws URISyntaxException {
+ CometdComponent component = context.getComponent("cometd",
CometdComponent.class);
+ component.addExtension(new Censor());
+
+ CometdComponent component2 = (CometdComponent)
context.getComponent("cometds");
+ component2.setSslPassword(pwd);
+ component2.setSslKeyPassword(pwd);
+ URI keyStoreUrl
+ =
CometdProducerConsumerInOutTest.class.getResource("/jsse/localhost.p12").toURI();
+ component2.setSslKeystore(keyStoreUrl.getPath());
+ component2.addExtension(new Censor());
+
+ from("direct:input").to(uri);
+ from("direct:input-ssl").to(uriSSL);
+ }
+ };
+ }
+
+ public static final class Censor implements BayeuxServer.Extension,
ServerSession.RemovedListener {
+
+ private HashSet<String> forbidden = new HashSet<>(Arrays.asList("one",
"two"));
+
+ @Override
+ public void removed(ServerSession session, ServerMessage message,
boolean timeout) {
+ // called on remove of client
+ }
+
+ @Override
+ public boolean rcv(ServerSession from, ServerMessage.Mutable message) {
+ return true;
+ }
+
+ @Override
+ public boolean rcvMeta(ServerSession from, ServerMessage.Mutable
message) {
+ return true;
+ }
+
+ @Override
+ public boolean send(ServerSession from, ServerSession to,
ServerMessage.Mutable message) {
+ Object data = message.getData();
+ if (forbidden.contains(data)) {
+ message.put("data", "***");
+ }
+ return true;
+ }
+
+ @Override
+ public boolean sendMeta(ServerSession from, ServerMessage.Mutable
message) {
+ return true;
+ }
+ }
+}
diff --git
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInOutInteractiveManualTest.java
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInOutInteractiveManualTest.java
deleted file mode 100644
index 1526f26470b9..000000000000
---
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInOutInteractiveManualTest.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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.cometd;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-
-import org.apache.camel.CamelContext;
-import org.apache.camel.Exchange;
-import org.apache.camel.ExchangePattern;
-import org.apache.camel.Message;
-import org.apache.camel.Processor;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.impl.DefaultCamelContext;
-import org.apache.camel.support.DefaultMessage;
-import org.junit.jupiter.api.Disabled;
-
-@Disabled("Run this test manually")
-public class CometdProducerConsumerInOutInteractiveManualTest {
-
- private static final String URI =
"cometd://127.0.0.1:9091/service/test?baseResource=file:./src/test/resources/webapp&"
- +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
-
- private static final String URI2 =
"cometds://127.0.0.1:9443/service/test?baseResource=file:./src/test/resources/webapp&"
- +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
-
- private CamelContext context;
-
- private String pwd = "changeit";
-
- public static void main(String[] args) throws Exception {
- CometdProducerConsumerInOutInteractiveManualTest me = new
CometdProducerConsumerInOutInteractiveManualTest();
- me.testCometdProducerConsumerInteractive();
- }
-
- public void testCometdProducerConsumerInteractive() throws Exception {
- context = new DefaultCamelContext();
- context.addRoutes(createRouteBuilder());
- context.start();
- }
-
- private RouteBuilder createRouteBuilder() {
- return new RouteBuilder() {
- public void configure() throws URISyntaxException {
- CometdComponent component = (CometdComponent)
context.getComponent("cometds");
- component.setSslPassword(pwd);
- component.setSslKeyPassword(pwd);
- URI keyStoreUrl
- =
CometdProducerConsumerInOutInteractiveManualTest.class.getResource("/jsse/localhost.p12").toURI();
- component.setSslKeystore(keyStoreUrl.getPath());
-
-
from(URI).setExchangePattern(ExchangePattern.InOut).process(new Processor() {
- public void process(Exchange exchange) {
- Message out = new
DefaultMessage(exchange.getContext());
- out.setBody("reply: " + exchange.getIn().getBody());
- exchange.setOut(out);
- }
- });
-
from(URI2).setExchangePattern(ExchangePattern.InOut).process(new Processor() {
- public void process(Exchange exchange) {
- Message out = new
DefaultMessage(exchange.getContext());
- out.setBody("reply: " + exchange.getIn().getBody());
- exchange.setOut(out);
- }
- });
- }
- };
- }
-
-}
diff --git
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInOutTest.java
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInOutTest.java
new file mode 100644
index 000000000000..eacad4dbd6cf
--- /dev/null
+++
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInOutTest.java
@@ -0,0 +1,151 @@
+/*
+ * 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.cometd;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.cometd.client.BayeuxClient;
+import org.cometd.client.http.jetty.JettyHttpClientTransport;
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class CometdProducerConsumerInOutTest extends CamelTestSupport {
+
+ private int port;
+ private String uri;
+
+ private int portSSL;
+ private String uriSSL;
+
+ private String pwd = "changeit";
+
+ @Test
+ void testInOutExchangePattern() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicReference<String> receivedReply = new AtomicReference<>();
+
+ HttpClient httpClient = new HttpClient();
+ httpClient.start();
+ try {
+ String bayeuxUrl = "http://127.0.0.1:"; + port + "/cometd";
+ BayeuxClient client = new BayeuxClient(bayeuxUrl, new
JettyHttpClientTransport(null, httpClient));
+ client.handshake();
+ assertTrue(client.waitFor(5000, BayeuxClient.State.CONNECTED));
+
+ client.getChannel("/service/test").subscribe((channel, message) ->
{
+ receivedReply.set((String) message.getData());
+ latch.countDown();
+ });
+
+ client.getChannel("/service/test").publish("hello", null);
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS), "Should receive reply
within timeout");
+ assertEquals("reply: hello", receivedReply.get());
+
+ client.disconnect();
+ client.waitFor(5000, BayeuxClient.State.DISCONNECTED);
+ } finally {
+ httpClient.stop();
+ }
+ }
+
+ @Test
+ void testInOutExchangePatternSSL() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicReference<String> receivedReply = new AtomicReference<>();
+
+ URL keyStoreUrl =
CometdProducerConsumerInOutTest.class.getResource("/jsse/localhost.p12");
+ SslContextFactory.Client sslContextFactory = new
SslContextFactory.Client();
+ sslContextFactory.setTrustStorePath(keyStoreUrl.getPath());
+ sslContextFactory.setTrustStorePassword("changeit");
+ sslContextFactory.setTrustStoreType("PKCS12");
+ HttpClient httpClient = new HttpClient();
+ httpClient.setSslContextFactory(sslContextFactory);
+ httpClient.start();
+ try {
+ // Use localhost to match the CN in localhost.p12 and endpoint is
still /cometd as this is the only path added to servlet
+ String bayeuxUrl = "https://localhost:"; + portSSL + "/cometd";
+ BayeuxClient client = new BayeuxClient(bayeuxUrl, new
JettyHttpClientTransport(null, httpClient));
+ client.handshake();
+ assertTrue(client.waitFor(5000, BayeuxClient.State.CONNECTED));
+
+ client.getChannel("/service/test").subscribe((channel, message) ->
{
+ receivedReply.set((String) message.getData());
+ latch.countDown();
+ });
+
+ client.getChannel("/service/test").publish("hello", null);
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS), "Should receive reply
within timeout");
+ assertEquals("reply SSL: hello", receivedReply.get());
+
+ client.disconnect();
+ client.waitFor(5000, BayeuxClient.State.DISCONNECTED);
+ } finally {
+ httpClient.stop();
+ }
+ }
+
+ @Override
+ public void setupResources() {
+ port = AvailablePortFinder.getNextAvailable();
+ uri = "cometd://127.0.0.1:" + port +
"/service/test?baseResource=file:./target/test-classes/webapp&"
+ +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
+
+ portSSL = AvailablePortFinder.getNextAvailable();
+ //cometds protocal to start getSslSocketConnector
+ uriSSL = "cometds://127.0.0.1:" + portSSL +
"/service/test?baseResource=file:./target/test-classes/webapp&"
+ +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
+ }
+
+ @Override
+ @SuppressWarnings("deprecation")
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws URISyntaxException {
+ CometdComponent component = (CometdComponent)
context.getComponent("cometds");
+ component.setSslPassword(pwd);
+ component.setSslKeyPassword(pwd);
+ URI keyStoreUrl
+ =
CometdProducerConsumerInOutTest.class.getResource("/jsse/localhost.p12").toURI();
+ component.setSslKeystore(keyStoreUrl.getPath());
+
+ from(uri)
+ .setExchangePattern(ExchangePattern.InOut)
+ .process(exchange -> exchange.getOut().setBody("reply:
" + exchange.getIn().getBody()));
+
+ from(uriSSL)
+ .setExchangePattern(ExchangePattern.InOut)
+ .process(exchange -> exchange.getOut().setBody("reply
SSL: " + exchange.getIn().getBody()));
+ }
+ };
+ }
+}
diff --git
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveAuthenticatedManualTest.java
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveAuthenticatedManualTest.java
deleted file mode 100644
index ac2ad917e341..000000000000
---
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveAuthenticatedManualTest.java
+++ /dev/null
@@ -1,166 +0,0 @@
-/*
- * 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.cometd;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.camel.CamelContext;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.impl.DefaultCamelContext;
-import org.cometd.bayeux.Channel;
-import org.cometd.bayeux.Message;
-import org.cometd.bayeux.server.BayeuxServer;
-import org.cometd.bayeux.server.ServerMessage;
-import org.cometd.bayeux.server.ServerSession;
-import org.cometd.server.DefaultSecurityPolicy;
-import org.junit.jupiter.api.Disabled;
-
-@Disabled("Run this test manually")
-public class CometdProducerConsumerInteractiveAuthenticatedManualTest {
-
- private static final String URI =
"cometd://127.0.0.1:9091/channel/test?baseResource=file:./src/test/resources/webapp&"
- +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
-
- private static final String URIS =
"cometds://127.0.0.1:9443/channel/test?baseResource=file:./src/test/resources/webapp&"
- +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
-
- private CamelContext context;
-
- private String pwd = "changeit";
-
- public static void main(String[] args) throws Exception {
- CometdProducerConsumerInteractiveAuthenticatedManualTest me
- = new
CometdProducerConsumerInteractiveAuthenticatedManualTest();
- me.testCometdProducerConsumerInteractive();
- }
-
- public void testCometdProducerConsumerInteractive() throws Exception {
- context = new DefaultCamelContext();
- context.addRoutes(createRouteBuilder());
- context.start();
- }
-
- private RouteBuilder createRouteBuilder() {
- return new RouteBuilder() {
- public void configure() throws URISyntaxException {
- CometdComponent component = (CometdComponent)
context.getComponent("cometds");
- component.setSslPassword(pwd);
- component.setSslKeyPassword(pwd);
-
- CometdComponent component2 = (CometdComponent)
context.getComponent("cometd");
- BayeuxAuthenticator bayeuxAuthenticator = new
BayeuxAuthenticator();
- component2.setSecurityPolicy(bayeuxAuthenticator);
- component2.addExtension(bayeuxAuthenticator);
-
- URI keyStoreUrl
- =
CometdProducerConsumerInteractiveAuthenticatedManualTest.class.getResource("/jsse/localhost.p12")
- .toURI();
- component.setSslKeystore(keyStoreUrl.getPath());
-
- from("stream:in").to(URI).to(URIS);
- }
- };
- }
-
- /**
- * Custom SecurityPolicy, see
http://cometd.org/documentation/howtos/authentication for details
- */
- public static final class BayeuxAuthenticator extends DefaultSecurityPolicy
- implements BayeuxServer.Extension, ServerSession.RemovedListener {
-
- private String user = "changeit";
- private String pwd = "changeit";
-
- @Override
- public boolean canHandshake(BayeuxServer server, ServerSession
session, ServerMessage message) {
- if (session.isLocalSession()) {
- return true;
- }
-
- Map<String, Object> ext = message.getExt();
- if (ext == null) {
- return false;
- }
-
- @SuppressWarnings("unchecked")
- Map<String, Object> authentication = (Map<String, Object>)
ext.get("authentication");
- if (authentication == null) {
- return false;
- }
-
- Object authenticationData = verify(authentication);
- if (authenticationData == null) {
- return false;
- }
-
- session.addListener(this);
-
- return true;
- }
-
- private Object verify(Map<String, Object> authentication) {
- if (!user.equals(authentication.get("user"))) {
- return null;
- }
- if (!pwd.equals(authentication.get("credentials"))) {
- return null;
- }
- return "OK";
- }
-
- @Override
- public boolean sendMeta(ServerSession to, ServerMessage.Mutable
message) {
- if (Channel.META_HANDSHAKE.equals(message.getChannel())) {
- if (!message.isSuccessful()) {
- Map<String, Object> advice = message.getAdvice(true);
- advice.put(Message.RECONNECT_FIELD,
Message.RECONNECT_HANDSHAKE_VALUE);
-
- Map<String, Object> ext = message.getExt(true);
- Map<String, Object> authentication = new HashMap<>();
- ext.put("authentication", authentication);
- authentication.put("failed", true);
- authentication.put("failureReason", "invalid_credentials");
- }
- }
- return true;
- }
-
- @Override
- public void removed(ServerSession session, ServerMessage message,
boolean timeout) {
- // Remove authentication data
- }
-
- @Override
- public boolean rcv(ServerSession from, ServerMessage.Mutable message) {
- return true;
- }
-
- @Override
- public boolean rcvMeta(ServerSession from, ServerMessage.Mutable
message) {
- return true;
- }
-
- @Override
- public boolean send(ServerSession from, ServerSession to,
ServerMessage.Mutable message) {
- return true;
- }
- }
-
-}
diff --git
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveExtensionManualTest.java
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveExtensionManualTest.java
deleted file mode 100644
index 0cf1e1b3cff6..000000000000
---
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveExtensionManualTest.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * 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.cometd;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.Arrays;
-import java.util.HashSet;
-
-import org.apache.camel.CamelContext;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.impl.DefaultCamelContext;
-import org.cometd.bayeux.server.BayeuxServer;
-import org.cometd.bayeux.server.ServerMessage;
-import org.cometd.bayeux.server.ServerSession;
-import org.junit.jupiter.api.Disabled;
-
-@Disabled("Run this test manually")
-public class CometdProducerConsumerInteractiveExtensionManualTest {
-
- private static final String URI =
"cometd://127.0.0.1:9091/channel/test?baseResource=file:./src/test/resources/webapp&"
- +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
-
- private static final String URIS =
"cometds://127.0.0.1:9443/channel/test?baseResource=file:./src/test/resources/webapp&"
- +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
-
- private CamelContext context;
-
- private String pwd = "changeit";
-
- public static void main(String[] args) throws Exception {
- CometdProducerConsumerInteractiveExtensionManualTest me = new
CometdProducerConsumerInteractiveExtensionManualTest();
- me.testCometdProducerConsumerInteractive();
- }
-
- public void testCometdProducerConsumerInteractive() throws Exception {
- context = new DefaultCamelContext();
- context.addRoutes(createRouteBuilder());
- context.start();
- }
-
- private RouteBuilder createRouteBuilder() {
- return new RouteBuilder() {
- public void configure() throws URISyntaxException {
- CometdComponent component = (CometdComponent)
context.getComponent("cometds");
- component.setSslPassword(pwd);
- component.setSslKeyPassword(pwd);
-
- CometdComponent component2 = (CometdComponent)
context.getComponent("cometd");
- Censor bayeuxAuthenticator = new Censor();
- component2.addExtension(bayeuxAuthenticator);
-
- URI keyStoreUrl
- =
CometdProducerConsumerInteractiveExtensionManualTest.class.getResource("/jsse/localhost.p12").toURI();
- component.setSslKeystore(keyStoreUrl.getPath());
-
- from("stream:in").to(URI).to(URIS);
- }
- };
- }
-
- public static final class Censor implements BayeuxServer.Extension,
ServerSession.RemovedListener {
-
- private HashSet<String> forbidden = new HashSet<>(Arrays.asList("one",
"two"));
-
- @Override
- public void removed(ServerSession session, ServerMessage message,
boolean timeout) {
- // called on remove of client
- }
-
- @Override
- public boolean rcv(ServerSession from, ServerMessage.Mutable message) {
- return true;
- }
-
- @Override
- public boolean rcvMeta(ServerSession from, ServerMessage.Mutable
message) {
- return true;
- }
-
- @Override
- public boolean send(ServerSession from, ServerSession to,
ServerMessage.Mutable message) {
- Object data = message.getData();
- if (forbidden.contains(data)) {
- message.put("data", "***");
- }
- return true;
- }
-
- @Override
- public boolean sendMeta(ServerSession from, ServerMessage.Mutable
message) {
- return true;
- }
- }
-}
diff --git
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveManualTest.java
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveManualTest.java
deleted file mode 100644
index 29f2c52e6ea2..000000000000
---
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveManualTest.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.cometd;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-
-import org.apache.camel.CamelContext;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.impl.DefaultCamelContext;
-import org.junit.jupiter.api.Disabled;
-
-@Disabled("Run this test manually")
-public class CometdProducerConsumerInteractiveManualTest {
-
- private static final String URI =
"cometd://127.0.0.1:9091/channel/test?baseResource=file:./src/test/resources/webapp&"
- +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
-
- private static final String URIS =
"cometds://127.0.0.1:9443/channel/test?baseResource=file:./src/test/resources/webapp&"
- +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
-
- private CamelContext context;
-
- private String pwd = "changeit";
-
- public static void main(String[] args) throws Exception {
- CometdProducerConsumerInteractiveManualTest me = new
CometdProducerConsumerInteractiveManualTest();
- me.testCometdProducerConsumerInteractive();
- }
-
- public void testCometdProducerConsumerInteractive() throws Exception {
- context = new DefaultCamelContext();
- context.addRoutes(createRouteBuilder());
- context.start();
- }
-
- private RouteBuilder createRouteBuilder() {
- return new RouteBuilder() {
- public void configure() throws URISyntaxException {
- CometdComponent component = (CometdComponent)
context.getComponent("cometds");
- component.setSslPassword(pwd);
- component.setSslKeyPassword(pwd);
- URI keyStoreUrl =
CometdProducerConsumerInteractiveManualTest.class.getResource("/jsse/localhost.p12").toURI();
- component.setSslKeystore(keyStoreUrl.getPath());
-
- from("stream:in").to(URI).to(URIS);
- }
- };
- }
-
-}
diff --git
a/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveTest.java
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveTest.java
new file mode 100644
index 000000000000..ce3cab6da861
--- /dev/null
+++
b/components/camel-cometd/src/test/java/org/apache/camel/component/cometd/CometdProducerConsumerInteractiveTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.cometd;
+
+import java.net.URL;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+public class CometdProducerConsumerInteractiveTest extends CamelTestSupport {
+
+ private int port;
+ private String uri;
+
+ private int portSSL;
+ private String uriSSL;
+
+ private final String pwd = "changeit";
+
+ @Test
+ void testProducerToPlainAndSslEndpoints() throws Exception {
+ getMockEndpoint("mock:plain").expectedBodiesReceived("hello");
+ getMockEndpoint("mock:ssl").expectedBodiesReceived("hello");
+ template.sendBody("direct:input", "hello");
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Override
+ public void setupResources() {
+ port = AvailablePortFinder.getNextAvailable();
+ uri = "cometd://127.0.0.1:" + port +
"/channel/test?baseResource=file:./src/test/resources/webapp&"
+ +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
+
+ portSSL = AvailablePortFinder.getNextAvailable();
+ uriSSL = "cometds://127.0.0.1:" + portSSL +
"/channel/test?baseResource=file:./src/test/resources/webapp&"
+ +
"timeout=240000&interval=0&maxInterval=30000&multiFrameInterval=1500&jsonCommented=true&logLevel=2";
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ CometdComponent sslComponent = context.getComponent("cometds",
CometdComponent.class);
+ sslComponent.setSslPassword(pwd);
+ sslComponent.setSslKeyPassword(pwd);
+ URL keyStoreUrl =
CometdProducerConsumerInteractiveTest.class.getResource("/jsse/localhost.p12");
+ sslComponent.setSslKeystore(keyStoreUrl.getPath());
+
+ from("direct:input").to(uri).to(uriSSL);
+
+ from(uri).to("mock:plain");
+ from(uriSSL).to("mock:ssl");
+ }
+ };
+ }
+}
