You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "Croway (via GitHub)" <gi...@apache.org> on 2026/10/01 14:14:22 UTC
[PR] CAMEL-25197: cli-connector - use the Quarkus WebSockets Next client for the WebSocket transport [camel-quarkus]
Croway opened a new pull request, #9264:
URL: https://github.com/apache/camel-quarkus/pull/9264
_Claude Code on behalf of Croway_
Camel Quarkus side of [CAMEL-25197](https://issues.apache.org/jira/browse/CAMEL-25197). **Depends on apache/camel#27210** (the `CliWebSocketClient` SPI of `camel-cli-connector`): draft until it is merged and in a Camel 4.23.0-SNAPSHOT.
With `camel.cli.transport=websocket`, the Camel CLI connector dials out to a developer tool over a WebSocket. This makes it use the Quarkus WebSockets Next client when the application already has `quarkus-websockets-next`, and the JDK client otherwise. No dependency is added to the application for this.
## Changes
- `QuarkusCliWebSocketClient` (runtime): implements `CliWebSocketClient` with `BasicWebSocketConnector`. Callbacks are non-blocking on the event loop (the transport parses and sends on its own threads), messages up to 16 MB, the handshake rejections are mapped to their HTTP status, the url path and query are sent as given.
- `CliConnectorProcessor`: registers the client bean only when the `io.quarkus.websockets.next` capability is present; `quarkus-websockets-next` is an `<optional>` dependency of the runtime module. No new build-time option. `camel.cli.websocket.client=jdk` (Camel) forces the JDK client.
- New runtime option `quarkus.camel.cli.websocket.tls-configuration-name`: TLS configuration from the Quarkus TLS registry for `wss://`.
- Docs: `usage.adoc` for the WebSocket transport and the client selection.
Workarounds for WebSockets Next (Quarkus 3.40.1), both documented in the code:
- Its client back-pressure (new in 3.40) fetches one frame per message received, so a connection receiving messages in several frames (over 64 KB) eventually stops reading. The extension defaults `quarkus.websockets-next.client.max-pending-messages` to `0` (as before 3.40) when `quarkus-websockets-next` is present; this applies to the other WebSockets Next clients of the application too, unless the property is set (documented).
- When a connection is lost without a close frame (status 1006), it fails creating the `CloseReason` for `onClose` and skips releasing the connection: the client registers no `onClose`, the transport detects the closed connection on its next send or by its heartbeat.
- Frames stay at 64 KB (the Vert.x default): Vert.x uses the same size to split what it sends, and servers commonly refuse larger frames.
## Tests
- `integration-tests-jvm/cli-connector` (JDK client) and the new `integration-tests-jvm/cli-connector-websockets-next` (same application and tests via `build-helper`, with `quarkus-websockets-next`): an in-test WebSocket tool server, `hello` reports the expected client, actions, large messages both ways (4 MB actions, trace snapshots over 64 KB frames), the HTTP 401 handshake reported and recovered from, encoded query parameters sent as given. 4 tests each, green (3 consecutive runs).
- `extensions-jvm/cli-connector/deployment`: existing tests green.
- Soak test (240 s; 5 Camel Main apps, 1 Camel Quarkus app with the JDK client, 1 with the WebSockets Next client; the tool closing connections, a network blackhole, the tool down, the token rejected): PASS, all apps recovered from every disruption, no unexpected errors, all exited on the `stop` action.
Native: the extension stays JVM-only. The WebSockets Next client itself needs no reflection and supports native, so it would not block native support; the rest of `camel-cli-connector` was not assessed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25197: cli-connector - use the Vert.x WebSocket client for the WebSocket transport [camel-quarkus]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on code in PR #9264:
URL: https://github.com/apache/camel-quarkus/pull/9264#discussion_r4157919421
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusLocalCliConnector.java:
##########
@@ -26,6 +28,18 @@ public QuarkusLocalCliConnector(CliConnectorFactory cliConnectorFactory) {
super(cliConnectorFactory);
}
+ @Override
+ protected CliConnectorTransport createTransport(String name) {
+ // Camel refuses it with the Camel prod profile, which Camel Quarkus does not set outside dev mode
+ if ("websocket".equals(name) && ConfigUtils.isProfileActive("prod")) {
Review Comment:
_Claude Code on behalf of Croway_
Right, fixed in 7933b8c95b with `equalsIgnoreCase`, and `CliConnectorWebSocketProdProfileTest` now uses `camel.cli.transport=WebSocket`. With the case-sensitive check, that test no longer passes: the packaged application starts the transport and keeps running instead of exiting.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/VertxCliWebSocketClient.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeoutException;
+
+import io.quarkus.tls.TlsConfiguration;
+import io.quarkus.tls.TlsConfigurationRegistry;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import io.vertx.core.http.WebSocket;
+import io.vertx.core.http.WebSocketClient;
+import io.vertx.core.http.WebSocketClientOptions;
+import io.vertx.core.http.WebSocketConnectOptions;
+import io.vertx.core.http.WebSocketFrame;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport on Camel Quarkus, with the Vert.x
+ * WebSocket client of the application. {@code camel.cli.websocket.client=jdk} uses the JDK client instead.
+ * <p/>
+ * The handlers run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class VertxCliWebSocketClient implements CliWebSocketClient {
+
+ static final String NAME = "vertx";
+ private static final Logger LOG = Logger.getLogger(VertxCliWebSocketClient.class);
+ private static final int TIMEOUT = 10000;
+ // the frames sent: servers commonly refuse frames over 64 KB (Vert.x, Quarkus), and a char takes up to 3 bytes
+ private static final int FRAME_CHARS = 16 * 1024;
+ // the messages received, in bytes: up to 3 per char
+ private static final int MAX_MESSAGE_BYTES = 3 * MAX_MESSAGE_SIZE;
+
+ @Inject
+ Vertx vertx;
+
+ @Inject
+ TlsConfigurationRegistry tlsRegistry;
+
+ @Inject
+ CamelCliConnectorRunTimeConfig config;
+
+ private TlsConfiguration tls;
+
+ @PostConstruct
+ void resolveTlsConfiguration() {
+ Optional<String> name = config.websocket().tlsConfigurationName();
+ if (name.isPresent()) {
+ tls = tlsRegistry.get(name.get()).orElseThrow(() -> new IllegalStateException(
+ "No TLS configuration named " + name.get() + " (quarkus.camel.cli.websocket.tls-configuration-name)"));
+ }
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public CompletionStage<Channel> connect(URI url, Map<String, String> headers, Listener listener) {
+ boolean ssl = "wss".equals(url.getScheme().toLowerCase(Locale.ROOT));
+ WebSocketClientOptions options = new WebSocketClientOptions()
+ .setConnectTimeout(TIMEOUT)
+ // how long the tool has to close the connection once the close frame is sent (seconds): abort() closes
+ // the client, which closes the connection like this
+ .setClosingTimeout(1)
+ // a tool can send a whole message in a single frame
+ .setMaxFrameSize(MAX_MESSAGE_BYTES)
+ .setMaxMessageSize(MAX_MESSAGE_BYTES);
+ if (ssl && tls != null) {
+ configure(options, tls);
+ }
+ String path = url.getRawPath() == null || url.getRawPath().isEmpty() ? "/" : url.getRawPath();
+ WebSocketConnectOptions connect = new WebSocketConnectOptions()
+ .setHost(url.getHost())
+ .setPort(url.getPort() != -1 ? url.getPort() : ssl ? 443 : 80)
+ .setSsl(ssl)
+ // the path and query as given, encoded characters included
+ .setURI(url.getRawQuery() != null ? path + "?" + url.getRawQuery() : path);
+ headers.forEach(connect::addHeader);
+
+ // a client for each connection, closed with it
+ WebSocketClient client = vertx.createWebSocketClient(options);
+ CompletableFuture<Channel> answer = new CompletableFuture<>();
+ // the handshake: not WebSocketConnectOptions.setTimeout, which is also an idle timeout that can close the
+ // connection once open
+ long timer = vertx.setTimer(TIMEOUT, id -> {
+ if (answer.completeExceptionally(new TimeoutException("WebSocket handshake timed out after " + TIMEOUT + " ms"))) {
+ client.close();
+ }
+ });
+ client.connect(connect).onComplete(ar -> {
+ vertx.cancelTimer(timer);
+ if (answer.isDone()) {
+ // timed out
+ if (ar.succeeded()) {
+ ar.result().close();
+ }
+ return;
+ }
+ if (ar.failed()) {
+ client.close();
+ answer.completeExceptionally(translate(ar.cause()));
+ return;
+ }
+ WebSocket ws = ar.result();
+ ws.textMessageHandler(listener::onText);
+ ws.pongHandler(data -> listener.onPong());
+ ws.exceptionHandler(e -> {
+ LOG.debugf(e, "Camel CLI connector WebSocket error");
+ listener.onError(e);
+ });
+ ws.closeHandler(v -> {
+ client.close();
+ // no status when the connection is lost without a close frame
+ Short code = ws.closeStatusCode();
+ LOG.debugf("Camel CLI connector WebSocket closed: %s %s", code, ws.closeReason());
+ listener.onClose(code != null ? code : 1006, ws.closeReason());
+ });
+ answer.complete(new VertxChannel(ws, client));
+ });
+ return answer;
+ }
+
+ private static void configure(WebSocketClientOptions options, TlsConfiguration tls) {
+ options.setSsl(true);
+ if (tls.getTrustStoreOptions() != null) {
+ options.setTrustOptions(tls.getTrustStoreOptions());
+ }
+ if (tls.getKeyStoreOptions() != null) {
+ options.setKeyCertOptions(tls.getKeyStoreOptions());
+ }
+ options.setTrustAll(tls.isTrustAll());
+ if ("NONE".equals(tls.getHostnameVerificationAlgorithm().orElse(null))) {
+ options.setVerifyHost(false);
+ }
+ }
Review Comment:
_Claude Code on behalf of Croway_
Agreed, replaced with `TlsConfigUtils.configure(WebSocketClientOptions, TlsConfiguration)` in 7933b8c95b, so cipher suites, protocols, CRLs and the rest of the named configuration apply.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25197: cli-connector - use the Vert.x WebSocket client for the WebSocket transport [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton merged PR #9264:
URL: https://github.com/apache/camel-quarkus/pull/9264
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25197: cli-connector - use the Quarkus WebSockets Next client for the WebSocket transport [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton commented on code in PR #9264:
URL: https://github.com/apache/camel-quarkus/pull/9264#discussion_r4156811566
##########
extensions-jvm/cli-connector/deployment/src/main/java/org/apache/camel/quarkus/component/cli/connector/deployment/CliConnectorProcessor.java:
##########
@@ -51,6 +59,29 @@ CamelBeanBuildItem camelBeanBuildItem(CamelCliConnectorRecorder recorder) {
recorder.createCliConnectorFactory(Version.getVersion()));
}
+ /**
+ * The WebSocket transport uses the WebSockets Next client when the application has it, the JDK client otherwise.
+ */
+ @BuildStep
+ void webSocketClient(
+ Capabilities capabilities,
+ BuildProducer<AdditionalBeanBuildItem> additionalBeans,
+ BuildProducer<RunTimeConfigurationDefaultBuildItem> configDefaults) {
+ if (capabilities.isPresent(Capability.WEBSOCKETS_NEXT)) {
+ // the back-pressure of the WebSockets Next client (since Quarkus 3.40) fetches one frame per message
+ // received, so every message received in several frames (over 64 KB) leaves fewer frames to fetch, until
+ // the connection stops reading: no back-pressure by default, as before Quarkus 3.40
+ // TODO: Remove when https://github.com/quarkusio/quarkus/issues/57079 is fixed
+ configDefaults.produce(new RunTimeConfigurationDefaultBuildItem(
+ "quarkus.websockets-next.client.max-pending-messages", "0"));
Review Comment:
This turns off back-pressure for every WebSockets Next client in the application whenever both extensions are present, including when the connector is on the file transport or `camel.cli.websocket.client=jdk`, and in production. I see it is documented and tracked in quarkusio/quarkus#57079, but until that is fixed an application's own clients silently lose the default limit of 256 pending messages.
Would it be safer to leave the Quarkus default alone, and instead log a warning at startup when the WebSocket transport is in use with the WebSockets Next client and `max-pending-messages` is not `0`?
##########
extensions-jvm/cli-connector/runtime/src/main/doc/usage.adoc:
##########
@@ -0,0 +1,38 @@
+=== WebSocket transport
+
+With `camel.cli.transport=websocket`, the application dials out to a developer tool over a WebSocket instead of
+exchanging files with the Camel CLI (see the
+xref:{cq-camel-components}:others:cli-connector.adoc[CLI Connector] documentation for the protocol and the options).
+
+[source,properties]
+----
+camel.cli.transport = websocket
+camel.cli.websocket.url = ws://127.0.0.1:8000/connect?executionId=run-1
+----
+
+The tool gets full control of the application: the transport refuses to start with the `prod` profile
+(`camel.main.profile=prod`).
Review Comment:
This guard checks the Camel profile, and Camel Quarkus only sets that in dev mode (`CamelMainRecorder.customizeDevModeCamelMain`). A packaged application running under the Quarkus `prod` profile has no Camel profile unless `camel.main.profile` is set explicitly, so the transport still starts if `camel.cli.transport=websocket` is left unscoped in `application.properties`.
Could `QuarkusLocalCliConnector` refuse the WebSocket transport in `LaunchMode.NORMAL`? Otherwise the docs should say the guard only applies when `camel.main.profile=prod` is set explicitly, since readers will assume the Quarkus `prod` profile is covered.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusCliWebSocketClient.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.concurrent.CompletionStage;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.InstanceHandle;
+import io.quarkus.websockets.next.BasicWebSocketConnector;
+import io.quarkus.websockets.next.BasicWebSocketConnector.ExecutionModel;
+import io.quarkus.websockets.next.CloseReason;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport, with the Quarkus WebSockets Next
+ * client: registered as a bean only when the application has {@code quarkus-websockets-next}.
+ * <p/>
+ * The callbacks run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class QuarkusCliWebSocketClient implements CliWebSocketClient {
+
+ static final String NAME = "quarkus-websockets-next";
+ private static final Logger LOG = Logger.getLogger(QuarkusCliWebSocketClient.class);
+ private static final String MAX_MESSAGE_SIZE_KEY = "quarkus.websockets-next.client.max-message-size";
+ private static final int GOING_AWAY = 1001;
+
+ @Inject
+ CamelCliConnectorRunTimeConfig config;
+
+ @PostConstruct
+ void checkMaxMessageSize() {
+ // set globally, it replaces the size of every connector (see customizeOptions)
+ OptionalInt size = ConfigProvider.getConfig().getOptionalValue(MAX_MESSAGE_SIZE_KEY, Integer.class)
+ .map(OptionalInt::of).orElse(OptionalInt.empty());
+ if (size.isPresent() && size.getAsInt() < MAX_MESSAGE_SIZE) {
+ LOG.warnf("%s=%d: the Camel CLI connector cannot receive actions larger than that (up to %d supported)",
+ MAX_MESSAGE_SIZE_KEY, size.getAsInt(), MAX_MESSAGE_SIZE);
+ }
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public CompletionStage<Channel> connect(URI url, Map<String, String> headers, Listener listener) {
+ // a new connector for each connection, released when it is closed
+ InstanceHandle<BasicWebSocketConnector> handle = Arc.container().instance(BasicWebSocketConnector.class);
+ String path = url.getRawPath() != null ? url.getRawPath() : "";
+ int last = path.lastIndexOf('/') + 1;
+ BasicWebSocketConnector connector = handle.get()
+ .baseUri(baseUri(url, path.substring(0, last)))
+ // the last segment as the path: WebSockets Next adds a '/' after the base uri path otherwise
+ // TODO: Remove when https://github.com/quarkusio/quarkus/issues/57081 is fixed
+ .path(path.substring(last))
+ .executionModel(ExecutionModel.NON_BLOCKING)
+ // not the frame size: Vert.x also splits the messages it sends into frames of that size, and servers
+ // commonly refuse frames over 64 KB
Review Comment:
Understood that raising the frame size also changes how outgoing messages are split. The consequence, though, is that a tool sending an unfragmented message over 64 KB gets its connection dropped with this client, while the same tool works with the JDK client. So adding `quarkus-websockets-next` to an application can break a working setup. I'm not sure that most WebSocket libraries fragment by default; several send one frame per message. The tests can't catch this, because the tool server is Vert.x, which fragments at 64 KB.
Is that limitation acceptable for the tools this targets?
Minor, related: this limit is counted in bytes while `CliWebSocketClient.MAX_MESSAGE_SIZE` is defined in chars, so a non-ASCII message under the limit can still be rejected.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusCliWebSocketClient.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.concurrent.CompletionStage;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.InstanceHandle;
+import io.quarkus.websockets.next.BasicWebSocketConnector;
+import io.quarkus.websockets.next.BasicWebSocketConnector.ExecutionModel;
+import io.quarkus.websockets.next.CloseReason;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport, with the Quarkus WebSockets Next
+ * client: registered as a bean only when the application has {@code quarkus-websockets-next}.
+ * <p/>
+ * The callbacks run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class QuarkusCliWebSocketClient implements CliWebSocketClient {
Review Comment:
An open question, not a change request: did you consider a plain Vert.x `WebSocketClient` from the injected `Vertx` instead of WebSockets Next?
This client has to work around WebSockets Next in several places: the global back-pressure default, no `onClose`, the path split and `%25` re-escaping in `baseUri()`, and an `abort()` that cannot drop the socket. Three of those now have upstream issues, so it may be fine to wait for them. The Vert.x client would give per-client options, a close handler with the status code and the raw request URI, with TLS still coming from the TLS registry. Vert.x is already on the classpath through `camel-quarkus-console` (`quarkus-vertx-http`), so it would also remove the optional dependency, the capability check and the by-name bean.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusCliWebSocketClient.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.concurrent.CompletionStage;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.InstanceHandle;
+import io.quarkus.websockets.next.BasicWebSocketConnector;
+import io.quarkus.websockets.next.BasicWebSocketConnector.ExecutionModel;
+import io.quarkus.websockets.next.CloseReason;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport, with the Quarkus WebSockets Next
+ * client: registered as a bean only when the application has {@code quarkus-websockets-next}.
+ * <p/>
+ * The callbacks run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class QuarkusCliWebSocketClient implements CliWebSocketClient {
+
+ static final String NAME = "quarkus-websockets-next";
+ private static final Logger LOG = Logger.getLogger(QuarkusCliWebSocketClient.class);
+ private static final String MAX_MESSAGE_SIZE_KEY = "quarkus.websockets-next.client.max-message-size";
+ private static final int GOING_AWAY = 1001;
+
+ @Inject
+ CamelCliConnectorRunTimeConfig config;
+
+ @PostConstruct
+ void checkMaxMessageSize() {
+ // set globally, it replaces the size of every connector (see customizeOptions)
+ OptionalInt size = ConfigProvider.getConfig().getOptionalValue(MAX_MESSAGE_SIZE_KEY, Integer.class)
+ .map(OptionalInt::of).orElse(OptionalInt.empty());
+ if (size.isPresent() && size.getAsInt() < MAX_MESSAGE_SIZE) {
+ LOG.warnf("%s=%d: the Camel CLI connector cannot receive actions larger than that (up to %d supported)",
+ MAX_MESSAGE_SIZE_KEY, size.getAsInt(), MAX_MESSAGE_SIZE);
+ }
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public CompletionStage<Channel> connect(URI url, Map<String, String> headers, Listener listener) {
+ // a new connector for each connection, released when it is closed
+ InstanceHandle<BasicWebSocketConnector> handle = Arc.container().instance(BasicWebSocketConnector.class);
+ String path = url.getRawPath() != null ? url.getRawPath() : "";
+ int last = path.lastIndexOf('/') + 1;
+ BasicWebSocketConnector connector = handle.get()
+ .baseUri(baseUri(url, path.substring(0, last)))
+ // the last segment as the path: WebSockets Next adds a '/' after the base uri path otherwise
+ // TODO: Remove when https://github.com/quarkusio/quarkus/issues/57081 is fixed
+ .path(path.substring(last))
+ .executionModel(ExecutionModel.NON_BLOCKING)
+ // not the frame size: Vert.x also splits the messages it sends into frames of that size, and servers
+ // commonly refuse frames over 64 KB
+ .customizeOptions((connect, client) -> client.setMaxMessageSize(MAX_MESSAGE_SIZE))
Review Comment:
No connect or handshake timeout is set here, whereas the JDK client uses 10 s for both. If the tool accepts the TCP connection but never answers the upgrade (paused in a debugger, or a port-forward with nothing behind it), I don't think `connect()` ever completes. As far as I can tell the transport only reconnects when that stage completes, so the connector would stay stuck until the application restarts.
Suggest setting `connect.setTimeout(...)` and `client.setConnectTimeout(...)` here to match the JDK client.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusCliWebSocketClient.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.concurrent.CompletionStage;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.InstanceHandle;
+import io.quarkus.websockets.next.BasicWebSocketConnector;
+import io.quarkus.websockets.next.BasicWebSocketConnector.ExecutionModel;
+import io.quarkus.websockets.next.CloseReason;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport, with the Quarkus WebSockets Next
+ * client: registered as a bean only when the application has {@code quarkus-websockets-next}.
+ * <p/>
+ * The callbacks run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class QuarkusCliWebSocketClient implements CliWebSocketClient {
+
+ static final String NAME = "quarkus-websockets-next";
+ private static final Logger LOG = Logger.getLogger(QuarkusCliWebSocketClient.class);
+ private static final String MAX_MESSAGE_SIZE_KEY = "quarkus.websockets-next.client.max-message-size";
+ private static final int GOING_AWAY = 1001;
+
+ @Inject
+ CamelCliConnectorRunTimeConfig config;
+
+ @PostConstruct
+ void checkMaxMessageSize() {
+ // set globally, it replaces the size of every connector (see customizeOptions)
+ OptionalInt size = ConfigProvider.getConfig().getOptionalValue(MAX_MESSAGE_SIZE_KEY, Integer.class)
+ .map(OptionalInt::of).orElse(OptionalInt.empty());
Review Comment:
Nit: the `OptionalInt` conversion isn't needed. `getOptionalValue(MAX_MESSAGE_SIZE_KEY, Integer.class).filter(s -> s < MAX_MESSAGE_SIZE).ifPresent(s -> LOG.warnf(...))` does the same.
##########
integration-tests-jvm/cli-connector-websockets-next/pom.xml:
##########
@@ -0,0 +1,184 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-build-parent-it</artifactId>
+ <version>4.0.0-SNAPSHOT</version>
+ <relativePath>../../poms/build-parent-it/pom.xml</relativePath>
+ </parent>
+
+ <artifactId>camel-quarkus-integration-test-cli-connector-websockets-next</artifactId>
+ <name>Camel Quarkus :: Integration Tests :: CLI Connector :: WebSockets Next</name>
+ <description>Integration tests for Camel Quarkus CLI Connector extension</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-cli-connector</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-websockets-next</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-management</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-direct</artifactId>
+ </dependency>
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.rest-assured</groupId>
+ <artifactId>rest-assured</artifactId>
+ <scope>test</scope>
+ </dependency>
Review Comment:
Nit: `rest-assured` looks unused now that `CliConnectorResource` and the RestAssured test are gone. Same in `integration-tests-jvm/cli-connector/pom.xml`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25197: cli-connector - use the Vert.x WebSocket client for the WebSocket transport [camel-quarkus]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on code in PR #9264:
URL: https://github.com/apache/camel-quarkus/pull/9264#discussion_r4157525426
##########
extensions-jvm/cli-connector/deployment/src/main/java/org/apache/camel/quarkus/component/cli/connector/deployment/CliConnectorProcessor.java:
##########
@@ -51,6 +59,29 @@ CamelBeanBuildItem camelBeanBuildItem(CamelCliConnectorRecorder recorder) {
recorder.createCliConnectorFactory(Version.getVersion()));
}
+ /**
+ * The WebSocket transport uses the WebSockets Next client when the application has it, the JDK client otherwise.
+ */
+ @BuildStep
+ void webSocketClient(
+ Capabilities capabilities,
+ BuildProducer<AdditionalBeanBuildItem> additionalBeans,
+ BuildProducer<RunTimeConfigurationDefaultBuildItem> configDefaults) {
+ if (capabilities.isPresent(Capability.WEBSOCKETS_NEXT)) {
+ // the back-pressure of the WebSockets Next client (since Quarkus 3.40) fetches one frame per message
+ // received, so every message received in several frames (over 64 KB) leaves fewer frames to fetch, until
+ // the connection stops reading: no back-pressure by default, as before Quarkus 3.40
+ // TODO: Remove when https://github.com/quarkusio/quarkus/issues/57079 is fixed
+ configDefaults.produce(new RunTimeConfigurationDefaultBuildItem(
+ "quarkus.websockets-next.client.max-pending-messages", "0"));
Review Comment:
_Claude Code on behalf of Croway_
Agreed. Fixed in fa4f77ac3c by dropping WebSockets Next (see the thread on the Vert.x client): the extension no longer touches any `quarkus.websockets-next` setting, so the application's own clients keep their back-pressure.
##########
extensions-jvm/cli-connector/runtime/src/main/doc/usage.adoc:
##########
@@ -0,0 +1,38 @@
+=== WebSocket transport
+
+With `camel.cli.transport=websocket`, the application dials out to a developer tool over a WebSocket instead of
+exchanging files with the Camel CLI (see the
+xref:{cq-camel-components}:others:cli-connector.adoc[CLI Connector] documentation for the protocol and the options).
+
+[source,properties]
+----
+camel.cli.transport = websocket
+camel.cli.websocket.url = ws://127.0.0.1:8000/connect?executionId=run-1
+----
+
+The tool gets full control of the application: the transport refuses to start with the `prod` profile
+(`camel.main.profile=prod`).
Review Comment:
_Claude Code on behalf of Croway_
Right, and that was already the case with the JDK client: `CamelMainDevModeProcessor` only sets the Camel profile in dev mode. Fixed in fa4f77ac3c: `QuarkusLocalCliConnector.createTransport()` refuses `websocket` when the Quarkus `prod` profile is active, with a message like Camel's. `CliConnectorWebSocketProdProfileTest` (`QuarkusProdModeTest`) checks that a packaged application refuses it and exits. The docs now scope the example to `%dev` and say that a packaged application, e.g. on a remote dev cluster, needs another profile. I went with the profile rather than `LaunchMode.NORMAL` so that this remote dev cluster case stays possible.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusCliWebSocketClient.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.concurrent.CompletionStage;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.InstanceHandle;
+import io.quarkus.websockets.next.BasicWebSocketConnector;
+import io.quarkus.websockets.next.BasicWebSocketConnector.ExecutionModel;
+import io.quarkus.websockets.next.CloseReason;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport, with the Quarkus WebSockets Next
+ * client: registered as a bean only when the application has {@code quarkus-websockets-next}.
+ * <p/>
+ * The callbacks run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class QuarkusCliWebSocketClient implements CliWebSocketClient {
+
+ static final String NAME = "quarkus-websockets-next";
+ private static final Logger LOG = Logger.getLogger(QuarkusCliWebSocketClient.class);
+ private static final String MAX_MESSAGE_SIZE_KEY = "quarkus.websockets-next.client.max-message-size";
+ private static final int GOING_AWAY = 1001;
+
+ @Inject
+ CamelCliConnectorRunTimeConfig config;
+
+ @PostConstruct
+ void checkMaxMessageSize() {
+ // set globally, it replaces the size of every connector (see customizeOptions)
+ OptionalInt size = ConfigProvider.getConfig().getOptionalValue(MAX_MESSAGE_SIZE_KEY, Integer.class)
+ .map(OptionalInt::of).orElse(OptionalInt.empty());
+ if (size.isPresent() && size.getAsInt() < MAX_MESSAGE_SIZE) {
+ LOG.warnf("%s=%d: the Camel CLI connector cannot receive actions larger than that (up to %d supported)",
+ MAX_MESSAGE_SIZE_KEY, size.getAsInt(), MAX_MESSAGE_SIZE);
+ }
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public CompletionStage<Channel> connect(URI url, Map<String, String> headers, Listener listener) {
+ // a new connector for each connection, released when it is closed
+ InstanceHandle<BasicWebSocketConnector> handle = Arc.container().instance(BasicWebSocketConnector.class);
+ String path = url.getRawPath() != null ? url.getRawPath() : "";
+ int last = path.lastIndexOf('/') + 1;
+ BasicWebSocketConnector connector = handle.get()
+ .baseUri(baseUri(url, path.substring(0, last)))
+ // the last segment as the path: WebSockets Next adds a '/' after the base uri path otherwise
+ // TODO: Remove when https://github.com/quarkusio/quarkus/issues/57081 is fixed
+ .path(path.substring(last))
+ .executionModel(ExecutionModel.NON_BLOCKING)
+ // not the frame size: Vert.x also splits the messages it sends into frames of that size, and servers
+ // commonly refuse frames over 64 KB
+ .customizeOptions((connect, client) -> client.setMaxMessageSize(MAX_MESSAGE_SIZE))
Review Comment:
_Claude Code on behalf of Croway_
Good catch, it would have stayed stuck. Fixed in fa4f77ac3c: 10 s TCP connect timeout in the client options and a 10 s handshake deadline (a Vert.x timer, closing the client), as the JDK client. Not `WebSocketConnectOptions.setTimeout`: in Vert.x 4.5 it also sets an idle timeout. `reconnectsWhenTheToolDoesNotAnswerTheHandshake` uses a tool that never answers the upgrade, and `staysConnected` checks that an open connection outlives the timeouts.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusCliWebSocketClient.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.concurrent.CompletionStage;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.InstanceHandle;
+import io.quarkus.websockets.next.BasicWebSocketConnector;
+import io.quarkus.websockets.next.BasicWebSocketConnector.ExecutionModel;
+import io.quarkus.websockets.next.CloseReason;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport, with the Quarkus WebSockets Next
+ * client: registered as a bean only when the application has {@code quarkus-websockets-next}.
+ * <p/>
+ * The callbacks run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class QuarkusCliWebSocketClient implements CliWebSocketClient {
+
+ static final String NAME = "quarkus-websockets-next";
+ private static final Logger LOG = Logger.getLogger(QuarkusCliWebSocketClient.class);
+ private static final String MAX_MESSAGE_SIZE_KEY = "quarkus.websockets-next.client.max-message-size";
+ private static final int GOING_AWAY = 1001;
+
+ @Inject
+ CamelCliConnectorRunTimeConfig config;
+
+ @PostConstruct
+ void checkMaxMessageSize() {
+ // set globally, it replaces the size of every connector (see customizeOptions)
+ OptionalInt size = ConfigProvider.getConfig().getOptionalValue(MAX_MESSAGE_SIZE_KEY, Integer.class)
+ .map(OptionalInt::of).orElse(OptionalInt.empty());
+ if (size.isPresent() && size.getAsInt() < MAX_MESSAGE_SIZE) {
+ LOG.warnf("%s=%d: the Camel CLI connector cannot receive actions larger than that (up to %d supported)",
+ MAX_MESSAGE_SIZE_KEY, size.getAsInt(), MAX_MESSAGE_SIZE);
+ }
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public CompletionStage<Channel> connect(URI url, Map<String, String> headers, Listener listener) {
+ // a new connector for each connection, released when it is closed
+ InstanceHandle<BasicWebSocketConnector> handle = Arc.container().instance(BasicWebSocketConnector.class);
+ String path = url.getRawPath() != null ? url.getRawPath() : "";
+ int last = path.lastIndexOf('/') + 1;
+ BasicWebSocketConnector connector = handle.get()
+ .baseUri(baseUri(url, path.substring(0, last)))
+ // the last segment as the path: WebSockets Next adds a '/' after the base uri path otherwise
+ // TODO: Remove when https://github.com/quarkusio/quarkus/issues/57081 is fixed
+ .path(path.substring(last))
+ .executionModel(ExecutionModel.NON_BLOCKING)
+ // not the frame size: Vert.x also splits the messages it sends into frames of that size, and servers
+ // commonly refuse frames over 64 KB
Review Comment:
_Claude Code on behalf of Croway_
Not acceptable, you're right: tools sending one frame per message would break. With the plain Vert.x client the two directions can differ, fixed in fa4f77ac3c: incoming frames and messages are accepted up to 3 x 16 M bytes (so a 16 M chars message fits whatever its characters, which also addresses bytes vs chars), and outgoing messages are split by the client into frames of 16 K chars (at most 48 KB in UTF-8, never splitting a surrogate pair) written with `writeFrame`. `receivesLargeActionsInASingleFrame` sends a 1 MB action in a single frame.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusCliWebSocketClient.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.concurrent.CompletionStage;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.InstanceHandle;
+import io.quarkus.websockets.next.BasicWebSocketConnector;
+import io.quarkus.websockets.next.BasicWebSocketConnector.ExecutionModel;
+import io.quarkus.websockets.next.CloseReason;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport, with the Quarkus WebSockets Next
+ * client: registered as a bean only when the application has {@code quarkus-websockets-next}.
+ * <p/>
+ * The callbacks run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class QuarkusCliWebSocketClient implements CliWebSocketClient {
Review Comment:
_Claude Code on behalf of Croway_
Yes, thanks for the suggestion: done in fa4f77ac3c, and it is now the default client on Camel Quarkus (`camel.cli.websocket.client=jdk` still selects the JDK client). `VertxCliWebSocketClient` uses a `WebSocketClient` per connection from the application's `Vertx`: per-client options, the close status code, the raw request URI, and `abort()` closes the client (with a 1 s closing timeout, so the socket is gone right away). TLS comes from the TLS registry (`quarkus.camel.cli.websocket.tls-configuration-name`), otherwise the JVM default trust store. The optional dependency, the capability check and the by-name bean are gone. The three Quarkus issues stay open upstream but no longer affect this extension. Soak test (240 s, with disruptions) green with the Vert.x client.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusCliWebSocketClient.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.concurrent.CompletionStage;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.InstanceHandle;
+import io.quarkus.websockets.next.BasicWebSocketConnector;
+import io.quarkus.websockets.next.BasicWebSocketConnector.ExecutionModel;
+import io.quarkus.websockets.next.CloseReason;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport, with the Quarkus WebSockets Next
+ * client: registered as a bean only when the application has {@code quarkus-websockets-next}.
+ * <p/>
+ * The callbacks run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class QuarkusCliWebSocketClient implements CliWebSocketClient {
+
+ static final String NAME = "quarkus-websockets-next";
+ private static final Logger LOG = Logger.getLogger(QuarkusCliWebSocketClient.class);
+ private static final String MAX_MESSAGE_SIZE_KEY = "quarkus.websockets-next.client.max-message-size";
+ private static final int GOING_AWAY = 1001;
+
+ @Inject
+ CamelCliConnectorRunTimeConfig config;
+
+ @PostConstruct
+ void checkMaxMessageSize() {
+ // set globally, it replaces the size of every connector (see customizeOptions)
+ OptionalInt size = ConfigProvider.getConfig().getOptionalValue(MAX_MESSAGE_SIZE_KEY, Integer.class)
+ .map(OptionalInt::of).orElse(OptionalInt.empty());
Review Comment:
_Claude Code on behalf of Croway_
Gone with the WebSockets Next client in fa4f77ac3c.
##########
integration-tests-jvm/cli-connector-websockets-next/pom.xml:
##########
@@ -0,0 +1,184 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-build-parent-it</artifactId>
+ <version>4.0.0-SNAPSHOT</version>
+ <relativePath>../../poms/build-parent-it/pom.xml</relativePath>
+ </parent>
+
+ <artifactId>camel-quarkus-integration-test-cli-connector-websockets-next</artifactId>
+ <name>Camel Quarkus :: Integration Tests :: CLI Connector :: WebSockets Next</name>
+ <description>Integration tests for Camel Quarkus CLI Connector extension</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-cli-connector</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-websockets-next</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-management</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-direct</artifactId>
+ </dependency>
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.rest-assured</groupId>
+ <artifactId>rest-assured</artifactId>
+ <scope>test</scope>
+ </dependency>
Review Comment:
_Claude Code on behalf of Croway_
Removed in fa4f77ac3c (the `cli-connector-websockets-next` module is gone too).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25197: cli-connector - use the Vert.x WebSocket client for the WebSocket transport [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton commented on code in PR #9264:
URL: https://github.com/apache/camel-quarkus/pull/9264#discussion_r4157724019
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/QuarkusLocalCliConnector.java:
##########
@@ -26,6 +28,18 @@ public QuarkusLocalCliConnector(CliConnectorFactory cliConnectorFactory) {
super(cliConnectorFactory);
}
+ @Override
+ protected CliConnectorTransport createTransport(String name) {
+ // Camel refuses it with the Camel prod profile, which Camel Quarkus does not set outside dev mode
+ if ("websocket".equals(name) && ConfigUtils.isProfileActive("prod")) {
Review Comment:
This check is case-sensitive, but `LocalCliConnector.createTransport` matches with `"websocket".equalsIgnoreCase(name)`. With `camel.cli.transport=WebSocket` under the `prod` profile the guard is skipped and `super.createTransport(name)` still returns the WebSocket transport.
Suggest `equalsIgnoreCase` here too, and a mixed-case value in `CliConnectorWebSocketProdProfileTest`.
##########
extensions-jvm/cli-connector/runtime/src/main/java/org/apache/camel/quarkus/component/cli/connector/VertxCliWebSocketClient.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.quarkus.component.cli.connector;
+
+import java.net.URI;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeoutException;
+
+import io.quarkus.tls.TlsConfiguration;
+import io.quarkus.tls.TlsConfigurationRegistry;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.UpgradeRejectedException;
+import io.vertx.core.http.WebSocket;
+import io.vertx.core.http.WebSocketClient;
+import io.vertx.core.http.WebSocketClientOptions;
+import io.vertx.core.http.WebSocketConnectOptions;
+import io.vertx.core.http.WebSocketFrame;
+import jakarta.annotation.PostConstruct;
+import jakarta.inject.Inject;
+import org.apache.camel.cli.connector.CliWebSocketClient;
+import org.apache.camel.cli.connector.CliWebSocketHandshakeException;
+import org.jboss.logging.Logger;
+
+/**
+ * The {@link CliWebSocketClient} of the Camel CLI connector WebSocket transport on Camel Quarkus, with the Vert.x
+ * WebSocket client of the application. {@code camel.cli.websocket.client=jdk} uses the JDK client instead.
+ * <p/>
+ * The handlers run on the Vert.x event loop: they only hand over to the transport, which parses and sends on its own
+ * threads.
+ */
+public class VertxCliWebSocketClient implements CliWebSocketClient {
+
+ static final String NAME = "vertx";
+ private static final Logger LOG = Logger.getLogger(VertxCliWebSocketClient.class);
+ private static final int TIMEOUT = 10000;
+ // the frames sent: servers commonly refuse frames over 64 KB (Vert.x, Quarkus), and a char takes up to 3 bytes
+ private static final int FRAME_CHARS = 16 * 1024;
+ // the messages received, in bytes: up to 3 per char
+ private static final int MAX_MESSAGE_BYTES = 3 * MAX_MESSAGE_SIZE;
+
+ @Inject
+ Vertx vertx;
+
+ @Inject
+ TlsConfigurationRegistry tlsRegistry;
+
+ @Inject
+ CamelCliConnectorRunTimeConfig config;
+
+ private TlsConfiguration tls;
+
+ @PostConstruct
+ void resolveTlsConfiguration() {
+ Optional<String> name = config.websocket().tlsConfigurationName();
+ if (name.isPresent()) {
+ tls = tlsRegistry.get(name.get()).orElseThrow(() -> new IllegalStateException(
+ "No TLS configuration named " + name.get() + " (quarkus.camel.cli.websocket.tls-configuration-name)"));
+ }
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public CompletionStage<Channel> connect(URI url, Map<String, String> headers, Listener listener) {
+ boolean ssl = "wss".equals(url.getScheme().toLowerCase(Locale.ROOT));
+ WebSocketClientOptions options = new WebSocketClientOptions()
+ .setConnectTimeout(TIMEOUT)
+ // how long the tool has to close the connection once the close frame is sent (seconds): abort() closes
+ // the client, which closes the connection like this
+ .setClosingTimeout(1)
+ // a tool can send a whole message in a single frame
+ .setMaxFrameSize(MAX_MESSAGE_BYTES)
+ .setMaxMessageSize(MAX_MESSAGE_BYTES);
+ if (ssl && tls != null) {
+ configure(options, tls);
+ }
+ String path = url.getRawPath() == null || url.getRawPath().isEmpty() ? "/" : url.getRawPath();
+ WebSocketConnectOptions connect = new WebSocketConnectOptions()
+ .setHost(url.getHost())
+ .setPort(url.getPort() != -1 ? url.getPort() : ssl ? 443 : 80)
+ .setSsl(ssl)
+ // the path and query as given, encoded characters included
+ .setURI(url.getRawQuery() != null ? path + "?" + url.getRawQuery() : path);
+ headers.forEach(connect::addHeader);
+
+ // a client for each connection, closed with it
+ WebSocketClient client = vertx.createWebSocketClient(options);
+ CompletableFuture<Channel> answer = new CompletableFuture<>();
+ // the handshake: not WebSocketConnectOptions.setTimeout, which is also an idle timeout that can close the
+ // connection once open
+ long timer = vertx.setTimer(TIMEOUT, id -> {
+ if (answer.completeExceptionally(new TimeoutException("WebSocket handshake timed out after " + TIMEOUT + " ms"))) {
+ client.close();
+ }
+ });
+ client.connect(connect).onComplete(ar -> {
+ vertx.cancelTimer(timer);
+ if (answer.isDone()) {
+ // timed out
+ if (ar.succeeded()) {
+ ar.result().close();
+ }
+ return;
+ }
+ if (ar.failed()) {
+ client.close();
+ answer.completeExceptionally(translate(ar.cause()));
+ return;
+ }
+ WebSocket ws = ar.result();
+ ws.textMessageHandler(listener::onText);
+ ws.pongHandler(data -> listener.onPong());
+ ws.exceptionHandler(e -> {
+ LOG.debugf(e, "Camel CLI connector WebSocket error");
+ listener.onError(e);
+ });
+ ws.closeHandler(v -> {
+ client.close();
+ // no status when the connection is lost without a close frame
+ Short code = ws.closeStatusCode();
+ LOG.debugf("Camel CLI connector WebSocket closed: %s %s", code, ws.closeReason());
+ listener.onClose(code != null ? code : 1006, ws.closeReason());
+ });
+ answer.complete(new VertxChannel(ws, client));
+ });
+ return answer;
+ }
+
+ private static void configure(WebSocketClientOptions options, TlsConfiguration tls) {
+ options.setSsl(true);
+ if (tls.getTrustStoreOptions() != null) {
+ options.setTrustOptions(tls.getTrustStoreOptions());
+ }
+ if (tls.getKeyStoreOptions() != null) {
+ options.setKeyCertOptions(tls.getKeyStoreOptions());
+ }
+ options.setTrustAll(tls.isTrustAll());
+ if ("NONE".equals(tls.getHostnameVerificationAlgorithm().orElse(null))) {
+ options.setVerifyHost(false);
+ }
+ }
Review Comment:
This copies only the trust store, key store, trust-all and the `NONE` hostname check, so cipher suites, enabled protocols and CRLs set on the named TLS configuration are silently ignored.
Quarkus already has `io.quarkus.tls.runtime.config.TlsConfigUtils.configure(WebSocketClientOptions, TlsConfiguration)`, which applies all of those. Could this method be replaced with that call?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org