Croway commented on code in PR #25905: URL: https://github.com/apache/camel/pull/25905#discussion_r3892915272
########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQConfiguration.java: ########## @@ -0,0 +1,177 @@ +/* + * 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.hivemq; + +import com.hivemq.client.mqtt.MqttVersion; +import com.hivemq.client.mqtt.datatypes.MqttQos; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriParams; + +@UriParams +public class HiveMQConfiguration implements Cloneable { + + /** + * Hostname or IP address of the HiveMQ MQTT broker. + */ + @UriParam(defaultValue = HiveMQConstants.DEFAULT_HOST) + private String host = HiveMQConstants.DEFAULT_HOST; + + /** + * Port number of the HiveMQ MQTT broker. + */ + @UriParam(defaultValue = "1883") + private int port = HiveMQConstants.DEFAULT_PORT; + + /** + * Client identifier used when connecting to the HiveMQ broker. + */ + @UriParam + private String clientId; + + /** + * MQTT protocol version to use for the connection. + */ + @UriParam(defaultValue = "MQTT_5_0") + private MqttVersion version = MqttVersion.MQTT_5_0; Review Comment: `version` is a dead option: `HiveMQEndpoint.createClient()` unconditionally calls `.useMqttVersion5()`, so setting `version=MQTT_3_1_1` (or `MQTT_3_1`) silently does nothing — the client's Mqtt3 flavor (`useMqttVersion3()` / `Mqtt3AsyncClient`) is never used. Please either wire this option (branching to the Mqtt3 client API) or remove it until MQTT 3 support is actually implemented, so users don't get a false sense of protocol choice. ########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQConfiguration.java: ########## @@ -0,0 +1,177 @@ +/* + * 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.hivemq; + +import com.hivemq.client.mqtt.MqttVersion; +import com.hivemq.client.mqtt.datatypes.MqttQos; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriParams; + +@UriParams +public class HiveMQConfiguration implements Cloneable { + + /** + * Hostname or IP address of the HiveMQ MQTT broker. + */ + @UriParam(defaultValue = HiveMQConstants.DEFAULT_HOST) + private String host = HiveMQConstants.DEFAULT_HOST; + + /** + * Port number of the HiveMQ MQTT broker. + */ + @UriParam(defaultValue = "1883") + private int port = HiveMQConstants.DEFAULT_PORT; + + /** + * Client identifier used when connecting to the HiveMQ broker. + */ + @UriParam + private String clientId; + + /** + * MQTT protocol version to use for the connection. + */ + @UriParam(defaultValue = "MQTT_5_0") + private MqttVersion version = MqttVersion.MQTT_5_0; + + /** + * Default Quality of Service (QoS) level to use for messages. + */ + @UriParam(defaultValue = "AT_LEAST_ONCE") + private MqttQos qos = MqttQos.AT_LEAST_ONCE; + + /** + * Whether published messages should be retained by the MQTT broker. + */ + @UriParam(defaultValue = "false") + private boolean retained; + + /** + * Whether to initiate a clean session upon connecting to the broker. + */ + @UriParam(defaultValue = "true") + private boolean cleanStart = true; Review Comment: `cleanStart` is a dead option: both consumer and producer call `client.connect()` with defaults, so this value is never passed via `connectWith().cleanStart(...)`. Note also that in MQTT 5, `cleanStart=false` alone does not give a persistent session — the client's default `sessionExpiryInterval` is 0 (session ends on disconnect), so `sessionExpiryInterval` should be exposed alongside it for this option to be meaningful. ########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQConstants.java: ########## @@ -0,0 +1,40 @@ +/* + * 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.hivemq; + +import org.apache.camel.spi.Metadata; + +public final class HiveMQConstants { + + @Metadata(description = "The topic to publish/subscribe to.", javaType = "String") + public static final String MQTT_TOPIC = "CamelHiveMQTopic"; + + @Metadata(description = "The QoS level of the message.", javaType = "Integer") + public static final String MQTT_QOS = "CamelHiveMQQos"; Review Comment: Header type inconsistency: this documents `javaType = "Integer"`, but `HiveMQConsumer` sets the header as an `MqttQos` enum and `HiveMQProducer` reads it as `MqttQos.class`. A user following this doc and setting an `Integer` header (e.g. `1`) will fail type conversion — there is no Integer→MqttQos converter registered. Please pick one representation: either map to/from `int` at the component boundary, or document the header as `com.hivemq.client.mqtt.datatypes.MqttQos`. ########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQEndpoint.java: ########## @@ -0,0 +1,110 @@ +/* + * 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.hivemq; + +import com.hivemq.client.mqtt.MqttClient; +import com.hivemq.client.mqtt.MqttClientBuilder; +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import org.apache.camel.Category; +import org.apache.camel.Consumer; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.spi.EndpointServiceLocation; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.UriEndpoint; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriPath; +import org.apache.camel.support.DefaultEndpoint; + +@UriEndpoint(firstVersion = "4.23.0", scheme = "hivemq", title = "HiveMQ", syntax = "hivemq:topic", + category = { Category.MESSAGING, Category.IOT }, headersClass = HiveMQConstants.class) +public class HiveMQEndpoint extends DefaultEndpoint implements EndpointServiceLocation { + + /** + * The MQTT topic name or pattern to subscribe to or publish on. + */ + @UriPath + @Metadata(required = true) + private String topic; + + /** + * The HiveMQ component configuration options. + */ + @UriParam + @Metadata(description = "To use a custom HiveMQConfiguration") + private HiveMQConfiguration configuration; + + public HiveMQEndpoint(String uri, HiveMQComponent component, HiveMQConfiguration configuration, String topic) { + super(uri, component); + this.configuration = configuration; + this.topic = topic; + } + + @Override + public Producer createProducer() throws Exception { + return new HiveMQProducer(this); + } + + @Override + public Consumer createConsumer(Processor processor) throws Exception { + HiveMQConsumer consumer = new HiveMQConsumer(this, processor); + configureConsumer(consumer); + return consumer; + } + + public Mqtt5AsyncClient createClient() { Review Comment: Two connection-robustness issues in `createClient()`: 1. **No automatic reconnect.** Any broken TCP connection permanently kills the consumer: the route stays `Started` but consumes nothing, and nothing resubscribes. The builder offers `automaticReconnect(...)`/`automaticReconnectWithDefaultConfig()` plus `addConnectedListener`/`addDisconnectedListener` for exactly this — and a resubscribe-on-reconnect step is needed too, since with `cleanStart=true` the session (and its subscriptions) is gone after a drop. camel-paho-mqtt5 exposes `automaticReconnect` and just received a 4.23 fix for the equivalent zombie-route case; a new MQTT component shouldn't reintroduce that failure mode. 2. **clientId collision.** Every producer and consumer builds its own client from the same copied configuration. If `clientId` is set and a route has both a consumer and a producer (or two endpoints share component config), two connections use the same client identifier and the broker must disconnect the older one (MQTT spec), causing a reconnect fight that's hard to debug. Consider a uniquifying suffix per client instance, or document the constraint clearly. ########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQSendDynamicAware.java: ########## @@ -0,0 +1,71 @@ +/* + * 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.hivemq; + +import java.util.Map; + +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.support.component.SendDynamicAwareSupport; +import org.apache.camel.util.StringHelper; + +public class HiveMQSendDynamicAware extends SendDynamicAwareSupport { + Review Comment: This class is currently dead code: it's missing the `@SendDynamic("hivemq")` annotation, so no `META-INF/services/org/apache/camel/send-dynamic/hivemq` file is generated and Camel never discovers it for `toD` (compare `JmsSendDynamicAware` / `PahoSendDynamicAware`). `HiveMQSendDynamicIT` passes trivially because `toD` simply creates a per-topic endpoint without this optimization ever running. Additionally, if it were registered, `resolveStaticUri()` returns `"hivemq:"` with an empty path — but `topic` is a `required = true` `@UriPath`, so creating the static endpoint would fail. `PahoSendDynamicAware` handles this by keeping the original URI's topic as the static endpoint and overriding per-message via header. Please either register it, fix the static URI, and add a real test (assert only one `hivemq:` endpoint exists after sending to multiple dynamic topics), or remove the class for now. ########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQConsumer.java: ########## @@ -0,0 +1,73 @@ +/* + * 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.hivemq; + +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish; +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.support.DefaultConsumer; + +public class HiveMQConsumer extends DefaultConsumer { + + private final HiveMQEndpoint endpoint; + private Mqtt5AsyncClient client; + + public HiveMQConsumer(HiveMQEndpoint endpoint, Processor processor) { + super(endpoint, processor); + this.endpoint = endpoint; + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + client = endpoint.createClient(); + client.connect().join(); + + client.subscribeWith() + .topicFilter(endpoint.getTopic()) + .qos(endpoint.getConfiguration().getQos()) + .callback(this::onMessage) Review Comment: The subscribe callback runs on the client's Netty event-loop thread, so the entire Camel route executes on the client's I/O thread — a slow or blocking route stalls the MQTT client itself (including keep-alive/ack traffic, which can get the client disconnected by the broker). This exact call chain supports `.executor(Executor)` — please wire it to a thread pool from Camel's `ExecutorServiceManager` (and ideally process exchanges asynchronously via `AsyncProcessor`, mirroring the `DefaultAsyncProducer` used on the producer side). Related (fine as a follow-up, but worth documenting): messages are auto-acknowledged on receipt, before the route processes them, so an exception in the route loses the message even at QoS 1/2. The client supports `.manualAcknowledgement(true)` + `Mqtt5Publish.acknowledge()` for ack-after-processing (paho-mqtt5 exposes this as `manualAcksEnabled`). Until then the docs should state the at-most-once processing semantics. ########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQProducer.java: ########## @@ -0,0 +1,75 @@ +/* + * 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.hivemq; + +import com.hivemq.client.mqtt.datatypes.MqttQos; +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish; +import org.apache.camel.AsyncCallback; +import org.apache.camel.Exchange; +import org.apache.camel.support.DefaultAsyncProducer; + +public class HiveMQProducer extends DefaultAsyncProducer { + + private final HiveMQEndpoint endpoint; + private Mqtt5AsyncClient client; + + public HiveMQProducer(HiveMQEndpoint endpoint) { + super(endpoint); + this.endpoint = endpoint; + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + client = endpoint.createClient(); + client.connect().join(); + } + + @Override + protected void doStop() throws Exception { + if (client != null && client.getState().isConnected()) { + client.disconnect().join(); + } + super.doStop(); + } + + @Override + public boolean process(Exchange exchange, AsyncCallback callback) { + String targetTopic = exchange.getIn().getHeader(HiveMQConstants.OVERRIDE_TOPIC, endpoint.getTopic(), String.class); + MqttQos qos = exchange.getIn().getHeader(HiveMQConstants.MQTT_QOS, endpoint.getConfiguration().getQos(), MqttQos.class); + boolean retained = exchange.getIn().getHeader(HiveMQConstants.MQTT_RETAINED, endpoint.getConfiguration().isRetained(), + Boolean.class); + + byte[] payload = exchange.getIn().getBody(byte[].class); Review Comment: If the exchange body is null, `payload` is null here and the behavior of `Mqtt5Publish.builder().payload(null)` is undefined/untested. MQTT explicitly allows empty payloads (publishing an empty retained message is how you clear a retained topic), so a null body should map deliberately to a no-payload publish rather than relying on the builder's null handling. Worth an explicit branch + test. ########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQEndpoint.java: ########## @@ -0,0 +1,110 @@ +/* + * 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.hivemq; + +import com.hivemq.client.mqtt.MqttClient; +import com.hivemq.client.mqtt.MqttClientBuilder; +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import org.apache.camel.Category; +import org.apache.camel.Consumer; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.spi.EndpointServiceLocation; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.UriEndpoint; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriPath; +import org.apache.camel.support.DefaultEndpoint; + +@UriEndpoint(firstVersion = "4.23.0", scheme = "hivemq", title = "HiveMQ", syntax = "hivemq:topic", + category = { Category.MESSAGING, Category.IOT }, headersClass = HiveMQConstants.class) +public class HiveMQEndpoint extends DefaultEndpoint implements EndpointServiceLocation { + + /** + * The MQTT topic name or pattern to subscribe to or publish on. + */ + @UriPath + @Metadata(required = true) + private String topic; + + /** + * The HiveMQ component configuration options. + */ + @UriParam + @Metadata(description = "To use a custom HiveMQConfiguration") + private HiveMQConfiguration configuration; + + public HiveMQEndpoint(String uri, HiveMQComponent component, HiveMQConfiguration configuration, String topic) { + super(uri, component); + this.configuration = configuration; + this.topic = topic; + } + + @Override + public Producer createProducer() throws Exception { + return new HiveMQProducer(this); + } + + @Override + public Consumer createConsumer(Processor processor) throws Exception { + HiveMQConsumer consumer = new HiveMQConsumer(this, processor); + configureConsumer(consumer); + return consumer; + } + + public Mqtt5AsyncClient createClient() { + MqttClientBuilder builder = MqttClient.builder() + .serverHost(configuration.getHost()) + .serverPort(configuration.getPort()); + + if (configuration.getClientId() != null) { + builder.identifier(configuration.getClientId()); + } + + if (configuration.isSsl()) { + builder.sslWithDefaultConfig(); Review Comment: TLS is limited to `sslWithDefaultConfig()`. Camel's convention is an `sslContextParameters` URI option resolving a JSSE `SSLContextParameters` from the registry — without it, custom truststores and mutual TLS (the norm for MQTT broker deployments) are impossible. The client supports this directly via `sslConfig(MqttClientSslConfig)` (key/trustmanager factories), so the mapping is straightforward. -- 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]
