gnodet-bot commented on code in PR #29421: URL: https://github.com/apache/camel/pull/29421#discussion_r4506947638
########## components/camel-vertx/camel-vertx-mqtt/src/main/java/org/apache/camel/component/vertx/mqtt/VertxMqttProperties.java: ########## @@ -0,0 +1,119 @@ +/* + * 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.vertx.mqtt; + +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.Map; + +import io.netty.handler.codec.mqtt.MqttProperties; +import io.netty.handler.codec.mqtt.MqttProperties.BinaryProperty; +import io.netty.handler.codec.mqtt.MqttProperties.IntegerProperty; +import io.netty.handler.codec.mqtt.MqttProperties.MqttProperty; +import io.netty.handler.codec.mqtt.MqttProperties.StringPair; +import io.netty.handler.codec.mqtt.MqttProperties.StringProperty; +import io.netty.handler.codec.mqtt.MqttProperties.UserProperty; +import org.apache.camel.Message; + +/** + * The MQTT 5 properties of a PUBLISH packet and the headers of a Camel message that carry them. + */ +final class VertxMqttProperties { + + private VertxMqttProperties() { + } + + /** + * @return the properties the headers of a message set, empty if they set none + */ + static MqttProperties fromHeaders(final Message message) { + final MqttProperties properties = new MqttProperties(); + final Map<?, ?> user = message.getHeader(VertxMqttConstants.USER_PROPERTIES, Map.class); + if (user != null) { + user.forEach((key, value) -> { + if (key != null && value != null) { + properties.add(new UserProperty(key.toString(), value.toString())); + } + }); + } + final String contentType = message.getHeader(VertxMqttConstants.CONTENT_TYPE, String.class); + if (contentType != null) { + properties.add(new StringProperty(MqttProperties.CONTENT_TYPE, contentType)); + } + final String responseTopic = message.getHeader(VertxMqttConstants.RESPONSE_TOPIC, String.class); + if (responseTopic != null) { + properties.add(new StringProperty(MqttProperties.RESPONSE_TOPIC, responseTopic)); + } + final byte[] correlationData = message.getHeader(VertxMqttConstants.CORRELATION_DATA, byte[].class); + if (correlationData != null) { + properties.add(new BinaryProperty(MqttProperties.CORRELATION_DATA, correlationData)); + } + final Long expiry = message.getHeader(VertxMqttConstants.MESSAGE_EXPIRY_INTERVAL, Long.class); + if (expiry != null) { + if (expiry < 0 || expiry > 0xFFFF_FFFFL) { + throw new IllegalArgumentException( + "The message expiry interval must be between 0 and 4294967295 seconds, not %d".formatted(expiry)); + } + // A four byte integer of the protocol, carried as the bits of an int + properties.add(new IntegerProperty( + MqttProperties.PUBLICATION_EXPIRY_INTERVAL, + (int) expiry.longValue())); + } + return properties; + } + + /** + * Sets the headers of a message from the properties of a received PUBLISH packet. + */ + static void toHeaders(final MqttProperties properties, final Message message) { + if (properties == null || properties.isEmpty()) { + return; + } + final Map<String, String> user = new LinkedHashMap<>(); + for (final MqttProperty<?> property : properties.getProperties(MqttProperties.USER_PROPERTY)) { + final Object value = property.value(); + if (value instanceof final StringPair pair) { + user.put(pair.key, pair.value); Review Comment: โ ๏ธ **MQTT 5 User Properties โ duplicate keys silently dropped** The MQTT 5 spec ยง3.3.2.3.7 explicitly allows the same key to appear more than once in User Properties (and ordering matters). `LinkedHashMap.put()` overwrites earlier values, so any duplicate key from the broker is silently lost. The catalog description documents this ("A key that occurs more than once keeps its last value") but that makes this a known data-loss design choice. Consider using `List<Map.Entry<String, String>>` to preserve all values, or at minimum log a warning when a key is overwritten so the loss is visible at runtime. ########## components/camel-vertx/camel-vertx-mqtt/src/main/java/org/apache/camel/component/vertx/mqtt/BrokerAddress.java: ########## @@ -0,0 +1,89 @@ +/* + * 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.vertx.mqtt; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.Locale; + +import org.apache.camel.util.URISupport; + +/** + * The broker of an endpoint, read from its broker URL. + * + * @param host the host name or address, an IPv6 address without brackets + * @param port the port + * @param ssl whether the connection is TLS + */ +record BrokerAddress(String host, int port, boolean ssl) { + + static final int DEFAULT_PORT = 1883; + static final int DEFAULT_TLS_PORT = 8883; + + /** + * @param url a broker URL such as {@code tcp://broker:1883} or {@code ssl://broker:8883} + * @return the broker it names + * @throws IllegalArgumentException if it is no broker URL of a supported scheme + */ + static BrokerAddress parse(final String url) { + if (url == null || url.isBlank()) { + throw new IllegalArgumentException("The broker URL is required"); + } + // A URL with a password in it must not show it in an error message + final String shown = URISupport.sanitizeUri(url.strip()); + final URI uri; + try { + uri = new URI(url.strip()); + } catch (final URISyntaxException e) { + throw new IllegalArgumentException("Not a broker URL: %s".formatted(shown), e); + } + final String scheme = uri.getScheme() == null ? "" : uri.getScheme().toLowerCase(Locale.ROOT); + final boolean ssl = switch (scheme) { + case "tcp", "mqtt" -> false; + case "ssl", "mqtts" -> true; + case "ws", "wss" -> throw new IllegalArgumentException( + "The WebSocket transport (%s) is not supported yet: %s".formatted(scheme, shown)); + default -> throw new IllegalArgumentException( + "The broker URL %s has no supported scheme, one of tcp, mqtt, ssl or mqtts".formatted(shown)); + }; + // Credentials belong into the username and password options, never into a URL that ends up in logs + if (uri.getRawUserInfo() != null) { + throw new IllegalArgumentException( + "The broker URL must not carry user information, use the username and password options: %s" + .formatted(uri.getHost())); Review Comment: ๐ง **Inconsistent error context โ `uri.getHost()` instead of `shown`** All other error messages in this method use `shown` (the full sanitized URL), which gives the user context about scheme, host, and port. This one formats only `uri.getHost()`, so a user supplying `ssl://user:pass@broker:8883` gets told `"use the username and password options: broker"` โ with no port or scheme. No credential leak (host only), but it's significantly less actionable than the other errors. ```suggestion "The broker URL must not carry user information, use the username and password options: %s" .formatted(shown)); ``` ########## components/camel-vertx/camel-vertx-mqtt/src/main/java/org/apache/camel/component/vertx/mqtt/VertxMqttComponent.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.vertx.mqtt; + +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import io.vertx.core.Vertx; +import io.vertx.core.VertxOptions; +import org.apache.camel.Endpoint; +import org.apache.camel.SSLContextParametersAware; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.annotations.Component; +import org.apache.camel.support.CamelContextHelper; +import org.apache.camel.support.DefaultComponent; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@Component(VertxMqttConstants.SCHEME) +public class VertxMqttComponent extends DefaultComponent implements SSLContextParametersAware { + + private static final Logger LOG = LoggerFactory.getLogger(VertxMqttComponent.class); + + /** + * The shared configuration of the endpoints. + */ + @Metadata + private VertxMqttConfiguration configuration = new VertxMqttConfiguration(); + /** + * The Vert.x instance to use. Without, the one of the registry if there is exactly one, else a Vert.x instance of + * the component's own, closed when the component stops. + */ + @Metadata(label = "advanced", autowired = true) + private Vertx vertx; + /** + * The options of the Vert.x instance the component creates when none is given. + */ + @Metadata(label = "advanced") + private VertxOptions vertxOptions; + /** + * Whether to use the global SSL context parameters of the Camel context for an endpoint that sets none. + */ + @Metadata(label = "security", defaultValue = "false") + private boolean useGlobalSslContextParameters; + + private boolean managedVertx; + + @Override + protected Endpoint createEndpoint(final String uri, final String remaining, final Map<String, Object> parameters) + throws Exception { + final VertxMqttConfiguration config = getConfiguration().copy(); + final VertxMqttEndpoint endpoint = new VertxMqttEndpoint(uri, this, config, remaining); + setProperties(endpoint, parameters); + if (config.getSslContextParameters() == null) { + config.setSslContextParameters(retrieveGlobalSslContextParameters()); + } + return endpoint; + } + + @Override + protected void doInit() throws Exception { + super.doInit(); + if (vertx != null) { + return; + } + final Vertx registered = CamelContextHelper.findSingleByType(getCamelContext(), Vertx.class); + if (registered != null) { + vertx = registered; + } else { + LOG.debug("Creating a Vert.x instance for the component"); + vertx = vertxOptions != null ? Vertx.vertx(vertxOptions) : Vertx.vertx(); + managedVertx = true; + } + } + + @Override + protected void doShutdown() throws Exception { + if (managedVertx && vertx != null) { + vertx.close().toCompletionStage().toCompletableFuture().get(30, TimeUnit.SECONDS); Review Comment: โ ๏ธ **`TimeoutException` from `vertx.close()` propagates out of `doShutdown()`** If Vert.x does not close within 30 seconds, `TimeoutException` propagates uncaught. Camel's shutdown machinery will log a WARN stack trace and the `vertx` field is left non-null (`managedVertx` stays `true`), so the instance is leaked. The pattern in `VertxMqttConnection.stop()` โ catching the exception and logging it โ is safer: ```suggestion try { vertx.close().toCompletionStage().toCompletableFuture().get(30, TimeUnit.SECONDS); } catch (final Exception e) { LOG.warn("Timed out or failed waiting for Vert.x to close: {}", e.getMessage()); } finally { vertx = null; managedVertx = false; } ``` -- 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]
