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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
