You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2019/10/11 09:44:10 UTC
[camel] branch master updated (9c75fc5 -> ed686ee)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 9c75fc5 Disable external entities for schematron component
new 65885be CAMEL-14052: camel-paho make it possible to configure all its options via compoent options that also works for Spring Boot
new ed686ee CAMEL-14052: camel-paho make it possible to configure all its options via compoent options that also works for Spring Boot
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../camel-paho/src/main/docs/paho-component.adoc | 67 +-
.../apache/camel/component/paho/PahoComponent.java | 91 +-
.../camel/component/paho/PahoConfiguration.java | 581 ++++++
.../apache/camel/component/paho/PahoConsumer.java | 71 +-
.../apache/camel/component/paho/PahoEndpoint.java | 281 +--
.../apache/camel/component/paho/PahoProducer.java | 51 +-
.../camel/component/paho/PahoComponentTest.java | 34 +-
.../component/paho/PahoOverrideTopicTest.java | 12 +-
.../endpoint/dsl/PahoEndpointBuilderFactory.java | 1857 ++++++++++++++++++--
.../springboot/PahoComponentConfiguration.java | 525 +++++-
10 files changed, 3073 insertions(+), 497 deletions(-)
create mode 100644 components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConfiguration.java
[camel] 02/02: CAMEL-14052: camel-paho make it possible to
configure all its options via compoent options that also works for Spring
Boot
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit ed686ee2984f23d71279d80e3fec6d5bc157ba07
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri Oct 11 11:41:56 2019 +0200
CAMEL-14052: camel-paho make it possible to configure all its options via compoent options that also works for Spring Boot
---
.../camel-paho/src/main/docs/paho-component.adoc | 67 +-
.../camel/component/paho/PahoConfiguration.java | 4 +-
.../apache/camel/component/paho/PahoConsumer.java | 71 +-
.../apache/camel/component/paho/PahoEndpoint.java | 81 +-
.../apache/camel/component/paho/PahoProducer.java | 47 +-
.../component/paho/PahoOverrideTopicTest.java | 7 +-
.../endpoint/dsl/PahoEndpointBuilderFactory.java | 1857 ++++++++++++++++++--
.../springboot/PahoComponentConfiguration.java | 525 +++++-
8 files changed, 2434 insertions(+), 225 deletions(-)
diff --git a/components/camel-paho/src/main/docs/paho-component.adoc b/components/camel-paho/src/main/docs/paho-component.adoc
index a03a06a..d83098b 100644
--- a/components/camel-paho/src/main/docs/paho-component.adoc
+++ b/components/camel-paho/src/main/docs/paho-component.adoc
@@ -51,18 +51,16 @@ Where *topic* is the name of the topic.
== Options
// component options: START
-The Paho component supports 6 options, which are listed below.
+The Paho component supports 4 options, which are listed below.
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
+| *configuration* (common) | To use the shared Paho configuration | | PahoConfiguration
+| *client* (advanced) | To use a shared Paho client | | MqttClient
| *brokerUrl* (common) | The URL of the MQTT broker. | | String
-| *clientId* (common) | MQTT client identifier. | | String
-| *userName* (security) | Username to be used for authentication against the MQTT broker | | String
-| *password* (security) | Password to be used for authentication against the MQTT broker | | String
-| *connectOptions* (advanced) | Client connection options | | MqttConnectOptions
| *basicPropertyBinding* (advanced) | Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
|===
// component options: END
@@ -87,28 +85,44 @@ with the following path and query parameters:
|===
-=== Query Parameters (17 parameters):
+=== Query Parameters (33 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
-| *autoReconnect* (common) | Client will automatically attempt to reconnect to the server if the connection is lost | true | boolean
+| *automaticReconnect* (common) | Sets whether the client will automatically attempt to reconnect to the server if the connection is lost. If set to false, the client will not attempt to automatically reconnect to the server in the event that the connection is lost. If set to true, in the event that the connection is lost, the client will attempt to reconnect to the server. It will initially wait 1 second before it attempts to reconnect, for every failed reconnect attempt, the delay will [...]
| *brokerUrl* (common) | The URL of the MQTT broker. | tcp://localhost:1883 | String
-| *clientId* (common) | MQTT client identifier. | | String
-| *connectOptions* (common) | Client connection options | | MqttConnectOptions
+| *cleanSession* (common) | Sets whether the client and server should remember state across restarts and reconnects. If set to false both the client and server will maintain state across restarts of the client, the server and the connection. As state is maintained: Message delivery will be reliable meeting the specified QOS even if the client, server or connection are restarted. The server will treat a subscription as durable. If set to true the client and server will not maintain state [...]
+| *clientId* (common) | MQTT client identifier. The identifier must be unique. | | String
+| *connectionTimeout* (common) | Sets the connection timeout value. This value, measured in seconds, defines the maximum time interval the client will wait for the network connection to the MQTT server to be established. The default timeout is 30 seconds. A value of 0 disables timeout processing meaning the client will wait until the network connection is made successfully or fails. | 30 | int
| *filePersistenceDirectory* (common) | Base directory used by file persistence. Will by default use user directory. | | String
+| *keepAliveInterval* (common) | Sets the keep alive interval. This value, measured in seconds, defines the maximum time interval between messages sent or received. It enables the client to detect if the server is no longer available, without having to wait for the TCP/IP timeout. The client will ensure that at least one message travels across the network within each keep alive period. In the absence of a data-related message during the time period, the client sends a very small ping mes [...]
+| *maxInflight* (common) | Sets the max inflight. please increase this value in a high traffic environment. The default value is 10 | 10 | int
+| *maxReconnectDelay* (common) | Get the maximum time (in millis) to wait between reconnects | 128000 | int
+| *mqttVersion* (common) | Sets the MQTT version. The default action is to connect with version 3.1.1, and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1 can be selected specifically, with no fall back, by using the MQTT_VERSION_3_1_1 or MQTT_VERSION_3_1 options respectively. | | int
| *persistence* (common) | Client persistence to be used - memory or file. | MEMORY | PahoPersistence
| *qos* (common) | Client quality of service level (0-2). | 2 | int
| *retained* (common) | Retain option | false | boolean
+| *serverURIs* (common) | Set a list of one or more serverURIs the client may connect to. Multiple servers can be separated by comma. Each serverURI specifies the address of a server that the client may connect to. Two types of connection are supported tcp:// for a TCP connection and ssl:// for a TCP connection secured by SSL/TLS. For example: tcp://localhost:1883 ssl://localhost:8883 If the port is not specified, it will default to 1883 for tcp:// URIs, and 8883 for ssl:// URIs. If serv [...]
+| *willPayload* (common) | Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to The byte payload for the message. The quality of service to publish the message at (0, 1 or 2). Whether or not the message should be retained. | | String
+| *willQos* (common) | Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to The byte payload for the message. The quality of service to publish the message at (0, 1 or 2). Whether or not the message should be retained. | | int
+| *willRetained* (common) | Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to The byte payload for the message. The quality of service to publish the message at (0, 1 or 2). Whether or not the message should be retained. | false | boolean
+| *willTopic* (common) | Sets the Last Will and Testament (LWT) for the connection. In the event that this client unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to The byte payload for the message. The quality of service to publish the message at (0, 1 or 2). Whether or not the message should be retained. | | String
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
| *basicPropertyBinding* (advanced) | Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
-| *resolveMqttConnectOptions* (advanced) | Define if you don't want to resolve the MQTT Connect Options from registry | true | boolean
+| *client* (advanced) | To use an existing mqtt client | | MqttClient
+| *customWebSocketHeaders* (advanced) | Sets the Custom WebSocket Headers for the WebSocket Connection. | | Properties
+| *executorServiceTimeout* (advanced) | Set the time in seconds that the executor service should wait when terminating before forcefully terminating. It is not recommended to change this value unless you are absolutely sure that you need to. | 1 | int
| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+| *httpsHostnameVerification Enabled* (security) | Whether SSL HostnameVerifier is enabled or not. The default value is true. | true | boolean
| *password* (security) | Password to be used for authentication against the MQTT broker | | String
+| *socketFactory* (security) | Sets the SocketFactory to use. This allows an application to apply its own policies around the creation of network sockets. If using an SSL connection, an SSLSocketFactory can be used to supply application-specific security settings. | | SocketFactory
+| *sslClientProps* (security) | Sets the SSL properties for the connection. Note that these properties are only valid if an implementation of the Java Secure Socket Extensions (JSSE) is available. These properties are not used if a custom SocketFactory has been set. The following properties can be used: com.ibm.ssl.protocol One of: SSL, SSLv3, TLS, TLSv1, SSL_TLS. com.ibm.ssl.contextProvider Underlying JSSE provider. For example IBMJSSE2 or SunJSSE com.ibm.ssl.keyStore The name of the fi [...]
+| *sslHostnameVerifier* (security) | Sets the HostnameVerifier for the SSL connection. Note that it will be used after handshake on a connection and you should do actions by yourself when hostname is verified error. There is no default HostnameVerifier | | HostnameVerifier
| *userName* (security) | Username to be used for authentication against the MQTT broker | | String
|===
// endpoint options: END
@@ -128,7 +142,7 @@ When using Spring Boot make sure to use the following Maven dependency to have s
----
-The component supports 7 options, which are listed below.
+The component supports 30 options, which are listed below.
@@ -137,11 +151,34 @@ The component supports 7 options, which are listed below.
| Name | Description | Default | Type
| *camel.component.paho.basic-property-binding* | Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | Boolean
| *camel.component.paho.broker-url* | The URL of the MQTT broker. | | String
-| *camel.component.paho.client-id* | MQTT client identifier. | | String
-| *camel.component.paho.connect-options* | Client connection options. The option is a org.eclipse.paho.client.mqttv3.MqttConnectOptions type. | | String
+| *camel.component.paho.client* | To use a shared Paho client. The option is a org.eclipse.paho.client.mqttv3.MqttClient type. | | String
+| *camel.component.paho.configuration.automatic-reconnect* | Sets whether the client will automatically attempt to reconnect to the server if the connection is lost. <ul> <li>If set to false, the client will not attempt to automatically reconnect to the server in the event that the connection is lost.</li> <li>If set to true, in the event that the connection is lost, the client will attempt to reconnect to the server. It will initially wait 1 second before it attempts to reconnect, for e [...]
+| *camel.component.paho.configuration.broker-url* | The URL of the MQTT broker. | tcp://localhost:1883 | String
+| *camel.component.paho.configuration.clean-session* | Sets whether the client and server should remember state across restarts and reconnects. <ul> <li>If set to false both the client and server will maintain state across restarts of the client, the server and the connection. As state is maintained: <ul> <li>Message delivery will be reliable meeting the specified QOS even if the client, server or connection are restarted. <li>The server will treat a subscription as durable. </ul> <li>If [...]
+| *camel.component.paho.configuration.client-id* | MQTT client identifier. The identifier must be unique. | | String
+| *camel.component.paho.configuration.connection-timeout* | Sets the connection timeout value. This value, measured in seconds, defines the maximum time interval the client will wait for the network connection to the MQTT server to be established. The default timeout is 30 seconds. A value of 0 disables timeout processing meaning the client will wait until the network connection is made successfully or fails. | 30 | Integer
+| *camel.component.paho.configuration.custom-web-socket-headers* | Sets the Custom WebSocket Headers for the WebSocket Connection. | | Properties
+| *camel.component.paho.configuration.executor-service-timeout* | Set the time in seconds that the executor service should wait when terminating before forcefully terminating. It is not recommended to change this value unless you are absolutely sure that you need to. | 1 | Integer
+| *camel.component.paho.configuration.file-persistence-directory* | Base directory used by file persistence. Will by default use user directory. | | String
+| *camel.component.paho.configuration.https-hostname-verification-enabled* | Whether SSL HostnameVerifier is enabled or not. The default value is true. | true | Boolean
+| *camel.component.paho.configuration.keep-alive-interval* | Sets the keep alive interval. This value, measured in seconds, defines the maximum time interval between messages sent or received. It enables the client to detect if the server is no longer available, without having to wait for the TCP/IP timeout. The client will ensure that at least one message travels across the network within each keep alive period. In the absence of a data-related message during the time period, the client [...]
+| *camel.component.paho.configuration.max-inflight* | Sets the max inflight. please increase this value in a high traffic environment. <p> The default value is 10 </p> | 10 | Integer
+| *camel.component.paho.configuration.max-reconnect-delay* | Get the maximum time (in millis) to wait between reconnects | 128000 | Integer
+| *camel.component.paho.configuration.mqtt-version* | Sets the MQTT version. The default action is to connect with version 3.1.1, and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1 can be selected specifically, with no fall back, by using the MQTT_VERSION_3_1_1 or MQTT_VERSION_3_1 options respectively. | | Integer
+| *camel.component.paho.configuration.password* | Password to be used for authentication against the MQTT broker | | String
+| *camel.component.paho.configuration.persistence* | Client persistence to be used - memory or file. | | PahoPersistence
+| *camel.component.paho.configuration.qos* | Client quality of service level (0-2). | 2 | Integer
+| *camel.component.paho.configuration.retained* | Retain option | false | Boolean
+| *camel.component.paho.configuration.server-u-r-is* | Set a list of one or more serverURIs the client may connect to. Multiple servers can be separated by comma. <p> Each <code>serverURI</code> specifies the address of a server that the client may connect to. Two types of connection are supported <code>tcp://</code> for a TCP connection and <code>ssl://</code> for a TCP connection secured by SSL/TLS. For example: <ul> <li><code>tcp://localhost:1883</code></li> <li><code>ssl://localhost: [...]
+| *camel.component.paho.configuration.socket-factory* | Sets the SocketFactory to use. This allows an application to apply its own policies around the creation of network sockets. If using an SSL connection, an SSLSocketFactory can be used to supply application-specific security settings. | | SocketFactory
+| *camel.component.paho.configuration.ssl-client-props* | Sets the SSL properties for the connection. <p> Note that these properties are only valid if an implementation of the Java Secure Socket Extensions (JSSE) is available. These properties are <em>not</em> used if a custom SocketFactory has been set. The following properties can be used: </p> <dl> <dt>com.ibm.ssl.protocol</dt> <dd>One of: SSL, SSLv3, TLS, TLSv1, SSL_TLS.</dd> <dt>com.ibm.ssl.contextProvider <dd>Underlying JSSE provid [...]
+| *camel.component.paho.configuration.ssl-hostname-verifier* | Sets the HostnameVerifier for the SSL connection. Note that it will be used after handshake on a connection and you should do actions by yourself when hostname is verified error. <p> There is no default HostnameVerifier </p> | | HostnameVerifier
+| *camel.component.paho.configuration.user-name* | Username to be used for authentication against the MQTT broker | | String
+| *camel.component.paho.configuration.will-payload* | Sets the "Last Will and Testament" (LWT) for the connection. In the event that this client unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to The byte payload for the message. The quality of service to publish the message at (0, 1 or 2). Whether or not the message should be retained. | | String
+| *camel.component.paho.configuration.will-qos* | Sets the "Last Will and Testament" (LWT) for the connection. In the event that this client unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to The byte payload for the message. The quality of service to publish the message at (0, 1 or 2). Whether or not the message should be retained. | | Integer
+| *camel.component.paho.configuration.will-retained* | Sets the "Last Will and Testament" (LWT) for the connection. In the event that this client unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to The byte payload for the message. The quality of service to publish the message at (0, 1 or 2). Whether or not the message should be retained. | false | Boolean
+| *camel.component.paho.configuration.will-topic* | Sets the "Last Will and Testament" (LWT) for the connection. In the event that this client unexpectedly loses its connection to the server, the server will publish a message to itself using the supplied details. The topic to publish to The byte payload for the message. The quality of service to publish the message at (0, 1 or 2). Whether or not the message should be retained. | | String
| *camel.component.paho.enabled* | Enable paho component | true | Boolean
-| *camel.component.paho.password* | Password to be used for authentication against the MQTT broker | | String
-| *camel.component.paho.user-name* | Username to be used for authentication against the MQTT broker | | String
|===
// spring-boot-auto-configure options: END
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConfiguration.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConfiguration.java
index 96eb341..af6e0a2 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConfiguration.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConfiguration.java
@@ -30,7 +30,7 @@ import org.eclipse.paho.client.mqttv3.MqttMessage;
public class PahoConfiguration implements Cloneable {
@UriParam
- private String clientId = "camel-" + System.nanoTime();
+ private String clientId;
@UriParam(defaultValue = PahoConstants.DEFAULT_BROKER_URL)
private String brokerUrl = PahoConstants.DEFAULT_BROKER_URL;
@UriParam(defaultValue = "2")
@@ -88,7 +88,7 @@ public class PahoConfiguration implements Cloneable {
}
/**
- * MQTT client identifier.
+ * MQTT client identifier. The identifier must be unique.
*/
public void setClientId(String clientId) {
this.clientId = clientId;
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
index 0fb419a..9d05842 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
@@ -20,30 +20,60 @@ import org.apache.camel.AsyncCallback;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.camel.SuspendableService;
import org.apache.camel.support.DefaultConsumer;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
-public class PahoConsumer extends DefaultConsumer {
+public class PahoConsumer extends DefaultConsumer implements SuspendableService {
+
+ private volatile MqttClient client;
+ private volatile String clientId;
+ private volatile boolean stopClient;
+ private volatile MqttConnectOptions connectOptions;
public PahoConsumer(Endpoint endpoint, Processor processor) {
super(endpoint, processor);
}
+ public MqttClient getClient() {
+ return client;
+ }
+
+ public void setClient(MqttClient client) {
+ this.client = client;
+ }
+
@Override
protected void doStart() throws Exception {
super.doStart();
- String topic = getEndpoint().getTopic();
- getEndpoint().getClient().subscribe(topic, getEndpoint().getConfiguration().getQos());
- getEndpoint().getClient().setCallback(new MqttCallbackExtended() {
+
+ connectOptions = PahoEndpoint.createMqttConnectOptions(getEndpoint().getConfiguration());
+
+ if (client == null) {
+ clientId = getEndpoint().getConfiguration().getClientId();
+ if (clientId == null) {
+ clientId = "camel-" + MqttClient.generateClientId();
+ }
+ stopClient = true;
+ client = new MqttClient(getEndpoint().getConfiguration().getBrokerUrl(),
+ clientId,
+ PahoEndpoint.createMqttClientPersistence(getEndpoint().getConfiguration()));
+ log.debug("Connecting client: {} to broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl());
+ client.connect(connectOptions);
+ }
+
+ client.setCallback(new MqttCallbackExtended() {
@Override
public void connectComplete(boolean reconnect, String serverURI) {
if (reconnect) {
try {
- getEndpoint().getClient().subscribe(topic, getEndpoint().getConfiguration().getQos());
+ client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos());
} catch (MqttException e) {
log.error("MQTT resubscribe failed {}", e.getMessage(), e);
}
@@ -73,15 +103,42 @@ public class PahoConsumer extends DefaultConsumer {
log.debug("Delivery complete. Token: {}", token);
}
});
+
+ log.debug("Subscribing client: {} to topic: {}", clientId, getEndpoint().getTopic());
+ client.subscribe(getEndpoint().getTopic(), getEndpoint().getConfiguration().getQos());
}
@Override
protected void doStop() throws Exception {
super.doStop();
- if (getEndpoint().getClient().isConnected()) {
+ if (stopClient && client != null && client.isConnected()) {
+ String topic = getEndpoint().getTopic();
+ log.debug("Un-unsubscribing client: {} from topic: {}", clientId, topic);
+ client.unsubscribe(topic);
+ log.debug("Connecting client: {} from broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl());
+ client.disconnect();
+ client = null;
+ }
+ }
+
+ @Override
+ protected void doSuspend() throws Exception {
+ super.doSuspend();
+ if (client != null) {
+ String topic = getEndpoint().getTopic();
+ log.debug("Un-unsubscribing client: {} from topic: {}", clientId, topic);
+ client.unsubscribe(topic);
+ }
+ }
+
+ @Override
+ protected void doResume() throws Exception {
+ super.doResume();
+ if (client != null) {
String topic = getEndpoint().getTopic();
- getEndpoint().getClient().unsubscribe(topic);
+ log.debug("Subscribing client: {} to topic: {}", clientId, topic);
+ client.subscribe(topic, getEndpoint().getConfiguration().getQos());
}
}
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoEndpoint.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoEndpoint.java
index f84b425..8994e18 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoEndpoint.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoEndpoint.java
@@ -46,8 +46,7 @@ public class PahoEndpoint extends DefaultEndpoint {
@UriParam
private final PahoConfiguration configuration;
@UriParam(label = "advanced")
- private MqttClient client;
- private transient boolean stopClient;
+ private volatile MqttClient client;
public PahoEndpoint(String uri, String topic, PahoComponent component, PahoConfiguration configuration) {
super(uri, component);
@@ -56,26 +55,42 @@ public class PahoEndpoint extends DefaultEndpoint {
}
@Override
- protected void doStart() throws Exception {
- super.doStart();
+ public Producer createProducer() throws Exception {
+ PahoProducer producer = new PahoProducer(this);
+ producer.setClient(client);
+ return producer;
+ }
- if (client == null) {
- stopClient = true;
- client = new MqttClient(configuration.getBrokerUrl(), configuration.getClientId(), resolvePersistence());
- client.connect(createMqttConnectOptions(configuration));
- }
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ PahoConsumer consumer = new PahoConsumer(this, processor);
+ consumer.setClient(client);
+ configureConsumer(consumer);
+ return consumer;
}
@Override
- protected void doStop() throws Exception {
- if (stopClient && client.isConnected()) {
- client.disconnect();
- }
+ public PahoComponent getComponent() {
+ return (PahoComponent) super.getComponent();
+ }
+
+ public String getTopic() {
+ return topic;
+ }
- super.doStop();
+ public Exchange createExchange(MqttMessage mqttMessage, String topic) {
+ Exchange exchange = createExchange();
+
+ PahoMessage paho = new PahoMessage(exchange.getContext(), mqttMessage);
+ paho.setBody(mqttMessage.getPayload());
+ paho.setHeader(PahoConstants.MQTT_TOPIC, topic);
+ paho.setHeader(PahoConstants.MQTT_QOS, mqttMessage.getQos());
+
+ exchange.setIn(paho);
+ return exchange;
}
- private static MqttConnectOptions createMqttConnectOptions(PahoConfiguration config) {
+ protected static MqttConnectOptions createMqttConnectOptions(PahoConfiguration config) {
MqttConnectOptions mq = new MqttConnectOptions();
if (ObjectHelper.isNotEmpty(config.getUserName()) && ObjectHelper.isNotEmpty(config.getPassword())) {
mq.setUserName(config.getUserName());
@@ -106,27 +121,7 @@ public class PahoEndpoint extends DefaultEndpoint {
return mq;
}
- @Override
- public Producer createProducer() throws Exception {
- return new PahoProducer(this);
- }
-
- @Override
- public Consumer createConsumer(Processor processor) throws Exception {
- return new PahoConsumer(this, processor);
- }
-
- @Override
- public PahoComponent getComponent() {
- return (PahoComponent) super.getComponent();
- }
-
- public String getTopic() {
- return topic;
- }
-
- // Resolvers
- protected MqttClientPersistence resolvePersistence() {
+ protected static MqttClientPersistence createMqttClientPersistence(PahoConfiguration configuration) {
if (configuration.getPersistence() == PahoPersistence.MEMORY) {
return new MemoryPersistence();
} else {
@@ -138,18 +133,6 @@ public class PahoEndpoint extends DefaultEndpoint {
}
}
- public Exchange createExchange(MqttMessage mqttMessage, String topic) {
- Exchange exchange = createExchange();
-
- PahoMessage paho = new PahoMessage(exchange.getContext(), mqttMessage);
- paho.setBody(mqttMessage.getPayload());
- paho.setHeader(PahoConstants.MQTT_TOPIC, topic);
- paho.setHeader(PahoConstants.MQTT_QOS, mqttMessage.getQos());
-
- exchange.setIn(paho);
- return exchange;
- }
-
public PahoConfiguration getConfiguration() {
return configuration;
}
@@ -159,7 +142,7 @@ public class PahoEndpoint extends DefaultEndpoint {
}
/**
- * To use an exiting mqtt client
+ * To use an existing mqtt client
*/
public void setClient(MqttClient client) {
this.client = client;
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
index 08428a5..88b1045 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
@@ -19,18 +19,22 @@ package org.apache.camel.component.paho;
import org.apache.camel.Exchange;
import org.apache.camel.support.DefaultProducer;
import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
public class PahoProducer extends DefaultProducer {
+ private volatile MqttClient client;
+ private volatile String clientId;
+ private volatile boolean stopClient;
+ private volatile MqttConnectOptions connectOptions;
+
public PahoProducer(PahoEndpoint endpoint) {
super(endpoint);
}
@Override
public void process(Exchange exchange) throws Exception {
- MqttClient client = getEndpoint().getClient();
-
String topic = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC, getEndpoint().getTopic(), String.class);
int qos = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_MSG_QOS, getEndpoint().getConfiguration().getQos(), Integer.class);
boolean retained = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_MSG_RETAINED, getEndpoint().getConfiguration().isRetained(), Boolean.class);
@@ -49,4 +53,43 @@ public class PahoProducer extends DefaultProducer {
return (PahoEndpoint)super.getEndpoint();
}
+ public MqttClient getClient() {
+ return client;
+ }
+
+ public void setClient(MqttClient client) {
+ this.client = client;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ connectOptions = PahoEndpoint.createMqttConnectOptions(getEndpoint().getConfiguration());
+
+ if (client == null) {
+ clientId = getEndpoint().getConfiguration().getClientId();
+ if (clientId == null) {
+ clientId = "camel-" + MqttClient.generateClientId();
+ }
+ stopClient = true;
+ client = new MqttClient(getEndpoint().getConfiguration().getBrokerUrl(),
+ clientId,
+ PahoEndpoint.createMqttClientPersistence(getEndpoint().getConfiguration()));
+ log.debug("Connecting client: {} to broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl());
+ client.connect(connectOptions);
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+
+ if (stopClient && client != null && client.isConnected()) {
+ log.debug("Disconnecting client: {} from broker: {}", clientId, getEndpoint().getConfiguration().getBrokerUrl());
+ client.disconnect();
+ client = null;
+ }
+ }
+
}
diff --git a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java
index ef751a4..43daac4 100644
--- a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java
+++ b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java
@@ -55,9 +55,12 @@ public class PahoOverrideTopicTest extends CamelTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:test").to("paho:queue?brokerUrl=tcp://localhost:" + mqttPort);
- from("paho:myoverride?brokerUrl=tcp://localhost:" + mqttPort).to("mock:test");
+ PahoComponent paho = context.getComponent("paho", PahoComponent.class);
+ paho.setBrokerUrl("tcp://localhost:" + mqttPort);
+ from("direct:test").to("paho:queue").log("Message sent");
+
+ from("paho:myoverride").log("Message received").to("mock:test");
}
};
}
diff --git a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/PahoEndpointBuilderFactory.java b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/PahoEndpointBuilderFactory.java
index e438339..1c17926 100644
--- a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/PahoEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/PahoEndpointBuilderFactory.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.builder.endpoint.dsl;
+import java.util.Properties;
import javax.annotation.Generated;
import org.apache.camel.ExchangePattern;
import org.apache.camel.builder.EndpointConsumerBuilder;
@@ -24,8 +25,8 @@ import org.apache.camel.builder.endpoint.AbstractEndpointBuilder;
import org.apache.camel.spi.ExceptionHandler;
/**
- * Component for communicating with MQTT M2M message brokers using Eclipse Paho
- * MQTT Client.
+ * Component for communicating with MQTT message brokers using Eclipse Paho MQTT
+ * Client.
*
* Generated by camel-package-maven-plugin - do not edit this file!
*/
@@ -43,27 +44,43 @@ public interface PahoEndpointBuilderFactory {
return (AdvancedPahoEndpointConsumerBuilder) this;
}
/**
- * Client will automatically attempt to reconnect to the server if the
- * connection is lost.
+ * Sets whether the client will automatically attempt to reconnect to
+ * the server if the connection is lost. If set to false, the client
+ * will not attempt to automatically reconnect to the server in the
+ * event that the connection is lost. If set to true, in the event that
+ * the connection is lost, the client will attempt to reconnect to the
+ * server. It will initially wait 1 second before it attempts to
+ * reconnect, for every failed reconnect attempt, the delay will double
+ * until it is at 2 minutes at which point the delay will stay at 2
+ * minutes.
*
* The option is a: <code>boolean</code> type.
*
* Group: common
*/
- default PahoEndpointConsumerBuilder autoReconnect(boolean autoReconnect) {
- doSetProperty("autoReconnect", autoReconnect);
+ default PahoEndpointConsumerBuilder automaticReconnect(
+ boolean automaticReconnect) {
+ doSetProperty("automaticReconnect", automaticReconnect);
return this;
}
/**
- * Client will automatically attempt to reconnect to the server if the
- * connection is lost.
+ * Sets whether the client will automatically attempt to reconnect to
+ * the server if the connection is lost. If set to false, the client
+ * will not attempt to automatically reconnect to the server in the
+ * event that the connection is lost. If set to true, in the event that
+ * the connection is lost, the client will attempt to reconnect to the
+ * server. It will initially wait 1 second before it attempts to
+ * reconnect, for every failed reconnect attempt, the delay will double
+ * until it is at 2 minutes at which point the delay will stay at 2
+ * minutes.
*
* The option will be converted to a <code>boolean</code> type.
*
* Group: common
*/
- default PahoEndpointConsumerBuilder autoReconnect(String autoReconnect) {
- doSetProperty("autoReconnect", autoReconnect);
+ default PahoEndpointConsumerBuilder automaticReconnect(
+ String automaticReconnect) {
+ doSetProperty("automaticReconnect", automaticReconnect);
return this;
}
/**
@@ -78,7 +95,49 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * MQTT client identifier.
+ * Sets whether the client and server should remember state across
+ * restarts and reconnects. If set to false both the client and server
+ * will maintain state across restarts of the client, the server and the
+ * connection. As state is maintained: Message delivery will be reliable
+ * meeting the specified QOS even if the client, server or connection
+ * are restarted. The server will treat a subscription as durable. If
+ * set to true the client and server will not maintain state across
+ * restarts of the client, the server or the connection. This means
+ * Message delivery to the specified QOS cannot be maintained if the
+ * client, server or connection are restarted The server will treat a
+ * subscription as non-durable.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder cleanSession(boolean cleanSession) {
+ doSetProperty("cleanSession", cleanSession);
+ return this;
+ }
+ /**
+ * Sets whether the client and server should remember state across
+ * restarts and reconnects. If set to false both the client and server
+ * will maintain state across restarts of the client, the server and the
+ * connection. As state is maintained: Message delivery will be reliable
+ * meeting the specified QOS even if the client, server or connection
+ * are restarted. The server will treat a subscription as durable. If
+ * set to true the client and server will not maintain state across
+ * restarts of the client, the server or the connection. This means
+ * Message delivery to the specified QOS cannot be maintained if the
+ * client, server or connection are restarted The server will treat a
+ * subscription as non-durable.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder cleanSession(String cleanSession) {
+ doSetProperty("cleanSession", cleanSession);
+ return this;
+ }
+ /**
+ * MQTT client identifier. The identifier must be unique.
*
* The option is a: <code>java.lang.String</code> type.
*
@@ -89,27 +148,37 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * Client connection options.
+ * Sets the connection timeout value. This value, measured in seconds,
+ * defines the maximum time interval the client will wait for the
+ * network connection to the MQTT server to be established. The default
+ * timeout is 30 seconds. A value of 0 disables timeout processing
+ * meaning the client will wait until the network connection is made
+ * successfully or fails.
*
- * The option is a:
- * <code>org.eclipse.paho.client.mqttv3.MqttConnectOptions</code> type.
+ * The option is a: <code>int</code> type.
*
* Group: common
*/
- default PahoEndpointConsumerBuilder connectOptions(Object connectOptions) {
- doSetProperty("connectOptions", connectOptions);
+ default PahoEndpointConsumerBuilder connectionTimeout(
+ int connectionTimeout) {
+ doSetProperty("connectionTimeout", connectionTimeout);
return this;
}
/**
- * Client connection options.
+ * Sets the connection timeout value. This value, measured in seconds,
+ * defines the maximum time interval the client will wait for the
+ * network connection to the MQTT server to be established. The default
+ * timeout is 30 seconds. A value of 0 disables timeout processing
+ * meaning the client will wait until the network connection is made
+ * successfully or fails.
*
- * The option will be converted to a
- * <code>org.eclipse.paho.client.mqttv3.MqttConnectOptions</code> type.
+ * The option will be converted to a <code>int</code> type.
*
* Group: common
*/
- default PahoEndpointConsumerBuilder connectOptions(String connectOptions) {
- doSetProperty("connectOptions", connectOptions);
+ default PahoEndpointConsumerBuilder connectionTimeout(
+ String connectionTimeout) {
+ doSetProperty("connectionTimeout", connectionTimeout);
return this;
}
/**
@@ -126,6 +195,122 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Sets the keep alive interval. This value, measured in seconds,
+ * defines the maximum time interval between messages sent or received.
+ * It enables the client to detect if the server is no longer available,
+ * without having to wait for the TCP/IP timeout. The client will ensure
+ * that at least one message travels across the network within each keep
+ * alive period. In the absence of a data-related message during the
+ * time period, the client sends a very small ping message, which the
+ * server will acknowledge. A value of 0 disables keepalive processing
+ * in the client. The default value is 60 seconds.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder keepAliveInterval(
+ int keepAliveInterval) {
+ doSetProperty("keepAliveInterval", keepAliveInterval);
+ return this;
+ }
+ /**
+ * Sets the keep alive interval. This value, measured in seconds,
+ * defines the maximum time interval between messages sent or received.
+ * It enables the client to detect if the server is no longer available,
+ * without having to wait for the TCP/IP timeout. The client will ensure
+ * that at least one message travels across the network within each keep
+ * alive period. In the absence of a data-related message during the
+ * time period, the client sends a very small ping message, which the
+ * server will acknowledge. A value of 0 disables keepalive processing
+ * in the client. The default value is 60 seconds.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder keepAliveInterval(
+ String keepAliveInterval) {
+ doSetProperty("keepAliveInterval", keepAliveInterval);
+ return this;
+ }
+ /**
+ * Sets the max inflight. please increase this value in a high traffic
+ * environment. The default value is 10.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder maxInflight(int maxInflight) {
+ doSetProperty("maxInflight", maxInflight);
+ return this;
+ }
+ /**
+ * Sets the max inflight. please increase this value in a high traffic
+ * environment. The default value is 10.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder maxInflight(String maxInflight) {
+ doSetProperty("maxInflight", maxInflight);
+ return this;
+ }
+ /**
+ * Get the maximum time (in millis) to wait between reconnects.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder maxReconnectDelay(
+ int maxReconnectDelay) {
+ doSetProperty("maxReconnectDelay", maxReconnectDelay);
+ return this;
+ }
+ /**
+ * Get the maximum time (in millis) to wait between reconnects.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder maxReconnectDelay(
+ String maxReconnectDelay) {
+ doSetProperty("maxReconnectDelay", maxReconnectDelay);
+ return this;
+ }
+ /**
+ * Sets the MQTT version. The default action is to connect with version
+ * 3.1.1, and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1
+ * can be selected specifically, with no fall back, by using the
+ * MQTT_VERSION_3_1_1 or MQTT_VERSION_3_1 options respectively.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder mqttVersion(int mqttVersion) {
+ doSetProperty("mqttVersion", mqttVersion);
+ return this;
+ }
+ /**
+ * Sets the MQTT version. The default action is to connect with version
+ * 3.1.1, and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1
+ * can be selected specifically, with no fall back, by using the
+ * MQTT_VERSION_3_1_1 or MQTT_VERSION_3_1 options respectively.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder mqttVersion(String mqttVersion) {
+ doSetProperty("mqttVersion", mqttVersion);
+ return this;
+ }
+ /**
* Client persistence to be used - memory or file.
*
* The option is a:
@@ -195,6 +380,137 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Set a list of one or more serverURIs the client may connect to.
+ * Multiple servers can be separated by comma. Each serverURI specifies
+ * the address of a server that the client may connect to. Two types of
+ * connection are supported tcp:// for a TCP connection and ssl:// for a
+ * TCP connection secured by SSL/TLS. For example: tcp://localhost:1883
+ * ssl://localhost:8883 If the port is not specified, it will default to
+ * 1883 for tcp:// URIs, and 8883 for ssl:// URIs. If serverURIs is set
+ * then it overrides the serverURI parameter passed in on the
+ * constructor of the MQTT client. When an attempt to connect is
+ * initiated the client will start with the first serverURI in the list
+ * and work through the list until a connection is established with a
+ * server. If a connection cannot be made to any of the servers then the
+ * connect attempt fails. Specifying a list of servers that a client may
+ * connect to has several uses: High Availability and reliable message
+ * delivery Some MQTT servers support a high availability feature where
+ * two or more equal MQTT servers share state. An MQTT client can
+ * connect to any of the equal servers and be assured that messages are
+ * reliably delivered and durable subscriptions are maintained no matter
+ * which server the client connects to. The cleansession flag must be
+ * set to false if durable subscriptions and/or reliable message
+ * delivery is required. Hunt List A set of servers may be specified
+ * that are not equal (as in the high availability option). As no state
+ * is shared across the servers reliable message delivery and durable
+ * subscriptions are not valid. The cleansession flag must be set to
+ * true if the hunt list mode is used.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder serverURIs(String serverURIs) {
+ doSetProperty("serverURIs", serverURIs);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder willPayload(String willPayload) {
+ doSetProperty("willPayload", willPayload);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder willQos(int willQos) {
+ doSetProperty("willQos", willQos);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder willQos(String willQos) {
+ doSetProperty("willQos", willQos);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder willRetained(boolean willRetained) {
+ doSetProperty("willRetained", willRetained);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder willRetained(String willRetained) {
+ doSetProperty("willRetained", willRetained);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointConsumerBuilder willTopic(String willTopic) {
+ doSetProperty("willTopic", willTopic);
+ return this;
+ }
+ /**
* Allows for bridging the consumer to the Camel routing Error Handler,
* which mean any exceptions occurred while the consumer is trying to
* pickup incoming messages, or the likes, will now be processed as a
@@ -231,6 +547,32 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Whether SSL HostnameVerifier is enabled or not. The default value is
+ * true.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointConsumerBuilder httpsHostnameVerificationEnabled(
+ boolean httpsHostnameVerificationEnabled) {
+ doSetProperty("httpsHostnameVerificationEnabled", httpsHostnameVerificationEnabled);
+ return this;
+ }
+ /**
+ * Whether SSL HostnameVerifier is enabled or not. The default value is
+ * true.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointConsumerBuilder httpsHostnameVerificationEnabled(
+ String httpsHostnameVerificationEnabled) {
+ doSetProperty("httpsHostnameVerificationEnabled", httpsHostnameVerificationEnabled);
+ return this;
+ }
+ /**
* Password to be used for authentication against the MQTT broker.
*
* The option is a: <code>java.lang.String</code> type.
@@ -242,6 +584,166 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Sets the SocketFactory to use. This allows an application to apply
+ * its own policies around the creation of network sockets. If using an
+ * SSL connection, an SSLSocketFactory can be used to supply
+ * application-specific security settings.
+ *
+ * The option is a: <code>javax.net.SocketFactory</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointConsumerBuilder socketFactory(Object socketFactory) {
+ doSetProperty("socketFactory", socketFactory);
+ return this;
+ }
+ /**
+ * Sets the SocketFactory to use. This allows an application to apply
+ * its own policies around the creation of network sockets. If using an
+ * SSL connection, an SSLSocketFactory can be used to supply
+ * application-specific security settings.
+ *
+ * The option will be converted to a
+ * <code>javax.net.SocketFactory</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointConsumerBuilder socketFactory(String socketFactory) {
+ doSetProperty("socketFactory", socketFactory);
+ return this;
+ }
+ /**
+ * Sets the SSL properties for the connection. Note that these
+ * properties are only valid if an implementation of the Java Secure
+ * Socket Extensions (JSSE) is available. These properties are not used
+ * if a custom SocketFactory has been set. The following properties can
+ * be used: com.ibm.ssl.protocol One of: SSL, SSLv3, TLS, TLSv1,
+ * SSL_TLS. com.ibm.ssl.contextProvider Underlying JSSE provider. For
+ * example IBMJSSE2 or SunJSSE com.ibm.ssl.keyStore The name of the file
+ * that contains the KeyStore object that you want the KeyManager to
+ * use. For example /mydir/etc/key.p12 com.ibm.ssl.keyStorePassword The
+ * password for the KeyStore object that you want the KeyManager to use.
+ * The password can either be in plain-text, or may be obfuscated using
+ * the static method: com.ibm.micro.security.Password.obfuscate(char
+ * password). This obfuscates the password using a simple and insecure
+ * XOR and Base64 encoding mechanism. Note that this is only a simple
+ * scrambler to obfuscate clear-text passwords. com.ibm.ssl.keyStoreType
+ * Type of key store, for example PKCS12, JKS, or JCEKS.
+ * com.ibm.ssl.keyStoreProvider Key store provider, for example IBMJCE
+ * or IBMJCEFIPS. com.ibm.ssl.trustStore The name of the file that
+ * contains the KeyStore object that you want the TrustManager to use.
+ * com.ibm.ssl.trustStorePassword The password for the TrustStore object
+ * that you want the TrustManager to use. The password can either be in
+ * plain-text, or may be obfuscated using the static method:
+ * com.ibm.micro.security.Password.obfuscate(char password). This
+ * obfuscates the password using a simple and insecure XOR and Base64
+ * encoding mechanism. Note that this is only a simple scrambler to
+ * obfuscate clear-text passwords. com.ibm.ssl.trustStoreType The type
+ * of KeyStore object that you want the default TrustManager to use.
+ * Same possible values as keyStoreType. com.ibm.ssl.trustStoreProvider
+ * Trust store provider, for example IBMJCE or IBMJCEFIPS.
+ * com.ibm.ssl.enabledCipherSuites A list of which ciphers are enabled.
+ * Values are dependent on the provider, for example:
+ * SSL_RSA_WITH_AES_128_CBC_SHA;SSL_RSA_WITH_3DES_EDE_CBC_SHA.
+ * com.ibm.ssl.keyManager Sets the algorithm that will be used to
+ * instantiate a KeyManagerFactory object instead of using the default
+ * algorithm available in the platform. Example values: IbmX509 or
+ * IBMJ9X509. com.ibm.ssl.trustManager Sets the algorithm that will be
+ * used to instantiate a TrustManagerFactory object instead of using the
+ * default algorithm available in the platform. Example values: PKIX or
+ * IBMJ9X509.
+ *
+ * The option is a: <code>java.util.Properties</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointConsumerBuilder sslClientProps(
+ Properties sslClientProps) {
+ doSetProperty("sslClientProps", sslClientProps);
+ return this;
+ }
+ /**
+ * Sets the SSL properties for the connection. Note that these
+ * properties are only valid if an implementation of the Java Secure
+ * Socket Extensions (JSSE) is available. These properties are not used
+ * if a custom SocketFactory has been set. The following properties can
+ * be used: com.ibm.ssl.protocol One of: SSL, SSLv3, TLS, TLSv1,
+ * SSL_TLS. com.ibm.ssl.contextProvider Underlying JSSE provider. For
+ * example IBMJSSE2 or SunJSSE com.ibm.ssl.keyStore The name of the file
+ * that contains the KeyStore object that you want the KeyManager to
+ * use. For example /mydir/etc/key.p12 com.ibm.ssl.keyStorePassword The
+ * password for the KeyStore object that you want the KeyManager to use.
+ * The password can either be in plain-text, or may be obfuscated using
+ * the static method: com.ibm.micro.security.Password.obfuscate(char
+ * password). This obfuscates the password using a simple and insecure
+ * XOR and Base64 encoding mechanism. Note that this is only a simple
+ * scrambler to obfuscate clear-text passwords. com.ibm.ssl.keyStoreType
+ * Type of key store, for example PKCS12, JKS, or JCEKS.
+ * com.ibm.ssl.keyStoreProvider Key store provider, for example IBMJCE
+ * or IBMJCEFIPS. com.ibm.ssl.trustStore The name of the file that
+ * contains the KeyStore object that you want the TrustManager to use.
+ * com.ibm.ssl.trustStorePassword The password for the TrustStore object
+ * that you want the TrustManager to use. The password can either be in
+ * plain-text, or may be obfuscated using the static method:
+ * com.ibm.micro.security.Password.obfuscate(char password). This
+ * obfuscates the password using a simple and insecure XOR and Base64
+ * encoding mechanism. Note that this is only a simple scrambler to
+ * obfuscate clear-text passwords. com.ibm.ssl.trustStoreType The type
+ * of KeyStore object that you want the default TrustManager to use.
+ * Same possible values as keyStoreType. com.ibm.ssl.trustStoreProvider
+ * Trust store provider, for example IBMJCE or IBMJCEFIPS.
+ * com.ibm.ssl.enabledCipherSuites A list of which ciphers are enabled.
+ * Values are dependent on the provider, for example:
+ * SSL_RSA_WITH_AES_128_CBC_SHA;SSL_RSA_WITH_3DES_EDE_CBC_SHA.
+ * com.ibm.ssl.keyManager Sets the algorithm that will be used to
+ * instantiate a KeyManagerFactory object instead of using the default
+ * algorithm available in the platform. Example values: IbmX509 or
+ * IBMJ9X509. com.ibm.ssl.trustManager Sets the algorithm that will be
+ * used to instantiate a TrustManagerFactory object instead of using the
+ * default algorithm available in the platform. Example values: PKIX or
+ * IBMJ9X509.
+ *
+ * The option will be converted to a <code>java.util.Properties</code>
+ * type.
+ *
+ * Group: security
+ */
+ default PahoEndpointConsumerBuilder sslClientProps(String sslClientProps) {
+ doSetProperty("sslClientProps", sslClientProps);
+ return this;
+ }
+ /**
+ * Sets the HostnameVerifier for the SSL connection. Note that it will
+ * be used after handshake on a connection and you should do actions by
+ * yourself when hostname is verified error. There is no default
+ * HostnameVerifier.
+ *
+ * The option is a: <code>javax.net.ssl.HostnameVerifier</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointConsumerBuilder sslHostnameVerifier(
+ Object sslHostnameVerifier) {
+ doSetProperty("sslHostnameVerifier", sslHostnameVerifier);
+ return this;
+ }
+ /**
+ * Sets the HostnameVerifier for the SSL connection. Note that it will
+ * be used after handshake on a connection and you should do actions by
+ * yourself when hostname is verified error. There is no default
+ * HostnameVerifier.
+ *
+ * The option will be converted to a
+ * <code>javax.net.ssl.HostnameVerifier</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointConsumerBuilder sslHostnameVerifier(
+ String sslHostnameVerifier) {
+ doSetProperty("sslHostnameVerifier", sslHostnameVerifier);
+ return this;
+ }
+ /**
* Username to be used for authentication against the MQTT broker.
*
* The option is a: <code>java.lang.String</code> type.
@@ -347,29 +849,80 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * Define if you don't want to resolve the MQTT Connect Options from
- * registry.
+ * To use an existing mqtt client.
*
- * The option is a: <code>boolean</code> type.
+ * The option is a:
+ * <code>org.eclipse.paho.client.mqttv3.MqttClient</code> type.
*
* Group: advanced
*/
- default AdvancedPahoEndpointConsumerBuilder resolveMqttConnectOptions(
- boolean resolveMqttConnectOptions) {
- doSetProperty("resolveMqttConnectOptions", resolveMqttConnectOptions);
+ default AdvancedPahoEndpointConsumerBuilder client(Object client) {
+ doSetProperty("client", client);
return this;
}
/**
- * Define if you don't want to resolve the MQTT Connect Options from
- * registry.
+ * To use an existing mqtt client.
*
- * The option will be converted to a <code>boolean</code> type.
+ * The option will be converted to a
+ * <code>org.eclipse.paho.client.mqttv3.MqttClient</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointConsumerBuilder client(String client) {
+ doSetProperty("client", client);
+ return this;
+ }
+ /**
+ * Sets the Custom WebSocket Headers for the WebSocket Connection.
+ *
+ * The option is a: <code>java.util.Properties</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointConsumerBuilder customWebSocketHeaders(
+ Properties customWebSocketHeaders) {
+ doSetProperty("customWebSocketHeaders", customWebSocketHeaders);
+ return this;
+ }
+ /**
+ * Sets the Custom WebSocket Headers for the WebSocket Connection.
+ *
+ * The option will be converted to a <code>java.util.Properties</code>
+ * type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointConsumerBuilder customWebSocketHeaders(
+ String customWebSocketHeaders) {
+ doSetProperty("customWebSocketHeaders", customWebSocketHeaders);
+ return this;
+ }
+ /**
+ * Set the time in seconds that the executor service should wait when
+ * terminating before forcefully terminating. It is not recommended to
+ * change this value unless you are absolutely sure that you need to.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointConsumerBuilder executorServiceTimeout(
+ int executorServiceTimeout) {
+ doSetProperty("executorServiceTimeout", executorServiceTimeout);
+ return this;
+ }
+ /**
+ * Set the time in seconds that the executor service should wait when
+ * terminating before forcefully terminating. It is not recommended to
+ * change this value unless you are absolutely sure that you need to.
+ *
+ * The option will be converted to a <code>int</code> type.
*
* Group: advanced
*/
- default AdvancedPahoEndpointConsumerBuilder resolveMqttConnectOptions(
- String resolveMqttConnectOptions) {
- doSetProperty("resolveMqttConnectOptions", resolveMqttConnectOptions);
+ default AdvancedPahoEndpointConsumerBuilder executorServiceTimeout(
+ String executorServiceTimeout) {
+ doSetProperty("executorServiceTimeout", executorServiceTimeout);
return this;
}
/**
@@ -410,27 +963,43 @@ public interface PahoEndpointBuilderFactory {
return (AdvancedPahoEndpointProducerBuilder) this;
}
/**
- * Client will automatically attempt to reconnect to the server if the
- * connection is lost.
+ * Sets whether the client will automatically attempt to reconnect to
+ * the server if the connection is lost. If set to false, the client
+ * will not attempt to automatically reconnect to the server in the
+ * event that the connection is lost. If set to true, in the event that
+ * the connection is lost, the client will attempt to reconnect to the
+ * server. It will initially wait 1 second before it attempts to
+ * reconnect, for every failed reconnect attempt, the delay will double
+ * until it is at 2 minutes at which point the delay will stay at 2
+ * minutes.
*
* The option is a: <code>boolean</code> type.
*
* Group: common
*/
- default PahoEndpointProducerBuilder autoReconnect(boolean autoReconnect) {
- doSetProperty("autoReconnect", autoReconnect);
+ default PahoEndpointProducerBuilder automaticReconnect(
+ boolean automaticReconnect) {
+ doSetProperty("automaticReconnect", automaticReconnect);
return this;
}
/**
- * Client will automatically attempt to reconnect to the server if the
- * connection is lost.
+ * Sets whether the client will automatically attempt to reconnect to
+ * the server if the connection is lost. If set to false, the client
+ * will not attempt to automatically reconnect to the server in the
+ * event that the connection is lost. If set to true, in the event that
+ * the connection is lost, the client will attempt to reconnect to the
+ * server. It will initially wait 1 second before it attempts to
+ * reconnect, for every failed reconnect attempt, the delay will double
+ * until it is at 2 minutes at which point the delay will stay at 2
+ * minutes.
*
* The option will be converted to a <code>boolean</code> type.
*
* Group: common
*/
- default PahoEndpointProducerBuilder autoReconnect(String autoReconnect) {
- doSetProperty("autoReconnect", autoReconnect);
+ default PahoEndpointProducerBuilder automaticReconnect(
+ String automaticReconnect) {
+ doSetProperty("automaticReconnect", automaticReconnect);
return this;
}
/**
@@ -445,7 +1014,49 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * MQTT client identifier.
+ * Sets whether the client and server should remember state across
+ * restarts and reconnects. If set to false both the client and server
+ * will maintain state across restarts of the client, the server and the
+ * connection. As state is maintained: Message delivery will be reliable
+ * meeting the specified QOS even if the client, server or connection
+ * are restarted. The server will treat a subscription as durable. If
+ * set to true the client and server will not maintain state across
+ * restarts of the client, the server or the connection. This means
+ * Message delivery to the specified QOS cannot be maintained if the
+ * client, server or connection are restarted The server will treat a
+ * subscription as non-durable.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder cleanSession(boolean cleanSession) {
+ doSetProperty("cleanSession", cleanSession);
+ return this;
+ }
+ /**
+ * Sets whether the client and server should remember state across
+ * restarts and reconnects. If set to false both the client and server
+ * will maintain state across restarts of the client, the server and the
+ * connection. As state is maintained: Message delivery will be reliable
+ * meeting the specified QOS even if the client, server or connection
+ * are restarted. The server will treat a subscription as durable. If
+ * set to true the client and server will not maintain state across
+ * restarts of the client, the server or the connection. This means
+ * Message delivery to the specified QOS cannot be maintained if the
+ * client, server or connection are restarted The server will treat a
+ * subscription as non-durable.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder cleanSession(String cleanSession) {
+ doSetProperty("cleanSession", cleanSession);
+ return this;
+ }
+ /**
+ * MQTT client identifier. The identifier must be unique.
*
* The option is a: <code>java.lang.String</code> type.
*
@@ -456,27 +1067,37 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * Client connection options.
+ * Sets the connection timeout value. This value, measured in seconds,
+ * defines the maximum time interval the client will wait for the
+ * network connection to the MQTT server to be established. The default
+ * timeout is 30 seconds. A value of 0 disables timeout processing
+ * meaning the client will wait until the network connection is made
+ * successfully or fails.
*
- * The option is a:
- * <code>org.eclipse.paho.client.mqttv3.MqttConnectOptions</code> type.
+ * The option is a: <code>int</code> type.
*
* Group: common
*/
- default PahoEndpointProducerBuilder connectOptions(Object connectOptions) {
- doSetProperty("connectOptions", connectOptions);
+ default PahoEndpointProducerBuilder connectionTimeout(
+ int connectionTimeout) {
+ doSetProperty("connectionTimeout", connectionTimeout);
return this;
}
/**
- * Client connection options.
+ * Sets the connection timeout value. This value, measured in seconds,
+ * defines the maximum time interval the client will wait for the
+ * network connection to the MQTT server to be established. The default
+ * timeout is 30 seconds. A value of 0 disables timeout processing
+ * meaning the client will wait until the network connection is made
+ * successfully or fails.
*
- * The option will be converted to a
- * <code>org.eclipse.paho.client.mqttv3.MqttConnectOptions</code> type.
+ * The option will be converted to a <code>int</code> type.
*
* Group: common
*/
- default PahoEndpointProducerBuilder connectOptions(String connectOptions) {
- doSetProperty("connectOptions", connectOptions);
+ default PahoEndpointProducerBuilder connectionTimeout(
+ String connectionTimeout) {
+ doSetProperty("connectionTimeout", connectionTimeout);
return this;
}
/**
@@ -493,20 +1114,136 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * Client persistence to be used - memory or file.
+ * Sets the keep alive interval. This value, measured in seconds,
+ * defines the maximum time interval between messages sent or received.
+ * It enables the client to detect if the server is no longer available,
+ * without having to wait for the TCP/IP timeout. The client will ensure
+ * that at least one message travels across the network within each keep
+ * alive period. In the absence of a data-related message during the
+ * time period, the client sends a very small ping message, which the
+ * server will acknowledge. A value of 0 disables keepalive processing
+ * in the client. The default value is 60 seconds.
*
- * The option is a:
- * <code>org.apache.camel.component.paho.PahoPersistence</code> type.
+ * The option is a: <code>int</code> type.
*
* Group: common
*/
- default PahoEndpointProducerBuilder persistence(
- PahoPersistence persistence) {
- doSetProperty("persistence", persistence);
+ default PahoEndpointProducerBuilder keepAliveInterval(
+ int keepAliveInterval) {
+ doSetProperty("keepAliveInterval", keepAliveInterval);
return this;
}
/**
- * Client persistence to be used - memory or file.
+ * Sets the keep alive interval. This value, measured in seconds,
+ * defines the maximum time interval between messages sent or received.
+ * It enables the client to detect if the server is no longer available,
+ * without having to wait for the TCP/IP timeout. The client will ensure
+ * that at least one message travels across the network within each keep
+ * alive period. In the absence of a data-related message during the
+ * time period, the client sends a very small ping message, which the
+ * server will acknowledge. A value of 0 disables keepalive processing
+ * in the client. The default value is 60 seconds.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder keepAliveInterval(
+ String keepAliveInterval) {
+ doSetProperty("keepAliveInterval", keepAliveInterval);
+ return this;
+ }
+ /**
+ * Sets the max inflight. please increase this value in a high traffic
+ * environment. The default value is 10.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder maxInflight(int maxInflight) {
+ doSetProperty("maxInflight", maxInflight);
+ return this;
+ }
+ /**
+ * Sets the max inflight. please increase this value in a high traffic
+ * environment. The default value is 10.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder maxInflight(String maxInflight) {
+ doSetProperty("maxInflight", maxInflight);
+ return this;
+ }
+ /**
+ * Get the maximum time (in millis) to wait between reconnects.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder maxReconnectDelay(
+ int maxReconnectDelay) {
+ doSetProperty("maxReconnectDelay", maxReconnectDelay);
+ return this;
+ }
+ /**
+ * Get the maximum time (in millis) to wait between reconnects.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder maxReconnectDelay(
+ String maxReconnectDelay) {
+ doSetProperty("maxReconnectDelay", maxReconnectDelay);
+ return this;
+ }
+ /**
+ * Sets the MQTT version. The default action is to connect with version
+ * 3.1.1, and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1
+ * can be selected specifically, with no fall back, by using the
+ * MQTT_VERSION_3_1_1 or MQTT_VERSION_3_1 options respectively.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder mqttVersion(int mqttVersion) {
+ doSetProperty("mqttVersion", mqttVersion);
+ return this;
+ }
+ /**
+ * Sets the MQTT version. The default action is to connect with version
+ * 3.1.1, and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1
+ * can be selected specifically, with no fall back, by using the
+ * MQTT_VERSION_3_1_1 or MQTT_VERSION_3_1 options respectively.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder mqttVersion(String mqttVersion) {
+ doSetProperty("mqttVersion", mqttVersion);
+ return this;
+ }
+ /**
+ * Client persistence to be used - memory or file.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.paho.PahoPersistence</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder persistence(
+ PahoPersistence persistence) {
+ doSetProperty("persistence", persistence);
+ return this;
+ }
+ /**
+ * Client persistence to be used - memory or file.
*
* The option will be converted to a
* <code>org.apache.camel.component.paho.PahoPersistence</code> type.
@@ -562,6 +1299,137 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Set a list of one or more serverURIs the client may connect to.
+ * Multiple servers can be separated by comma. Each serverURI specifies
+ * the address of a server that the client may connect to. Two types of
+ * connection are supported tcp:// for a TCP connection and ssl:// for a
+ * TCP connection secured by SSL/TLS. For example: tcp://localhost:1883
+ * ssl://localhost:8883 If the port is not specified, it will default to
+ * 1883 for tcp:// URIs, and 8883 for ssl:// URIs. If serverURIs is set
+ * then it overrides the serverURI parameter passed in on the
+ * constructor of the MQTT client. When an attempt to connect is
+ * initiated the client will start with the first serverURI in the list
+ * and work through the list until a connection is established with a
+ * server. If a connection cannot be made to any of the servers then the
+ * connect attempt fails. Specifying a list of servers that a client may
+ * connect to has several uses: High Availability and reliable message
+ * delivery Some MQTT servers support a high availability feature where
+ * two or more equal MQTT servers share state. An MQTT client can
+ * connect to any of the equal servers and be assured that messages are
+ * reliably delivered and durable subscriptions are maintained no matter
+ * which server the client connects to. The cleansession flag must be
+ * set to false if durable subscriptions and/or reliable message
+ * delivery is required. Hunt List A set of servers may be specified
+ * that are not equal (as in the high availability option). As no state
+ * is shared across the servers reliable message delivery and durable
+ * subscriptions are not valid. The cleansession flag must be set to
+ * true if the hunt list mode is used.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder serverURIs(String serverURIs) {
+ doSetProperty("serverURIs", serverURIs);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder willPayload(String willPayload) {
+ doSetProperty("willPayload", willPayload);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder willQos(int willQos) {
+ doSetProperty("willQos", willQos);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder willQos(String willQos) {
+ doSetProperty("willQos", willQos);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder willRetained(boolean willRetained) {
+ doSetProperty("willRetained", willRetained);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder willRetained(String willRetained) {
+ doSetProperty("willRetained", willRetained);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointProducerBuilder willTopic(String willTopic) {
+ doSetProperty("willTopic", willTopic);
+ return this;
+ }
+ /**
* Whether the producer should be started lazy (on the first message).
* By starting lazy you can use this to allow CamelContext and routes to
* startup in situations where a producer may otherwise fail during
@@ -602,6 +1470,32 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Whether SSL HostnameVerifier is enabled or not. The default value is
+ * true.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointProducerBuilder httpsHostnameVerificationEnabled(
+ boolean httpsHostnameVerificationEnabled) {
+ doSetProperty("httpsHostnameVerificationEnabled", httpsHostnameVerificationEnabled);
+ return this;
+ }
+ /**
+ * Whether SSL HostnameVerifier is enabled or not. The default value is
+ * true.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointProducerBuilder httpsHostnameVerificationEnabled(
+ String httpsHostnameVerificationEnabled) {
+ doSetProperty("httpsHostnameVerificationEnabled", httpsHostnameVerificationEnabled);
+ return this;
+ }
+ /**
* Password to be used for authentication against the MQTT broker.
*
* The option is a: <code>java.lang.String</code> type.
@@ -613,6 +1507,166 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Sets the SocketFactory to use. This allows an application to apply
+ * its own policies around the creation of network sockets. If using an
+ * SSL connection, an SSLSocketFactory can be used to supply
+ * application-specific security settings.
+ *
+ * The option is a: <code>javax.net.SocketFactory</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointProducerBuilder socketFactory(Object socketFactory) {
+ doSetProperty("socketFactory", socketFactory);
+ return this;
+ }
+ /**
+ * Sets the SocketFactory to use. This allows an application to apply
+ * its own policies around the creation of network sockets. If using an
+ * SSL connection, an SSLSocketFactory can be used to supply
+ * application-specific security settings.
+ *
+ * The option will be converted to a
+ * <code>javax.net.SocketFactory</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointProducerBuilder socketFactory(String socketFactory) {
+ doSetProperty("socketFactory", socketFactory);
+ return this;
+ }
+ /**
+ * Sets the SSL properties for the connection. Note that these
+ * properties are only valid if an implementation of the Java Secure
+ * Socket Extensions (JSSE) is available. These properties are not used
+ * if a custom SocketFactory has been set. The following properties can
+ * be used: com.ibm.ssl.protocol One of: SSL, SSLv3, TLS, TLSv1,
+ * SSL_TLS. com.ibm.ssl.contextProvider Underlying JSSE provider. For
+ * example IBMJSSE2 or SunJSSE com.ibm.ssl.keyStore The name of the file
+ * that contains the KeyStore object that you want the KeyManager to
+ * use. For example /mydir/etc/key.p12 com.ibm.ssl.keyStorePassword The
+ * password for the KeyStore object that you want the KeyManager to use.
+ * The password can either be in plain-text, or may be obfuscated using
+ * the static method: com.ibm.micro.security.Password.obfuscate(char
+ * password). This obfuscates the password using a simple and insecure
+ * XOR and Base64 encoding mechanism. Note that this is only a simple
+ * scrambler to obfuscate clear-text passwords. com.ibm.ssl.keyStoreType
+ * Type of key store, for example PKCS12, JKS, or JCEKS.
+ * com.ibm.ssl.keyStoreProvider Key store provider, for example IBMJCE
+ * or IBMJCEFIPS. com.ibm.ssl.trustStore The name of the file that
+ * contains the KeyStore object that you want the TrustManager to use.
+ * com.ibm.ssl.trustStorePassword The password for the TrustStore object
+ * that you want the TrustManager to use. The password can either be in
+ * plain-text, or may be obfuscated using the static method:
+ * com.ibm.micro.security.Password.obfuscate(char password). This
+ * obfuscates the password using a simple and insecure XOR and Base64
+ * encoding mechanism. Note that this is only a simple scrambler to
+ * obfuscate clear-text passwords. com.ibm.ssl.trustStoreType The type
+ * of KeyStore object that you want the default TrustManager to use.
+ * Same possible values as keyStoreType. com.ibm.ssl.trustStoreProvider
+ * Trust store provider, for example IBMJCE or IBMJCEFIPS.
+ * com.ibm.ssl.enabledCipherSuites A list of which ciphers are enabled.
+ * Values are dependent on the provider, for example:
+ * SSL_RSA_WITH_AES_128_CBC_SHA;SSL_RSA_WITH_3DES_EDE_CBC_SHA.
+ * com.ibm.ssl.keyManager Sets the algorithm that will be used to
+ * instantiate a KeyManagerFactory object instead of using the default
+ * algorithm available in the platform. Example values: IbmX509 or
+ * IBMJ9X509. com.ibm.ssl.trustManager Sets the algorithm that will be
+ * used to instantiate a TrustManagerFactory object instead of using the
+ * default algorithm available in the platform. Example values: PKIX or
+ * IBMJ9X509.
+ *
+ * The option is a: <code>java.util.Properties</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointProducerBuilder sslClientProps(
+ Properties sslClientProps) {
+ doSetProperty("sslClientProps", sslClientProps);
+ return this;
+ }
+ /**
+ * Sets the SSL properties for the connection. Note that these
+ * properties are only valid if an implementation of the Java Secure
+ * Socket Extensions (JSSE) is available. These properties are not used
+ * if a custom SocketFactory has been set. The following properties can
+ * be used: com.ibm.ssl.protocol One of: SSL, SSLv3, TLS, TLSv1,
+ * SSL_TLS. com.ibm.ssl.contextProvider Underlying JSSE provider. For
+ * example IBMJSSE2 or SunJSSE com.ibm.ssl.keyStore The name of the file
+ * that contains the KeyStore object that you want the KeyManager to
+ * use. For example /mydir/etc/key.p12 com.ibm.ssl.keyStorePassword The
+ * password for the KeyStore object that you want the KeyManager to use.
+ * The password can either be in plain-text, or may be obfuscated using
+ * the static method: com.ibm.micro.security.Password.obfuscate(char
+ * password). This obfuscates the password using a simple and insecure
+ * XOR and Base64 encoding mechanism. Note that this is only a simple
+ * scrambler to obfuscate clear-text passwords. com.ibm.ssl.keyStoreType
+ * Type of key store, for example PKCS12, JKS, or JCEKS.
+ * com.ibm.ssl.keyStoreProvider Key store provider, for example IBMJCE
+ * or IBMJCEFIPS. com.ibm.ssl.trustStore The name of the file that
+ * contains the KeyStore object that you want the TrustManager to use.
+ * com.ibm.ssl.trustStorePassword The password for the TrustStore object
+ * that you want the TrustManager to use. The password can either be in
+ * plain-text, or may be obfuscated using the static method:
+ * com.ibm.micro.security.Password.obfuscate(char password). This
+ * obfuscates the password using a simple and insecure XOR and Base64
+ * encoding mechanism. Note that this is only a simple scrambler to
+ * obfuscate clear-text passwords. com.ibm.ssl.trustStoreType The type
+ * of KeyStore object that you want the default TrustManager to use.
+ * Same possible values as keyStoreType. com.ibm.ssl.trustStoreProvider
+ * Trust store provider, for example IBMJCE or IBMJCEFIPS.
+ * com.ibm.ssl.enabledCipherSuites A list of which ciphers are enabled.
+ * Values are dependent on the provider, for example:
+ * SSL_RSA_WITH_AES_128_CBC_SHA;SSL_RSA_WITH_3DES_EDE_CBC_SHA.
+ * com.ibm.ssl.keyManager Sets the algorithm that will be used to
+ * instantiate a KeyManagerFactory object instead of using the default
+ * algorithm available in the platform. Example values: IbmX509 or
+ * IBMJ9X509. com.ibm.ssl.trustManager Sets the algorithm that will be
+ * used to instantiate a TrustManagerFactory object instead of using the
+ * default algorithm available in the platform. Example values: PKIX or
+ * IBMJ9X509.
+ *
+ * The option will be converted to a <code>java.util.Properties</code>
+ * type.
+ *
+ * Group: security
+ */
+ default PahoEndpointProducerBuilder sslClientProps(String sslClientProps) {
+ doSetProperty("sslClientProps", sslClientProps);
+ return this;
+ }
+ /**
+ * Sets the HostnameVerifier for the SSL connection. Note that it will
+ * be used after handshake on a connection and you should do actions by
+ * yourself when hostname is verified error. There is no default
+ * HostnameVerifier.
+ *
+ * The option is a: <code>javax.net.ssl.HostnameVerifier</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointProducerBuilder sslHostnameVerifier(
+ Object sslHostnameVerifier) {
+ doSetProperty("sslHostnameVerifier", sslHostnameVerifier);
+ return this;
+ }
+ /**
+ * Sets the HostnameVerifier for the SSL connection. Note that it will
+ * be used after handshake on a connection and you should do actions by
+ * yourself when hostname is verified error. There is no default
+ * HostnameVerifier.
+ *
+ * The option will be converted to a
+ * <code>javax.net.ssl.HostnameVerifier</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointProducerBuilder sslHostnameVerifier(
+ String sslHostnameVerifier) {
+ doSetProperty("sslHostnameVerifier", sslHostnameVerifier);
+ return this;
+ }
+ /**
* Username to be used for authentication against the MQTT broker.
*
* The option is a: <code>java.lang.String</code> type.
@@ -661,29 +1715,80 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * Define if you don't want to resolve the MQTT Connect Options from
- * registry.
+ * To use an existing mqtt client.
*
- * The option is a: <code>boolean</code> type.
+ * The option is a:
+ * <code>org.eclipse.paho.client.mqttv3.MqttClient</code> type.
*
* Group: advanced
*/
- default AdvancedPahoEndpointProducerBuilder resolveMqttConnectOptions(
- boolean resolveMqttConnectOptions) {
- doSetProperty("resolveMqttConnectOptions", resolveMqttConnectOptions);
+ default AdvancedPahoEndpointProducerBuilder client(Object client) {
+ doSetProperty("client", client);
return this;
}
/**
- * Define if you don't want to resolve the MQTT Connect Options from
- * registry.
+ * To use an existing mqtt client.
*
- * The option will be converted to a <code>boolean</code> type.
+ * The option will be converted to a
+ * <code>org.eclipse.paho.client.mqttv3.MqttClient</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointProducerBuilder client(String client) {
+ doSetProperty("client", client);
+ return this;
+ }
+ /**
+ * Sets the Custom WebSocket Headers for the WebSocket Connection.
+ *
+ * The option is a: <code>java.util.Properties</code> type.
*
* Group: advanced
*/
- default AdvancedPahoEndpointProducerBuilder resolveMqttConnectOptions(
- String resolveMqttConnectOptions) {
- doSetProperty("resolveMqttConnectOptions", resolveMqttConnectOptions);
+ default AdvancedPahoEndpointProducerBuilder customWebSocketHeaders(
+ Properties customWebSocketHeaders) {
+ doSetProperty("customWebSocketHeaders", customWebSocketHeaders);
+ return this;
+ }
+ /**
+ * Sets the Custom WebSocket Headers for the WebSocket Connection.
+ *
+ * The option will be converted to a <code>java.util.Properties</code>
+ * type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointProducerBuilder customWebSocketHeaders(
+ String customWebSocketHeaders) {
+ doSetProperty("customWebSocketHeaders", customWebSocketHeaders);
+ return this;
+ }
+ /**
+ * Set the time in seconds that the executor service should wait when
+ * terminating before forcefully terminating. It is not recommended to
+ * change this value unless you are absolutely sure that you need to.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointProducerBuilder executorServiceTimeout(
+ int executorServiceTimeout) {
+ doSetProperty("executorServiceTimeout", executorServiceTimeout);
+ return this;
+ }
+ /**
+ * Set the time in seconds that the executor service should wait when
+ * terminating before forcefully terminating. It is not recommended to
+ * change this value unless you are absolutely sure that you need to.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointProducerBuilder executorServiceTimeout(
+ String executorServiceTimeout) {
+ doSetProperty("executorServiceTimeout", executorServiceTimeout);
return this;
}
/**
@@ -724,27 +1829,42 @@ public interface PahoEndpointBuilderFactory {
return (AdvancedPahoEndpointBuilder) this;
}
/**
- * Client will automatically attempt to reconnect to the server if the
- * connection is lost.
+ * Sets whether the client will automatically attempt to reconnect to
+ * the server if the connection is lost. If set to false, the client
+ * will not attempt to automatically reconnect to the server in the
+ * event that the connection is lost. If set to true, in the event that
+ * the connection is lost, the client will attempt to reconnect to the
+ * server. It will initially wait 1 second before it attempts to
+ * reconnect, for every failed reconnect attempt, the delay will double
+ * until it is at 2 minutes at which point the delay will stay at 2
+ * minutes.
*
* The option is a: <code>boolean</code> type.
*
* Group: common
*/
- default PahoEndpointBuilder autoReconnect(boolean autoReconnect) {
- doSetProperty("autoReconnect", autoReconnect);
+ default PahoEndpointBuilder automaticReconnect(
+ boolean automaticReconnect) {
+ doSetProperty("automaticReconnect", automaticReconnect);
return this;
}
/**
- * Client will automatically attempt to reconnect to the server if the
- * connection is lost.
+ * Sets whether the client will automatically attempt to reconnect to
+ * the server if the connection is lost. If set to false, the client
+ * will not attempt to automatically reconnect to the server in the
+ * event that the connection is lost. If set to true, in the event that
+ * the connection is lost, the client will attempt to reconnect to the
+ * server. It will initially wait 1 second before it attempts to
+ * reconnect, for every failed reconnect attempt, the delay will double
+ * until it is at 2 minutes at which point the delay will stay at 2
+ * minutes.
*
* The option will be converted to a <code>boolean</code> type.
*
* Group: common
*/
- default PahoEndpointBuilder autoReconnect(String autoReconnect) {
- doSetProperty("autoReconnect", autoReconnect);
+ default PahoEndpointBuilder automaticReconnect(String automaticReconnect) {
+ doSetProperty("automaticReconnect", automaticReconnect);
return this;
}
/**
@@ -759,7 +1879,49 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * MQTT client identifier.
+ * Sets whether the client and server should remember state across
+ * restarts and reconnects. If set to false both the client and server
+ * will maintain state across restarts of the client, the server and the
+ * connection. As state is maintained: Message delivery will be reliable
+ * meeting the specified QOS even if the client, server or connection
+ * are restarted. The server will treat a subscription as durable. If
+ * set to true the client and server will not maintain state across
+ * restarts of the client, the server or the connection. This means
+ * Message delivery to the specified QOS cannot be maintained if the
+ * client, server or connection are restarted The server will treat a
+ * subscription as non-durable.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder cleanSession(boolean cleanSession) {
+ doSetProperty("cleanSession", cleanSession);
+ return this;
+ }
+ /**
+ * Sets whether the client and server should remember state across
+ * restarts and reconnects. If set to false both the client and server
+ * will maintain state across restarts of the client, the server and the
+ * connection. As state is maintained: Message delivery will be reliable
+ * meeting the specified QOS even if the client, server or connection
+ * are restarted. The server will treat a subscription as durable. If
+ * set to true the client and server will not maintain state across
+ * restarts of the client, the server or the connection. This means
+ * Message delivery to the specified QOS cannot be maintained if the
+ * client, server or connection are restarted The server will treat a
+ * subscription as non-durable.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder cleanSession(String cleanSession) {
+ doSetProperty("cleanSession", cleanSession);
+ return this;
+ }
+ /**
+ * MQTT client identifier. The identifier must be unique.
*
* The option is a: <code>java.lang.String</code> type.
*
@@ -770,27 +1932,35 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * Client connection options.
+ * Sets the connection timeout value. This value, measured in seconds,
+ * defines the maximum time interval the client will wait for the
+ * network connection to the MQTT server to be established. The default
+ * timeout is 30 seconds. A value of 0 disables timeout processing
+ * meaning the client will wait until the network connection is made
+ * successfully or fails.
*
- * The option is a:
- * <code>org.eclipse.paho.client.mqttv3.MqttConnectOptions</code> type.
+ * The option is a: <code>int</code> type.
*
* Group: common
*/
- default PahoEndpointBuilder connectOptions(Object connectOptions) {
- doSetProperty("connectOptions", connectOptions);
+ default PahoEndpointBuilder connectionTimeout(int connectionTimeout) {
+ doSetProperty("connectionTimeout", connectionTimeout);
return this;
}
/**
- * Client connection options.
+ * Sets the connection timeout value. This value, measured in seconds,
+ * defines the maximum time interval the client will wait for the
+ * network connection to the MQTT server to be established. The default
+ * timeout is 30 seconds. A value of 0 disables timeout processing
+ * meaning the client will wait until the network connection is made
+ * successfully or fails.
*
- * The option will be converted to a
- * <code>org.eclipse.paho.client.mqttv3.MqttConnectOptions</code> type.
+ * The option will be converted to a <code>int</code> type.
*
* Group: common
*/
- default PahoEndpointBuilder connectOptions(String connectOptions) {
- doSetProperty("connectOptions", connectOptions);
+ default PahoEndpointBuilder connectionTimeout(String connectionTimeout) {
+ doSetProperty("connectionTimeout", connectionTimeout);
return this;
}
/**
@@ -807,6 +1977,118 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Sets the keep alive interval. This value, measured in seconds,
+ * defines the maximum time interval between messages sent or received.
+ * It enables the client to detect if the server is no longer available,
+ * without having to wait for the TCP/IP timeout. The client will ensure
+ * that at least one message travels across the network within each keep
+ * alive period. In the absence of a data-related message during the
+ * time period, the client sends a very small ping message, which the
+ * server will acknowledge. A value of 0 disables keepalive processing
+ * in the client. The default value is 60 seconds.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder keepAliveInterval(int keepAliveInterval) {
+ doSetProperty("keepAliveInterval", keepAliveInterval);
+ return this;
+ }
+ /**
+ * Sets the keep alive interval. This value, measured in seconds,
+ * defines the maximum time interval between messages sent or received.
+ * It enables the client to detect if the server is no longer available,
+ * without having to wait for the TCP/IP timeout. The client will ensure
+ * that at least one message travels across the network within each keep
+ * alive period. In the absence of a data-related message during the
+ * time period, the client sends a very small ping message, which the
+ * server will acknowledge. A value of 0 disables keepalive processing
+ * in the client. The default value is 60 seconds.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder keepAliveInterval(String keepAliveInterval) {
+ doSetProperty("keepAliveInterval", keepAliveInterval);
+ return this;
+ }
+ /**
+ * Sets the max inflight. please increase this value in a high traffic
+ * environment. The default value is 10.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder maxInflight(int maxInflight) {
+ doSetProperty("maxInflight", maxInflight);
+ return this;
+ }
+ /**
+ * Sets the max inflight. please increase this value in a high traffic
+ * environment. The default value is 10.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder maxInflight(String maxInflight) {
+ doSetProperty("maxInflight", maxInflight);
+ return this;
+ }
+ /**
+ * Get the maximum time (in millis) to wait between reconnects.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder maxReconnectDelay(int maxReconnectDelay) {
+ doSetProperty("maxReconnectDelay", maxReconnectDelay);
+ return this;
+ }
+ /**
+ * Get the maximum time (in millis) to wait between reconnects.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder maxReconnectDelay(String maxReconnectDelay) {
+ doSetProperty("maxReconnectDelay", maxReconnectDelay);
+ return this;
+ }
+ /**
+ * Sets the MQTT version. The default action is to connect with version
+ * 3.1.1, and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1
+ * can be selected specifically, with no fall back, by using the
+ * MQTT_VERSION_3_1_1 or MQTT_VERSION_3_1 options respectively.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder mqttVersion(int mqttVersion) {
+ doSetProperty("mqttVersion", mqttVersion);
+ return this;
+ }
+ /**
+ * Sets the MQTT version. The default action is to connect with version
+ * 3.1.1, and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1
+ * can be selected specifically, with no fall back, by using the
+ * MQTT_VERSION_3_1_1 or MQTT_VERSION_3_1 options respectively.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder mqttVersion(String mqttVersion) {
+ doSetProperty("mqttVersion", mqttVersion);
+ return this;
+ }
+ /**
* Client persistence to be used - memory or file.
*
* The option is a:
@@ -875,6 +2157,163 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Set a list of one or more serverURIs the client may connect to.
+ * Multiple servers can be separated by comma. Each serverURI specifies
+ * the address of a server that the client may connect to. Two types of
+ * connection are supported tcp:// for a TCP connection and ssl:// for a
+ * TCP connection secured by SSL/TLS. For example: tcp://localhost:1883
+ * ssl://localhost:8883 If the port is not specified, it will default to
+ * 1883 for tcp:// URIs, and 8883 for ssl:// URIs. If serverURIs is set
+ * then it overrides the serverURI parameter passed in on the
+ * constructor of the MQTT client. When an attempt to connect is
+ * initiated the client will start with the first serverURI in the list
+ * and work through the list until a connection is established with a
+ * server. If a connection cannot be made to any of the servers then the
+ * connect attempt fails. Specifying a list of servers that a client may
+ * connect to has several uses: High Availability and reliable message
+ * delivery Some MQTT servers support a high availability feature where
+ * two or more equal MQTT servers share state. An MQTT client can
+ * connect to any of the equal servers and be assured that messages are
+ * reliably delivered and durable subscriptions are maintained no matter
+ * which server the client connects to. The cleansession flag must be
+ * set to false if durable subscriptions and/or reliable message
+ * delivery is required. Hunt List A set of servers may be specified
+ * that are not equal (as in the high availability option). As no state
+ * is shared across the servers reliable message delivery and durable
+ * subscriptions are not valid. The cleansession flag must be set to
+ * true if the hunt list mode is used.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder serverURIs(String serverURIs) {
+ doSetProperty("serverURIs", serverURIs);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder willPayload(String willPayload) {
+ doSetProperty("willPayload", willPayload);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder willQos(int willQos) {
+ doSetProperty("willQos", willQos);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder willQos(String willQos) {
+ doSetProperty("willQos", willQos);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder willRetained(boolean willRetained) {
+ doSetProperty("willRetained", willRetained);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder willRetained(String willRetained) {
+ doSetProperty("willRetained", willRetained);
+ return this;
+ }
+ /**
+ * Sets the Last Will and Testament (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default PahoEndpointBuilder willTopic(String willTopic) {
+ doSetProperty("willTopic", willTopic);
+ return this;
+ }
+ /**
+ * Whether SSL HostnameVerifier is enabled or not. The default value is
+ * true.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointBuilder httpsHostnameVerificationEnabled(
+ boolean httpsHostnameVerificationEnabled) {
+ doSetProperty("httpsHostnameVerificationEnabled", httpsHostnameVerificationEnabled);
+ return this;
+ }
+ /**
+ * Whether SSL HostnameVerifier is enabled or not. The default value is
+ * true.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointBuilder httpsHostnameVerificationEnabled(
+ String httpsHostnameVerificationEnabled) {
+ doSetProperty("httpsHostnameVerificationEnabled", httpsHostnameVerificationEnabled);
+ return this;
+ }
+ /**
* Password to be used for authentication against the MQTT broker.
*
* The option is a: <code>java.lang.String</code> type.
@@ -886,6 +2325,165 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
+ * Sets the SocketFactory to use. This allows an application to apply
+ * its own policies around the creation of network sockets. If using an
+ * SSL connection, an SSLSocketFactory can be used to supply
+ * application-specific security settings.
+ *
+ * The option is a: <code>javax.net.SocketFactory</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointBuilder socketFactory(Object socketFactory) {
+ doSetProperty("socketFactory", socketFactory);
+ return this;
+ }
+ /**
+ * Sets the SocketFactory to use. This allows an application to apply
+ * its own policies around the creation of network sockets. If using an
+ * SSL connection, an SSLSocketFactory can be used to supply
+ * application-specific security settings.
+ *
+ * The option will be converted to a
+ * <code>javax.net.SocketFactory</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointBuilder socketFactory(String socketFactory) {
+ doSetProperty("socketFactory", socketFactory);
+ return this;
+ }
+ /**
+ * Sets the SSL properties for the connection. Note that these
+ * properties are only valid if an implementation of the Java Secure
+ * Socket Extensions (JSSE) is available. These properties are not used
+ * if a custom SocketFactory has been set. The following properties can
+ * be used: com.ibm.ssl.protocol One of: SSL, SSLv3, TLS, TLSv1,
+ * SSL_TLS. com.ibm.ssl.contextProvider Underlying JSSE provider. For
+ * example IBMJSSE2 or SunJSSE com.ibm.ssl.keyStore The name of the file
+ * that contains the KeyStore object that you want the KeyManager to
+ * use. For example /mydir/etc/key.p12 com.ibm.ssl.keyStorePassword The
+ * password for the KeyStore object that you want the KeyManager to use.
+ * The password can either be in plain-text, or may be obfuscated using
+ * the static method: com.ibm.micro.security.Password.obfuscate(char
+ * password). This obfuscates the password using a simple and insecure
+ * XOR and Base64 encoding mechanism. Note that this is only a simple
+ * scrambler to obfuscate clear-text passwords. com.ibm.ssl.keyStoreType
+ * Type of key store, for example PKCS12, JKS, or JCEKS.
+ * com.ibm.ssl.keyStoreProvider Key store provider, for example IBMJCE
+ * or IBMJCEFIPS. com.ibm.ssl.trustStore The name of the file that
+ * contains the KeyStore object that you want the TrustManager to use.
+ * com.ibm.ssl.trustStorePassword The password for the TrustStore object
+ * that you want the TrustManager to use. The password can either be in
+ * plain-text, or may be obfuscated using the static method:
+ * com.ibm.micro.security.Password.obfuscate(char password). This
+ * obfuscates the password using a simple and insecure XOR and Base64
+ * encoding mechanism. Note that this is only a simple scrambler to
+ * obfuscate clear-text passwords. com.ibm.ssl.trustStoreType The type
+ * of KeyStore object that you want the default TrustManager to use.
+ * Same possible values as keyStoreType. com.ibm.ssl.trustStoreProvider
+ * Trust store provider, for example IBMJCE or IBMJCEFIPS.
+ * com.ibm.ssl.enabledCipherSuites A list of which ciphers are enabled.
+ * Values are dependent on the provider, for example:
+ * SSL_RSA_WITH_AES_128_CBC_SHA;SSL_RSA_WITH_3DES_EDE_CBC_SHA.
+ * com.ibm.ssl.keyManager Sets the algorithm that will be used to
+ * instantiate a KeyManagerFactory object instead of using the default
+ * algorithm available in the platform. Example values: IbmX509 or
+ * IBMJ9X509. com.ibm.ssl.trustManager Sets the algorithm that will be
+ * used to instantiate a TrustManagerFactory object instead of using the
+ * default algorithm available in the platform. Example values: PKIX or
+ * IBMJ9X509.
+ *
+ * The option is a: <code>java.util.Properties</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointBuilder sslClientProps(Properties sslClientProps) {
+ doSetProperty("sslClientProps", sslClientProps);
+ return this;
+ }
+ /**
+ * Sets the SSL properties for the connection. Note that these
+ * properties are only valid if an implementation of the Java Secure
+ * Socket Extensions (JSSE) is available. These properties are not used
+ * if a custom SocketFactory has been set. The following properties can
+ * be used: com.ibm.ssl.protocol One of: SSL, SSLv3, TLS, TLSv1,
+ * SSL_TLS. com.ibm.ssl.contextProvider Underlying JSSE provider. For
+ * example IBMJSSE2 or SunJSSE com.ibm.ssl.keyStore The name of the file
+ * that contains the KeyStore object that you want the KeyManager to
+ * use. For example /mydir/etc/key.p12 com.ibm.ssl.keyStorePassword The
+ * password for the KeyStore object that you want the KeyManager to use.
+ * The password can either be in plain-text, or may be obfuscated using
+ * the static method: com.ibm.micro.security.Password.obfuscate(char
+ * password). This obfuscates the password using a simple and insecure
+ * XOR and Base64 encoding mechanism. Note that this is only a simple
+ * scrambler to obfuscate clear-text passwords. com.ibm.ssl.keyStoreType
+ * Type of key store, for example PKCS12, JKS, or JCEKS.
+ * com.ibm.ssl.keyStoreProvider Key store provider, for example IBMJCE
+ * or IBMJCEFIPS. com.ibm.ssl.trustStore The name of the file that
+ * contains the KeyStore object that you want the TrustManager to use.
+ * com.ibm.ssl.trustStorePassword The password for the TrustStore object
+ * that you want the TrustManager to use. The password can either be in
+ * plain-text, or may be obfuscated using the static method:
+ * com.ibm.micro.security.Password.obfuscate(char password). This
+ * obfuscates the password using a simple and insecure XOR and Base64
+ * encoding mechanism. Note that this is only a simple scrambler to
+ * obfuscate clear-text passwords. com.ibm.ssl.trustStoreType The type
+ * of KeyStore object that you want the default TrustManager to use.
+ * Same possible values as keyStoreType. com.ibm.ssl.trustStoreProvider
+ * Trust store provider, for example IBMJCE or IBMJCEFIPS.
+ * com.ibm.ssl.enabledCipherSuites A list of which ciphers are enabled.
+ * Values are dependent on the provider, for example:
+ * SSL_RSA_WITH_AES_128_CBC_SHA;SSL_RSA_WITH_3DES_EDE_CBC_SHA.
+ * com.ibm.ssl.keyManager Sets the algorithm that will be used to
+ * instantiate a KeyManagerFactory object instead of using the default
+ * algorithm available in the platform. Example values: IbmX509 or
+ * IBMJ9X509. com.ibm.ssl.trustManager Sets the algorithm that will be
+ * used to instantiate a TrustManagerFactory object instead of using the
+ * default algorithm available in the platform. Example values: PKIX or
+ * IBMJ9X509.
+ *
+ * The option will be converted to a <code>java.util.Properties</code>
+ * type.
+ *
+ * Group: security
+ */
+ default PahoEndpointBuilder sslClientProps(String sslClientProps) {
+ doSetProperty("sslClientProps", sslClientProps);
+ return this;
+ }
+ /**
+ * Sets the HostnameVerifier for the SSL connection. Note that it will
+ * be used after handshake on a connection and you should do actions by
+ * yourself when hostname is verified error. There is no default
+ * HostnameVerifier.
+ *
+ * The option is a: <code>javax.net.ssl.HostnameVerifier</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointBuilder sslHostnameVerifier(
+ Object sslHostnameVerifier) {
+ doSetProperty("sslHostnameVerifier", sslHostnameVerifier);
+ return this;
+ }
+ /**
+ * Sets the HostnameVerifier for the SSL connection. Note that it will
+ * be used after handshake on a connection and you should do actions by
+ * yourself when hostname is verified error. There is no default
+ * HostnameVerifier.
+ *
+ * The option will be converted to a
+ * <code>javax.net.ssl.HostnameVerifier</code> type.
+ *
+ * Group: security
+ */
+ default PahoEndpointBuilder sslHostnameVerifier(
+ String sslHostnameVerifier) {
+ doSetProperty("sslHostnameVerifier", sslHostnameVerifier);
+ return this;
+ }
+ /**
* Username to be used for authentication against the MQTT broker.
*
* The option is a: <code>java.lang.String</code> type.
@@ -934,29 +2532,80 @@ public interface PahoEndpointBuilderFactory {
return this;
}
/**
- * Define if you don't want to resolve the MQTT Connect Options from
- * registry.
+ * To use an existing mqtt client.
*
- * The option is a: <code>boolean</code> type.
+ * The option is a:
+ * <code>org.eclipse.paho.client.mqttv3.MqttClient</code> type.
*
* Group: advanced
*/
- default AdvancedPahoEndpointBuilder resolveMqttConnectOptions(
- boolean resolveMqttConnectOptions) {
- doSetProperty("resolveMqttConnectOptions", resolveMqttConnectOptions);
+ default AdvancedPahoEndpointBuilder client(Object client) {
+ doSetProperty("client", client);
return this;
}
/**
- * Define if you don't want to resolve the MQTT Connect Options from
- * registry.
+ * To use an existing mqtt client.
*
- * The option will be converted to a <code>boolean</code> type.
+ * The option will be converted to a
+ * <code>org.eclipse.paho.client.mqttv3.MqttClient</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointBuilder client(String client) {
+ doSetProperty("client", client);
+ return this;
+ }
+ /**
+ * Sets the Custom WebSocket Headers for the WebSocket Connection.
+ *
+ * The option is a: <code>java.util.Properties</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointBuilder customWebSocketHeaders(
+ Properties customWebSocketHeaders) {
+ doSetProperty("customWebSocketHeaders", customWebSocketHeaders);
+ return this;
+ }
+ /**
+ * Sets the Custom WebSocket Headers for the WebSocket Connection.
+ *
+ * The option will be converted to a <code>java.util.Properties</code>
+ * type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointBuilder customWebSocketHeaders(
+ String customWebSocketHeaders) {
+ doSetProperty("customWebSocketHeaders", customWebSocketHeaders);
+ return this;
+ }
+ /**
+ * Set the time in seconds that the executor service should wait when
+ * terminating before forcefully terminating. It is not recommended to
+ * change this value unless you are absolutely sure that you need to.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedPahoEndpointBuilder executorServiceTimeout(
+ int executorServiceTimeout) {
+ doSetProperty("executorServiceTimeout", executorServiceTimeout);
+ return this;
+ }
+ /**
+ * Set the time in seconds that the executor service should wait when
+ * terminating before forcefully terminating. It is not recommended to
+ * change this value unless you are absolutely sure that you need to.
+ *
+ * The option will be converted to a <code>int</code> type.
*
* Group: advanced
*/
- default AdvancedPahoEndpointBuilder resolveMqttConnectOptions(
- String resolveMqttConnectOptions) {
- doSetProperty("resolveMqttConnectOptions", resolveMqttConnectOptions);
+ default AdvancedPahoEndpointBuilder executorServiceTimeout(
+ String executorServiceTimeout) {
+ doSetProperty("executorServiceTimeout", executorServiceTimeout);
return this;
}
/**
@@ -995,8 +2644,8 @@ public interface PahoEndpointBuilderFactory {
}
/**
* Paho (camel-paho)
- * Component for communicating with MQTT M2M message brokers using Eclipse
- * Paho MQTT Client.
+ * Component for communicating with MQTT message brokers using Eclipse Paho
+ * MQTT Client.
*
* Category: messaging,iot
* Available as of version: 2.16
diff --git a/platforms/spring-boot/components-starter/camel-paho-starter/src/main/java/org/apache/camel/component/paho/springboot/PahoComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-paho-starter/src/main/java/org/apache/camel/component/paho/springboot/PahoComponentConfiguration.java
index c0f237a..6b6b4bd 100644
--- a/platforms/spring-boot/components-starter/camel-paho-starter/src/main/java/org/apache/camel/component/paho/springboot/PahoComponentConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-paho-starter/src/main/java/org/apache/camel/component/paho/springboot/PahoComponentConfiguration.java
@@ -16,13 +16,17 @@
*/
package org.apache.camel.component.paho.springboot;
+import java.util.Properties;
import javax.annotation.Generated;
+import javax.net.SocketFactory;
+import javax.net.ssl.HostnameVerifier;
+import org.apache.camel.component.paho.PahoPersistence;
import org.apache.camel.spring.boot.ComponentConfigurationPropertiesCommon;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
- * Component for communicating with MQTT M2M message brokers using Eclipse Paho
- * MQTT Client.
+ * Component for communicating with MQTT message brokers using Eclipse Paho MQTT
+ * Client.
*
* Generated by camel-package-maven-plugin - do not edit this file!
*/
@@ -38,77 +42,510 @@ public class PahoComponentConfiguration
*/
private Boolean enabled;
/**
- * The URL of the MQTT broker.
- */
- private String brokerUrl;
- /**
- * MQTT client identifier.
- */
- private String clientId;
- /**
- * Username to be used for authentication against the MQTT broker
+ * To use the shared Paho configuration
*/
- private String userName;
+ private PahoConfigurationNestedConfiguration configuration;
/**
- * Password to be used for authentication against the MQTT broker
+ * To use a shared Paho client. The option is a
+ * org.eclipse.paho.client.mqttv3.MqttClient type.
*/
- private String password;
+ private String client;
/**
- * Client connection options. The option is a
- * org.eclipse.paho.client.mqttv3.MqttConnectOptions type.
+ * The URL of the MQTT broker.
*/
- private String connectOptions;
+ private String brokerUrl;
/**
* Whether the component should use basic property binding (Camel 2.x) or
* the newer property binding with additional capabilities
*/
private Boolean basicPropertyBinding = false;
- public String getBrokerUrl() {
- return brokerUrl;
+ public PahoConfigurationNestedConfiguration getConfiguration() {
+ return configuration;
}
- public void setBrokerUrl(String brokerUrl) {
- this.brokerUrl = brokerUrl;
+ public void setConfiguration(
+ PahoConfigurationNestedConfiguration configuration) {
+ this.configuration = configuration;
}
- public String getClientId() {
- return clientId;
+ public String getClient() {
+ return client;
}
- public void setClientId(String clientId) {
- this.clientId = clientId;
+ public void setClient(String client) {
+ this.client = client;
}
- public String getUserName() {
- return userName;
+ public String getBrokerUrl() {
+ return brokerUrl;
}
- public void setUserName(String userName) {
- this.userName = userName;
+ public void setBrokerUrl(String brokerUrl) {
+ this.brokerUrl = brokerUrl;
}
- public String getPassword() {
- return password;
+ public Boolean getBasicPropertyBinding() {
+ return basicPropertyBinding;
}
- public void setPassword(String password) {
- this.password = password;
+ public void setBasicPropertyBinding(Boolean basicPropertyBinding) {
+ this.basicPropertyBinding = basicPropertyBinding;
}
- public String getConnectOptions() {
- return connectOptions;
- }
+ public static class PahoConfigurationNestedConfiguration {
+ public static final Class CAMEL_NESTED_CLASS = org.apache.camel.component.paho.PahoConfiguration.class;
+ /**
+ * MQTT client identifier. The identifier must be unique.
+ */
+ private String clientId;
+ /**
+ * The URL of the MQTT broker.
+ */
+ private String brokerUrl = "tcp://localhost:1883";
+ /**
+ * Client quality of service level (0-2).
+ */
+ private Integer qos = 2;
+ /**
+ * Retain option
+ */
+ private Boolean retained = false;
+ /**
+ * Client persistence to be used - memory or file.
+ */
+ private PahoPersistence persistence = PahoPersistence.MEMORY;
+ /**
+ * Base directory used by file persistence. Will by default use user
+ * directory.
+ */
+ private String filePersistenceDirectory;
+ /**
+ * Username to be used for authentication against the MQTT broker
+ */
+ private String userName;
+ /**
+ * Password to be used for authentication against the MQTT broker
+ */
+ private String password;
+ /**
+ * Sets the keep alive interval. This value, measured in seconds,
+ * defines the maximum time interval between messages sent or received.
+ * It enables the client to detect if the server is no longer available,
+ * without having to wait for the TCP/IP timeout. The client will ensure
+ * that at least one message travels across the network within each keep
+ * alive period. In the absence of a data-related message during the
+ * time period, the client sends a very small ping message, which the
+ * server will acknowledge. A value of 0 disables keepalive processing
+ * in the client. <p> The default value is 60 seconds </p>
+ */
+ private Integer keepAliveInterval = 60;
+ /**
+ * Sets the max inflight. please increase this value in a high traffic
+ * environment. <p> The default value is 10 </p>
+ */
+ private Integer maxInflight = 10;
+ /**
+ * Sets the "Last Will and Testament" (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ */
+ private String willTopic;
+ /**
+ * Sets the "Last Will and Testament" (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ */
+ private String willPayload;
+ /**
+ * Sets the "Last Will and Testament" (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ */
+ private Integer willQos;
+ /**
+ * Sets the "Last Will and Testament" (LWT) for the connection. In the
+ * event that this client unexpectedly loses its connection to the
+ * server, the server will publish a message to itself using the
+ * supplied details. The topic to publish to The byte payload for the
+ * message. The quality of service to publish the message at (0, 1 or
+ * 2). Whether or not the message should be retained.
+ */
+ private Boolean willRetained = false;
+ /**
+ * Sets the SocketFactory to use. This allows an application to apply
+ * its own policies around the creation of network sockets. If using an
+ * SSL connection, an SSLSocketFactory can be used to supply
+ * application-specific security settings.
+ */
+ private SocketFactory socketFactory;
+ /**
+ * Sets the SSL properties for the connection. <p> Note that these
+ * properties are only valid if an implementation of the Java Secure
+ * Socket Extensions (JSSE) is available. These properties are
+ * <em>not</em> used if a custom SocketFactory has been set. The
+ * following properties can be used: </p> <dl>
+ * <dt>com.ibm.ssl.protocol</dt> <dd>One of: SSL, SSLv3, TLS, TLSv1,
+ * SSL_TLS.</dd> <dt>com.ibm.ssl.contextProvider <dd>Underlying JSSE
+ * provider. For example "IBMJSSE2" or "SunJSSE"</dd>
+ * <dt>com.ibm.ssl.keyStore</dt> <dd>The name of the file that contains
+ * the KeyStore object that you want the KeyManager to use. For example
+ * /mydir/etc/key.p12</dd> <dt>com.ibm.ssl.keyStorePassword</dt> <dd>The
+ * password for the KeyStore object that you want the KeyManager to use.
+ * The password can either be in plain-text, or may be obfuscated using
+ * the static method:
+ * <code>com.ibm.micro.security.Password.obfuscate(char[]
+ * password)</code>. This obfuscates the password using a simple and
+ * insecure XOR and Base64 encoding mechanism. Note that this is only a
+ * simple scrambler to obfuscate clear-text passwords.</dd>
+ * <dt>com.ibm.ssl.keyStoreType</dt> <dd>Type of key store, for example
+ * "PKCS12", "JKS", or "JCEKS".</dd>
+ * <dt>com.ibm.ssl.keyStoreProvider</dt> <dd>Key store provider, for
+ * example "IBMJCE" or "IBMJCEFIPS".</dd>
+ * <dt>com.ibm.ssl.trustStore</dt> <dd>The name of the file that
+ * contains the KeyStore object that you want the TrustManager to
+ * use.</dd> <dt>com.ibm.ssl.trustStorePassword</dt> <dd>The password
+ * for the TrustStore object that you want the TrustManager to use. The
+ * password can either be in plain-text, or may be obfuscated using the
+ * static method: <code>com.ibm.micro.security.Password.obfuscate(char[]
+ * password)</code>. This obfuscates the password using a simple and
+ * insecure XOR and Base64 encoding mechanism. Note that this is only a
+ * simple scrambler to obfuscate clear-text passwords.</dd>
+ * <dt>com.ibm.ssl.trustStoreType</dt> <dd>The type of KeyStore object
+ * that you want the default TrustManager to use. Same possible values
+ * as "keyStoreType".</dd> <dt>com.ibm.ssl.trustStoreProvider</dt>
+ * <dd>Trust store provider, for example "IBMJCE" or "IBMJCEFIPS".</dd>
+ * <dt>com.ibm.ssl.enabledCipherSuites</dt> <dd>A list of which ciphers
+ * are enabled. Values are dependent on the provider, for example:
+ * SSL_RSA_WITH_AES_128_CBC_SHA;SSL_RSA_WITH_3DES_EDE_CBC_SHA.</dd>
+ * <dt>com.ibm.ssl.keyManager</dt> <dd>Sets the algorithm that will be
+ * used to instantiate a KeyManagerFactory object instead of using the
+ * default algorithm available in the platform. Example values:
+ * "IbmX509" or "IBMJ9X509".</dd> <dt>com.ibm.ssl.trustManager</dt>
+ * <dd>Sets the algorithm that will be used to instantiate a
+ * TrustManagerFactory object instead of using the default algorithm
+ * available in the platform. Example values: "PKIX" or
+ * "IBMJ9X509".</dd> </dl>
+ */
+ private Properties sslClientProps;
+ /**
+ * Whether SSL HostnameVerifier is enabled or not. The default value is
+ * true.
+ */
+ private Boolean httpsHostnameVerificationEnabled = true;
+ /**
+ * Sets the HostnameVerifier for the SSL connection. Note that it will
+ * be used after handshake on a connection and you should do actions by
+ * yourself when hostname is verified error. <p> There is no default
+ * HostnameVerifier </p>
+ */
+ private HostnameVerifier sslHostnameVerifier;
+ /**
+ * Sets whether the client and server should remember state across
+ * restarts and reconnects. <ul> <li>If set to false both the client and
+ * server will maintain state across restarts of the client, the server
+ * and the connection. As state is maintained: <ul> <li>Message delivery
+ * will be reliable meeting the specified QOS even if the client, server
+ * or connection are restarted. <li>The server will treat a subscription
+ * as durable. </ul> <li>If set to true the client and server will not
+ * maintain state across restarts of the client, the server or the
+ * connection. This means <ul> <li>Message delivery to the specified QOS
+ * cannot be maintained if the client, server or connection are
+ * restarted <li>The server will treat a subscription as non-durable
+ * </ul> </ul>
+ */
+ private Boolean cleanSession = true;
+ /**
+ * Sets the connection timeout value. This value, measured in seconds,
+ * defines the maximum time interval the client will wait for the
+ * network connection to the MQTT server to be established. The default
+ * timeout is 30 seconds. A value of 0 disables timeout processing
+ * meaning the client will wait until the network connection is made
+ * successfully or fails.
+ */
+ private Integer connectionTimeout = 30;
+ /**
+ * Set a list of one or more serverURIs the client may connect to.
+ * Multiple servers can be separated by comma. <p> Each
+ * <code>serverURI</code> specifies the address of a server that the
+ * client may connect to. Two types of connection are supported
+ * <code>tcp://</code> for a TCP connection and <code>ssl://</code> for
+ * a TCP connection secured by SSL/TLS. For example: <ul>
+ * <li><code>tcp://localhost:1883</code></li>
+ * <li><code>ssl://localhost:8883</code></li> </ul> If the port is not
+ * specified, it will default to 1883 for <code>tcp://</code>" URIs, and
+ * 8883 for <code>ssl://</code> URIs. <p> If serverURIs is set then it
+ * overrides the serverURI parameter passed in on the constructor of the
+ * MQTT client. <p> When an attempt to connect is initiated the client
+ * will start with the first serverURI in the list and work through the
+ * list until a connection is established with a server. If a connection
+ * cannot be made to any of the servers then the connect attempt fails.
+ * <p> Specifying a list of servers that a client may connect to has
+ * several uses: <ol> <li>High Availability and reliable message
+ * delivery <p> Some MQTT servers support a high availability feature
+ * where two or more "equal" MQTT servers share state. An MQTT client
+ * can connect to any of the "equal" servers and be assured that
+ * messages are reliably delivered and durable subscriptions are
+ * maintained no matter which server the client connects to. </p> <p>
+ * The cleansession flag must be set to false if durable subscriptions
+ * and/or reliable message delivery is required. </p> </li> <li>Hunt
+ * List <p> A set of servers may be specified that are not "equal" (as
+ * in the high availability option). As no state is shared across the
+ * servers reliable message delivery and durable subscriptions are not
+ * valid. The cleansession flag must be set to true if the hunt list
+ * mode is used </p> </li> </ol>
+ */
+ private String serverURIs;
+ /**
+ * Sets the MQTT version. The default action is to connect with version
+ * 3.1.1, and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1
+ * can be selected specifically, with no fall back, by using the
+ * MQTT_VERSION_3_1_1 or MQTT_VERSION_3_1 options respectively.
+ */
+ private Integer mqttVersion;
+ /**
+ * Sets whether the client will automatically attempt to reconnect to
+ * the server if the connection is lost. <ul> <li>If set to false, the
+ * client will not attempt to automatically reconnect to the server in
+ * the event that the connection is lost.</li> <li>If set to true, in
+ * the event that the connection is lost, the client will attempt to
+ * reconnect to the server. It will initially wait 1 second before it
+ * attempts to reconnect, for every failed reconnect attempt, the delay
+ * will double until it is at 2 minutes at which point the delay will
+ * stay at 2 minutes.</li> </ul>
+ */
+ private Boolean automaticReconnect = true;
+ /**
+ * Get the maximum time (in millis) to wait between reconnects
+ */
+ private Integer maxReconnectDelay = 128000;
+ /**
+ * Sets the Custom WebSocket Headers for the WebSocket Connection.
+ */
+ private Properties customWebSocketHeaders;
+ /**
+ * Set the time in seconds that the executor service should wait when
+ * terminating before forcefully terminating. It is not recommended to
+ * change this value unless you are absolutely sure that you need to.
+ */
+ private Integer executorServiceTimeout = 1;
- public void setConnectOptions(String connectOptions) {
- this.connectOptions = connectOptions;
- }
+ public String getClientId() {
+ return clientId;
+ }
- public Boolean getBasicPropertyBinding() {
- return basicPropertyBinding;
- }
+ public void setClientId(String clientId) {
+ this.clientId = clientId;
+ }
- public void setBasicPropertyBinding(Boolean basicPropertyBinding) {
- this.basicPropertyBinding = basicPropertyBinding;
+ public String getBrokerUrl() {
+ return brokerUrl;
+ }
+
+ public void setBrokerUrl(String brokerUrl) {
+ this.brokerUrl = brokerUrl;
+ }
+
+ public Integer getQos() {
+ return qos;
+ }
+
+ public void setQos(Integer qos) {
+ this.qos = qos;
+ }
+
+ public Boolean getRetained() {
+ return retained;
+ }
+
+ public void setRetained(Boolean retained) {
+ this.retained = retained;
+ }
+
+ public PahoPersistence getPersistence() {
+ return persistence;
+ }
+
+ public void setPersistence(PahoPersistence persistence) {
+ this.persistence = persistence;
+ }
+
+ public String getFilePersistenceDirectory() {
+ return filePersistenceDirectory;
+ }
+
+ public void setFilePersistenceDirectory(String filePersistenceDirectory) {
+ this.filePersistenceDirectory = filePersistenceDirectory;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ public void setUserName(String userName) {
+ this.userName = userName;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public Integer getKeepAliveInterval() {
+ return keepAliveInterval;
+ }
+
+ public void setKeepAliveInterval(Integer keepAliveInterval) {
+ this.keepAliveInterval = keepAliveInterval;
+ }
+
+ public Integer getMaxInflight() {
+ return maxInflight;
+ }
+
+ public void setMaxInflight(Integer maxInflight) {
+ this.maxInflight = maxInflight;
+ }
+
+ public String getWillTopic() {
+ return willTopic;
+ }
+
+ public void setWillTopic(String willTopic) {
+ this.willTopic = willTopic;
+ }
+
+ public String getWillPayload() {
+ return willPayload;
+ }
+
+ public void setWillPayload(String willPayload) {
+ this.willPayload = willPayload;
+ }
+
+ public Integer getWillQos() {
+ return willQos;
+ }
+
+ public void setWillQos(Integer willQos) {
+ this.willQos = willQos;
+ }
+
+ public Boolean getWillRetained() {
+ return willRetained;
+ }
+
+ public void setWillRetained(Boolean willRetained) {
+ this.willRetained = willRetained;
+ }
+
+ public SocketFactory getSocketFactory() {
+ return socketFactory;
+ }
+
+ public void setSocketFactory(SocketFactory socketFactory) {
+ this.socketFactory = socketFactory;
+ }
+
+ public Properties getSslClientProps() {
+ return sslClientProps;
+ }
+
+ public void setSslClientProps(Properties sslClientProps) {
+ this.sslClientProps = sslClientProps;
+ }
+
+ public Boolean getHttpsHostnameVerificationEnabled() {
+ return httpsHostnameVerificationEnabled;
+ }
+
+ public void setHttpsHostnameVerificationEnabled(
+ Boolean httpsHostnameVerificationEnabled) {
+ this.httpsHostnameVerificationEnabled = httpsHostnameVerificationEnabled;
+ }
+
+ public HostnameVerifier getSslHostnameVerifier() {
+ return sslHostnameVerifier;
+ }
+
+ public void setSslHostnameVerifier(HostnameVerifier sslHostnameVerifier) {
+ this.sslHostnameVerifier = sslHostnameVerifier;
+ }
+
+ public Boolean getCleanSession() {
+ return cleanSession;
+ }
+
+ public void setCleanSession(Boolean cleanSession) {
+ this.cleanSession = cleanSession;
+ }
+
+ public Integer getConnectionTimeout() {
+ return connectionTimeout;
+ }
+
+ public void setConnectionTimeout(Integer connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ }
+
+ public String getServerURIs() {
+ return serverURIs;
+ }
+
+ public void setServerURIs(String serverURIs) {
+ this.serverURIs = serverURIs;
+ }
+
+ public Integer getMqttVersion() {
+ return mqttVersion;
+ }
+
+ public void setMqttVersion(Integer mqttVersion) {
+ this.mqttVersion = mqttVersion;
+ }
+
+ public Boolean getAutomaticReconnect() {
+ return automaticReconnect;
+ }
+
+ public void setAutomaticReconnect(Boolean automaticReconnect) {
+ this.automaticReconnect = automaticReconnect;
+ }
+
+ public Integer getMaxReconnectDelay() {
+ return maxReconnectDelay;
+ }
+
+ public void setMaxReconnectDelay(Integer maxReconnectDelay) {
+ this.maxReconnectDelay = maxReconnectDelay;
+ }
+
+ public Properties getCustomWebSocketHeaders() {
+ return customWebSocketHeaders;
+ }
+
+ public void setCustomWebSocketHeaders(Properties customWebSocketHeaders) {
+ this.customWebSocketHeaders = customWebSocketHeaders;
+ }
+
+ public Integer getExecutorServiceTimeout() {
+ return executorServiceTimeout;
+ }
+
+ public void setExecutorServiceTimeout(Integer executorServiceTimeout) {
+ this.executorServiceTimeout = executorServiceTimeout;
+ }
}
}
\ No newline at end of file
[camel] 01/02: CAMEL-14052: camel-paho make it possible to
configure all its options via compoent options that also works for Spring
Boot
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 65885bee95091034b3625690f2ef1bebf649089a
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri Oct 11 10:52:03 2019 +0200
CAMEL-14052: camel-paho make it possible to configure all its options via compoent options that also works for Spring Boot
---
.../apache/camel/component/paho/PahoComponent.java | 91 +---
.../camel/component/paho/PahoConfiguration.java | 581 +++++++++++++++++++++
.../apache/camel/component/paho/PahoConsumer.java | 4 +-
.../apache/camel/component/paho/PahoEndpoint.java | 264 +++-------
.../apache/camel/component/paho/PahoProducer.java | 4 +-
.../camel/component/paho/PahoComponentTest.java | 34 +-
.../component/paho/PahoOverrideTopicTest.java | 5 -
7 files changed, 675 insertions(+), 308 deletions(-)
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoComponent.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoComponent.java
index d045bbc..30f43ce 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoComponent.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoComponent.java
@@ -23,7 +23,7 @@ import org.apache.camel.Endpoint;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.annotations.Component;
import org.apache.camel.support.DefaultComponent;
-import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttClient;
/**
* Component to integrate with the Eclipse Paho MQTT library.
@@ -31,15 +31,11 @@ import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
@Component("paho")
public class PahoComponent extends DefaultComponent {
- private String brokerUrl;
- private String clientId;
- @Metadata(label = "security", secret = true)
- private String userName;
- @Metadata(label = "security", secret = true)
- private String password;
+ private PahoConfiguration configuration = new PahoConfiguration();
+
@Metadata(label = "advanced")
- private MqttConnectOptions connectOptions;
-
+ private MqttClient client;
+
public PahoComponent() {
this(null);
}
@@ -52,83 +48,48 @@ public class PahoComponent extends DefaultComponent {
@Override
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
- PahoEndpoint answer = new PahoEndpoint(uri, remaining, this);
+ // Each endpoint can have its own configuration so make
+ // a copy of the configuration
+ PahoConfiguration configuration = getConfiguration().copy();
- if (brokerUrl != null) {
- answer.setBrokerUrl(brokerUrl);
- }
- if (clientId != null) {
- answer.setClientId(clientId);
- }
- if (userName != null) {
- answer.setUserName(userName);
- }
- if (password != null) {
- answer.setPassword(password);
- }
- if (connectOptions != null) {
- answer.setConnectOptions(connectOptions);
- }
+ PahoEndpoint answer = new PahoEndpoint(uri, remaining, this, configuration);
+ answer.setClient(client);
+ setProperties(configuration, parameters);
setProperties(answer, parameters);
return answer;
}
- // Getters and setters
-
- public String getBrokerUrl() {
- return brokerUrl;
+ public PahoConfiguration getConfiguration() {
+ return configuration;
}
/**
- * The URL of the MQTT broker.
+ * To use the shared Paho configuration
*/
- public void setBrokerUrl(String brokerUrl) {
- this.brokerUrl = brokerUrl;
+ public void setConfiguration(PahoConfiguration configuration) {
+ this.configuration = configuration;
}
- public String getClientId() {
- return clientId;
+ public MqttClient getClient() {
+ return client;
}
/**
- * MQTT client identifier.
+ * To use a shared Paho client
*/
- public void setClientId(String clientId) {
- this.clientId = clientId;
- }
-
- public String getUserName() {
- return userName;
+ public void setClient(MqttClient client) {
+ this.client = client;
}
- /**
- * Username to be used for authentication against the MQTT broker
- */
- public void setUserName(String userName) {
- this.userName = userName;
- }
-
- public String getPassword() {
- return password;
- }
-
- /**
- * Password to be used for authentication against the MQTT broker
- */
- public void setPassword(String password) {
- this.password = password;
- }
-
- public MqttConnectOptions getConnectOptions() {
- return connectOptions;
+ public String getBrokerUrl() {
+ return configuration.getBrokerUrl();
}
/**
- * Client connection options
+ * The URL of the MQTT broker.
*/
- public void setConnectOptions(MqttConnectOptions connectOptions) {
- this.connectOptions = connectOptions;
+ public void setBrokerUrl(String brokerUrl) {
+ configuration.setBrokerUrl(brokerUrl);
}
-
}
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConfiguration.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConfiguration.java
new file mode 100644
index 0000000..96eb341
--- /dev/null
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConfiguration.java
@@ -0,0 +1,581 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.paho;
+
+import java.util.Properties;
+import javax.net.SocketFactory;
+import javax.net.ssl.HostnameVerifier;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+@UriParams
+public class PahoConfiguration implements Cloneable {
+
+ @UriParam
+ private String clientId = "camel-" + System.nanoTime();
+ @UriParam(defaultValue = PahoConstants.DEFAULT_BROKER_URL)
+ private String brokerUrl = PahoConstants.DEFAULT_BROKER_URL;
+ @UriParam(defaultValue = "2")
+ private int qos = PahoConstants.DEFAULT_QOS;
+ @UriParam
+ private boolean retained;
+ @UriParam(defaultValue = "MEMORY")
+ private PahoPersistence persistence = PahoPersistence.MEMORY;
+ @UriParam
+ private String filePersistenceDirectory;
+
+ @UriParam(defaultValue = "60")
+ private int keepAliveInterval = 60;
+ @UriParam(defaultValue = "10")
+ private int maxInflight = 10;
+ @UriParam
+ private String willTopic;
+ @UriParam
+ private String willPayload;
+ @UriParam
+ private int willQos;
+ @UriParam
+ private boolean willRetained;
+ @UriParam(label = "security") @Metadata(secret = true)
+ private String userName;
+ @UriParam(label = "security") @Metadata(secret = true)
+ private String password;
+ @UriParam(label = "security")
+ private SocketFactory socketFactory;
+ @UriParam(label = "security")
+ private Properties sslClientProps;
+ @UriParam(label = "security", defaultValue = "true")
+ private boolean httpsHostnameVerificationEnabled = true;
+ @UriParam(label = "security")
+ private HostnameVerifier sslHostnameVerifier;
+ @UriParam(defaultValue = "true")
+ private boolean cleanSession = true;
+ @UriParam(defaultValue = "30")
+ private int connectionTimeout = 30;
+ @UriParam
+ private String serverURIs;
+ @UriParam
+ private int mqttVersion;
+ @UriParam(defaultValue = "true")
+ private boolean automaticReconnect = true;
+ @UriParam(defaultValue = "128000")
+ private int maxReconnectDelay = 128000;
+ @UriParam(label = "advanced")
+ private Properties customWebSocketHeaders;
+ @UriParam(label = "advanced", defaultValue = "1")
+ private int executorServiceTimeout = 1;
+
+ public String getClientId() {
+ return clientId;
+ }
+
+ /**
+ * MQTT client identifier.
+ */
+ public void setClientId(String clientId) {
+ this.clientId = clientId;
+ }
+
+ public String getBrokerUrl() {
+ return brokerUrl;
+ }
+
+ /**
+ * The URL of the MQTT broker.
+ */
+ public void setBrokerUrl(String brokerUrl) {
+ this.brokerUrl = brokerUrl;
+ }
+
+ public int getQos() {
+ return qos;
+ }
+
+ /**
+ * Client quality of service level (0-2).
+ */
+ public void setQos(int qos) {
+ this.qos = qos;
+ }
+
+ public boolean isRetained() {
+ return retained;
+ }
+
+ /**
+ * Retain option
+ */
+ public void setRetained(boolean retained) {
+ this.retained = retained;
+ }
+
+ public PahoPersistence getPersistence() {
+ return persistence;
+ }
+
+ /**
+ * Client persistence to be used - memory or file.
+ */
+ public void setPersistence(PahoPersistence persistence) {
+ this.persistence = persistence;
+ }
+
+ public String getFilePersistenceDirectory() {
+ return filePersistenceDirectory;
+ }
+
+ /**
+ * Base directory used by file persistence. Will by default use user directory.
+ */
+ public void setFilePersistenceDirectory(String filePersistenceDirectory) {
+ this.filePersistenceDirectory = filePersistenceDirectory;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ /**
+ * Username to be used for authentication against the MQTT broker
+ */
+ public void setUserName(String userName) {
+ this.userName = userName;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ /**
+ * Password to be used for authentication against the MQTT broker
+ */
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public int getKeepAliveInterval() {
+ return keepAliveInterval;
+ }
+
+ /**
+ * Sets the keep alive interval. This value, measured in seconds, defines the
+ * maximum time interval between messages sent or received. It enables the
+ * client to detect if the server is no longer available, without having to wait
+ * for the TCP/IP timeout. The client will ensure that at least one message
+ * travels across the network within each keep alive period. In the absence of a
+ * data-related message during the time period, the client sends a very small
+ * ping message, which the server will acknowledge. A value of 0 disables
+ * keepalive processing in the client.
+ * <p>
+ * The default value is 60 seconds
+ * </p>
+ */
+ public void setKeepAliveInterval(int keepAliveInterval) {
+ this.keepAliveInterval = keepAliveInterval;
+ }
+
+ public int getMaxInflight() {
+ return maxInflight;
+ }
+
+ /**
+ * Sets the max inflight. please increase this value in a high traffic
+ * environment.
+ * <p>
+ * The default value is 10
+ * </p>
+ */
+ public void setMaxInflight(int maxInflight) {
+ this.maxInflight = maxInflight;
+ }
+
+ public String getWillTopic() {
+ return willTopic;
+ }
+
+ /**
+ * Sets the "Last Will and Testament" (LWT) for the connection. In the event
+ * that this client unexpectedly loses its connection to the server, the server
+ * will publish a message to itself using the supplied details.
+ *
+ * The topic to publish to
+ * The byte payload for the message.
+ * The quality of service to publish the message at (0, 1 or 2).
+ * Whether or not the message should be retained.
+ */
+ public void setWillTopic(String willTopic) {
+ this.willTopic = willTopic;
+ }
+
+ public String getWillPayload() {
+ return willPayload;
+ }
+
+ /**
+ * Sets the "Last Will and Testament" (LWT) for the connection. In the event
+ * that this client unexpectedly loses its connection to the server, the server
+ * will publish a message to itself using the supplied details.
+ *
+ * The topic to publish to
+ * The byte payload for the message.
+ * The quality of service to publish the message at (0, 1 or 2).
+ * Whether or not the message should be retained.
+ */
+ public void setWillPayload(String willPayload) {
+ this.willPayload = willPayload;
+ }
+
+ public int getWillQos() {
+ return willQos;
+ }
+
+ /**
+ * Sets the "Last Will and Testament" (LWT) for the connection. In the event
+ * that this client unexpectedly loses its connection to the server, the server
+ * will publish a message to itself using the supplied details.
+ *
+ * The topic to publish to
+ * The byte payload for the message.
+ * The quality of service to publish the message at (0, 1 or 2).
+ * Whether or not the message should be retained.
+ */
+ public void setWillQos(int willQos) {
+ this.willQos = willQos;
+ }
+
+ public boolean isWillRetained() {
+ return willRetained;
+ }
+
+ /**
+ * Sets the "Last Will and Testament" (LWT) for the connection. In the event
+ * that this client unexpectedly loses its connection to the server, the server
+ * will publish a message to itself using the supplied details.
+ *
+ * The topic to publish to
+ * The byte payload for the message.
+ * The quality of service to publish the message at (0, 1 or 2).
+ * Whether or not the message should be retained.
+ */
+ public void setWillRetained(boolean willRetained) {
+ this.willRetained = willRetained;
+ }
+
+ public SocketFactory getSocketFactory() {
+ return socketFactory;
+ }
+
+ /**
+ * Sets the SocketFactory to use. This allows an application to
+ * apply its own policies around the creation of network sockets. If using an
+ * SSL connection, an SSLSocketFactory can be used to supply
+ * application-specific security settings.
+ */
+ public void setSocketFactory(SocketFactory socketFactory) {
+ this.socketFactory = socketFactory;
+ }
+
+ public Properties getSslClientProps() {
+ return sslClientProps;
+ }
+
+ /**
+ * Sets the SSL properties for the connection.
+ * <p>
+ * Note that these properties are only valid if an implementation of the Java
+ * Secure Socket Extensions (JSSE) is available. These properties are
+ * <em>not</em> used if a custom SocketFactory has been set.
+ *
+ * The following properties can be used:
+ * </p>
+ * <dl>
+ * <dt>com.ibm.ssl.protocol</dt>
+ * <dd>One of: SSL, SSLv3, TLS, TLSv1, SSL_TLS.</dd>
+ * <dt>com.ibm.ssl.contextProvider
+ * <dd>Underlying JSSE provider. For example "IBMJSSE2" or "SunJSSE"</dd>
+ *
+ * <dt>com.ibm.ssl.keyStore</dt>
+ * <dd>The name of the file that contains the KeyStore object that you want the
+ * KeyManager to use. For example /mydir/etc/key.p12</dd>
+ *
+ * <dt>com.ibm.ssl.keyStorePassword</dt>
+ * <dd>The password for the KeyStore object that you want the KeyManager to use.
+ * The password can either be in plain-text, or may be obfuscated using the
+ * static method:
+ * <code>com.ibm.micro.security.Password.obfuscate(char[] password)</code>. This
+ * obfuscates the password using a simple and insecure XOR and Base64 encoding
+ * mechanism. Note that this is only a simple scrambler to obfuscate clear-text
+ * passwords.</dd>
+ *
+ * <dt>com.ibm.ssl.keyStoreType</dt>
+ * <dd>Type of key store, for example "PKCS12", "JKS", or "JCEKS".</dd>
+ *
+ * <dt>com.ibm.ssl.keyStoreProvider</dt>
+ * <dd>Key store provider, for example "IBMJCE" or "IBMJCEFIPS".</dd>
+ *
+ * <dt>com.ibm.ssl.trustStore</dt>
+ * <dd>The name of the file that contains the KeyStore object that you want the
+ * TrustManager to use.</dd>
+ *
+ * <dt>com.ibm.ssl.trustStorePassword</dt>
+ * <dd>The password for the TrustStore object that you want the TrustManager to
+ * use. The password can either be in plain-text, or may be obfuscated using the
+ * static method:
+ * <code>com.ibm.micro.security.Password.obfuscate(char[] password)</code>. This
+ * obfuscates the password using a simple and insecure XOR and Base64 encoding
+ * mechanism. Note that this is only a simple scrambler to obfuscate clear-text
+ * passwords.</dd>
+ *
+ * <dt>com.ibm.ssl.trustStoreType</dt>
+ * <dd>The type of KeyStore object that you want the default TrustManager to
+ * use. Same possible values as "keyStoreType".</dd>
+ *
+ * <dt>com.ibm.ssl.trustStoreProvider</dt>
+ * <dd>Trust store provider, for example "IBMJCE" or "IBMJCEFIPS".</dd>
+ *
+ * <dt>com.ibm.ssl.enabledCipherSuites</dt>
+ * <dd>A list of which ciphers are enabled. Values are dependent on the
+ * provider, for example:
+ * SSL_RSA_WITH_AES_128_CBC_SHA;SSL_RSA_WITH_3DES_EDE_CBC_SHA.</dd>
+ *
+ * <dt>com.ibm.ssl.keyManager</dt>
+ * <dd>Sets the algorithm that will be used to instantiate a KeyManagerFactory
+ * object instead of using the default algorithm available in the platform.
+ * Example values: "IbmX509" or "IBMJ9X509".</dd>
+ *
+ * <dt>com.ibm.ssl.trustManager</dt>
+ * <dd>Sets the algorithm that will be used to instantiate a TrustManagerFactory
+ * object instead of using the default algorithm available in the platform.
+ * Example values: "PKIX" or "IBMJ9X509".</dd>
+ * </dl>
+ */
+ public void setSslClientProps(Properties sslClientProps) {
+ this.sslClientProps = sslClientProps;
+ }
+
+ public boolean isHttpsHostnameVerificationEnabled() {
+ return httpsHostnameVerificationEnabled;
+ }
+
+ /**
+ * Whether SSL HostnameVerifier is enabled or not.
+ * The default value is true.
+ */
+ public void setHttpsHostnameVerificationEnabled(boolean httpsHostnameVerificationEnabled) {
+ this.httpsHostnameVerificationEnabled = httpsHostnameVerificationEnabled;
+ }
+
+ public HostnameVerifier getSslHostnameVerifier() {
+ return sslHostnameVerifier;
+ }
+
+ /**
+ * Sets the HostnameVerifier for the SSL connection. Note that it will be used
+ * after handshake on a connection and you should do actions by yourself when
+ * hostname is verified error.
+ * <p>
+ * There is no default HostnameVerifier
+ * </p>
+ */
+ public void setSslHostnameVerifier(HostnameVerifier sslHostnameVerifier) {
+ this.sslHostnameVerifier = sslHostnameVerifier;
+ }
+
+ public boolean isCleanSession() {
+ return cleanSession;
+ }
+
+ /**
+ * Sets whether the client and server should remember state across restarts and
+ * reconnects.
+ * <ul>
+ * <li>If set to false both the client and server will maintain state across
+ * restarts of the client, the server and the connection. As state is
+ * maintained:
+ * <ul>
+ * <li>Message delivery will be reliable meeting the specified QOS even if the
+ * client, server or connection are restarted.
+ * <li>The server will treat a subscription as durable.
+ * </ul>
+ * <li>If set to true the client and server will not maintain state across
+ * restarts of the client, the server or the connection. This means
+ * <ul>
+ * <li>Message delivery to the specified QOS cannot be maintained if the client,
+ * server or connection are restarted
+ * <li>The server will treat a subscription as non-durable
+ * </ul>
+ * </ul>
+ */
+ public void setCleanSession(boolean cleanSession) {
+ this.cleanSession = cleanSession;
+ }
+
+ public int getConnectionTimeout() {
+ return connectionTimeout;
+ }
+
+ /**
+ * Sets the connection timeout value. This value, measured in seconds, defines
+ * the maximum time interval the client will wait for the network connection to
+ * the MQTT server to be established. The default timeout is 30 seconds. A value
+ * of 0 disables timeout processing meaning the client will wait until the
+ * network connection is made successfully or fails.
+ */
+ public void setConnectionTimeout(int connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ }
+
+ public String getServerURIs() {
+ return serverURIs;
+ }
+
+ /**
+ * Set a list of one or more serverURIs the client may connect to.
+ * Multiple servers can be separated by comma.
+ * <p>
+ * Each <code>serverURI</code> specifies the address of a server that the client
+ * may connect to. Two types of connection are supported <code>tcp://</code> for
+ * a TCP connection and <code>ssl://</code> for a TCP connection secured by
+ * SSL/TLS. For example:
+ * <ul>
+ * <li><code>tcp://localhost:1883</code></li>
+ * <li><code>ssl://localhost:8883</code></li>
+ * </ul>
+ * If the port is not specified, it will default to 1883 for
+ * <code>tcp://</code>" URIs, and 8883 for <code>ssl://</code> URIs.
+ * <p>
+ * If serverURIs is set then it overrides the serverURI parameter passed in on
+ * the constructor of the MQTT client.
+ * <p>
+ * When an attempt to connect is initiated the client will start with the first
+ * serverURI in the list and work through the list until a connection is
+ * established with a server. If a connection cannot be made to any of the
+ * servers then the connect attempt fails.
+ * <p>
+ * Specifying a list of servers that a client may connect to has several uses:
+ * <ol>
+ * <li>High Availability and reliable message delivery
+ * <p>
+ * Some MQTT servers support a high availability feature where two or more
+ * "equal" MQTT servers share state. An MQTT client can connect to any of the
+ * "equal" servers and be assured that messages are reliably delivered and
+ * durable subscriptions are maintained no matter which server the client
+ * connects to.
+ * </p>
+ * <p>
+ * The cleansession flag must be set to false if durable subscriptions and/or
+ * reliable message delivery is required.
+ * </p>
+ * </li>
+ * <li>Hunt List
+ * <p>
+ * A set of servers may be specified that are not "equal" (as in the high
+ * availability option). As no state is shared across the servers reliable
+ * message delivery and durable subscriptions are not valid. The cleansession
+ * flag must be set to true if the hunt list mode is used
+ * </p>
+ * </li>
+ * </ol>
+ */
+ public void setServerURIs(String serverURIs) {
+ this.serverURIs = serverURIs;
+ }
+
+ public int getMqttVersion() {
+ return mqttVersion;
+ }
+
+ /**
+ * Sets the MQTT version. The default action is to connect with version 3.1.1,
+ * and to fall back to 3.1 if that fails. Version 3.1.1 or 3.1 can be selected
+ * specifically, with no fall back, by using the MQTT_VERSION_3_1_1 or
+ * MQTT_VERSION_3_1 options respectively.
+ */
+ public void setMqttVersion(int mqttVersion) {
+ this.mqttVersion = mqttVersion;
+ }
+
+ public boolean isAutomaticReconnect() {
+ return automaticReconnect;
+ }
+
+ /**
+ * Sets whether the client will automatically attempt to reconnect to the server
+ * if the connection is lost.
+ * <ul>
+ * <li>If set to false, the client will not attempt to automatically reconnect
+ * to the server in the event that the connection is lost.</li>
+ * <li>If set to true, in the event that the connection is lost, the client will
+ * attempt to reconnect to the server. It will initially wait 1 second before it
+ * attempts to reconnect, for every failed reconnect attempt, the delay will
+ * double until it is at 2 minutes at which point the delay will stay at 2
+ * minutes.</li>
+ * </ul>
+ */
+ public void setAutomaticReconnect(boolean automaticReconnect) {
+ this.automaticReconnect = automaticReconnect;
+ }
+
+ public int getMaxReconnectDelay() {
+ return maxReconnectDelay;
+ }
+
+ /**
+ * Get the maximum time (in millis) to wait between reconnects
+ */
+ public void setMaxReconnectDelay(int maxReconnectDelay) {
+ this.maxReconnectDelay = maxReconnectDelay;
+ }
+
+ public Properties getCustomWebSocketHeaders() {
+ return customWebSocketHeaders;
+ }
+
+ /**
+ * Sets the Custom WebSocket Headers for the WebSocket Connection.
+ */
+ public void setCustomWebSocketHeaders(Properties customWebSocketHeaders) {
+ this.customWebSocketHeaders = customWebSocketHeaders;
+ }
+
+ public int getExecutorServiceTimeout() {
+ return executorServiceTimeout;
+ }
+
+ /**
+ * Set the time in seconds that the executor service should wait when
+ * terminating before forcefully terminating. It is not recommended to change
+ * this value unless you are absolutely sure that you need to.
+ */
+ public void setExecutorServiceTimeout(int executorServiceTimeout) {
+ this.executorServiceTimeout = executorServiceTimeout;
+ }
+
+ public PahoConfiguration copy() {
+ try {
+ return (PahoConfiguration) clone();
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+
+
+}
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
index 6825b3a..0fb419a 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
@@ -36,14 +36,14 @@ public class PahoConsumer extends DefaultConsumer {
protected void doStart() throws Exception {
super.doStart();
String topic = getEndpoint().getTopic();
- getEndpoint().getClient().subscribe(topic, getEndpoint().getQos());
+ getEndpoint().getClient().subscribe(topic, getEndpoint().getConfiguration().getQos());
getEndpoint().getClient().setCallback(new MqttCallbackExtended() {
@Override
public void connectComplete(boolean reconnect, String serverURI) {
if (reconnect) {
try {
- getEndpoint().getClient().subscribe(topic, getEndpoint().getQos());
+ getEndpoint().getClient().subscribe(topic, getEndpoint().getConfiguration().getQos());
} catch (MqttException e) {
log.error("MQTT resubscribe failed {}", e.getMessage(), e);
}
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoEndpoint.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoEndpoint.java
index 905e234..f84b425 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoEndpoint.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoEndpoint.java
@@ -16,9 +16,6 @@
*/
package org.apache.camel.component.paho;
-import java.util.Set;
-
-import org.apache.camel.Component;
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -37,64 +34,78 @@ import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
/**
- * Component for communicating with MQTT M2M message brokers using Eclipse Paho MQTT Client.
+ * Component for communicating with MQTT message brokers using Eclipse Paho MQTT Client.
*/
@UriEndpoint(firstVersion = "2.16.0", scheme = "paho", title = "Paho", label = "messaging,iot", syntax = "paho:topic")
public class PahoEndpoint extends DefaultEndpoint {
// Configuration members
- @UriPath
+ @UriPath(description = "Name of the topic")
@Metadata(required = true)
- private String topic;
- @UriParam
- private String clientId = "camel-" + System.nanoTime();
- @UriParam(defaultValue = PahoConstants.DEFAULT_BROKER_URL)
- private String brokerUrl = PahoConstants.DEFAULT_BROKER_URL;
- @UriParam(defaultValue = "2")
- private int qos = PahoConstants.DEFAULT_QOS;
+ private final String topic;
@UriParam
- private boolean retained;
- @UriParam(defaultValue = "MEMORY")
- private PahoPersistence persistence = PahoPersistence.MEMORY;
- @UriParam
- private String filePersistenceDirectory;
- @UriParam(defaultValue = "true")
- private boolean autoReconnect = true;
- @UriParam(label = "security") @Metadata(secret = true)
- private String userName;
- @UriParam(label = "security") @Metadata(secret = true)
- private String password;
- @UriParam(label = "advanced", defaultValue = "true")
- private boolean resolveMqttConnectOptions = true;
-
- // Collaboration members
- @UriParam
- private MqttConnectOptions connectOptions;
-
- // Auto-configuration members
-
- private transient MqttClient client;
+ private final PahoConfiguration configuration;
+ @UriParam(label = "advanced")
+ private MqttClient client;
+ private transient boolean stopClient;
- public PahoEndpoint(String uri, String topic, Component component) {
+ public PahoEndpoint(String uri, String topic, PahoComponent component, PahoConfiguration configuration) {
super(uri, component);
this.topic = topic;
+ this.configuration = configuration;
}
@Override
protected void doStart() throws Exception {
super.doStart();
- client = new MqttClient(getBrokerUrl(), getClientId(), resolvePersistence());
- client.connect(resolveMqttConnectOptions());
+
+ if (client == null) {
+ stopClient = true;
+ client = new MqttClient(configuration.getBrokerUrl(), configuration.getClientId(), resolvePersistence());
+ client.connect(createMqttConnectOptions(configuration));
+ }
}
@Override
protected void doStop() throws Exception {
- if (getClient().isConnected()) {
- getClient().disconnect();
+ if (stopClient && client.isConnected()) {
+ client.disconnect();
}
+
super.doStop();
}
+ private static MqttConnectOptions createMqttConnectOptions(PahoConfiguration config) {
+ MqttConnectOptions mq = new MqttConnectOptions();
+ if (ObjectHelper.isNotEmpty(config.getUserName()) && ObjectHelper.isNotEmpty(config.getPassword())) {
+ mq.setUserName(config.getUserName());
+ mq.setPassword(config.getPassword().toCharArray());
+ }
+ mq.setAutomaticReconnect(config.isAutomaticReconnect());
+ mq.setCleanSession(config.isCleanSession());
+ mq.setConnectionTimeout(config.getConnectionTimeout());
+ mq.setExecutorServiceTimeout(config.getExecutorServiceTimeout());
+ mq.setCustomWebSocketHeaders(config.getCustomWebSocketHeaders());
+ mq.setHttpsHostnameVerificationEnabled(config.isHttpsHostnameVerificationEnabled());
+ mq.setKeepAliveInterval(config.getKeepAliveInterval());
+ mq.setMaxInflight(config.getMaxInflight());
+ mq.setMaxReconnectDelay(config.getMaxReconnectDelay());
+ mq.setMqttVersion(config.getMqttVersion());
+ mq.setSocketFactory(config.getSocketFactory());
+ mq.setSSLHostnameVerifier(config.getSslHostnameVerifier());
+ mq.setSSLProperties(config.getSslClientProps());
+ if (config.getWillTopic() != null && config.getWillPayload() != null) {
+ mq.setWill(config.getWillTopic(),
+ config.getWillPayload().getBytes(),
+ config.getWillQos(),
+ config.isWillRetained());
+ }
+ if (config.getServerURIs() != null) {
+ mq.setServerURIs(config.getServerURIs().split(","));
+ }
+ return mq;
+ }
+
@Override
public Producer createProducer() throws Exception {
return new PahoProducer(this);
@@ -107,48 +118,26 @@ public class PahoEndpoint extends DefaultEndpoint {
@Override
public PahoComponent getComponent() {
- return (PahoComponent)super.getComponent();
+ return (PahoComponent) super.getComponent();
+ }
+
+ public String getTopic() {
+ return topic;
}
// Resolvers
protected MqttClientPersistence resolvePersistence() {
- if (persistence == PahoPersistence.MEMORY) {
+ if (configuration.getPersistence() == PahoPersistence.MEMORY) {
return new MemoryPersistence();
} else {
- if (filePersistenceDirectory != null) {
- return new MqttDefaultFilePersistence(filePersistenceDirectory);
+ if (configuration.getFilePersistenceDirectory() != null) {
+ return new MqttDefaultFilePersistence(configuration.getFilePersistenceDirectory());
} else {
return new MqttDefaultFilePersistence();
}
}
}
- protected MqttConnectOptions resolveMqttConnectOptions() {
- if (connectOptions != null) {
- return connectOptions;
- }
-
- if (resolveMqttConnectOptions) {
- Set<MqttConnectOptions> connectOptions = getCamelContext().getRegistry().findByType(MqttConnectOptions.class);
- if (connectOptions.size() == 1) {
- log.info("Single MqttConnectOptions instance found in the registry. It will be used by the endpoint.");
- return connectOptions.iterator().next();
- } else if (connectOptions.size() > 1) {
- log.warn("Found {} instances of the MqttConnectOptions in the registry. None of these will be used by the endpoint. "
- + "Please use 'connectOptions' endpoint option to select one.", connectOptions.size());
- }
- }
-
- MqttConnectOptions options = new MqttConnectOptions();
- options.setAutomaticReconnect(autoReconnect);
-
- if (ObjectHelper.isNotEmpty(userName) && ObjectHelper.isNotEmpty(password)) {
- options.setUserName(userName);
- options.setPassword(password.toCharArray());
- }
- return options;
- }
-
public Exchange createExchange(MqttMessage mqttMessage, String topic) {
Exchange exchange = createExchange();
@@ -161,85 +150,8 @@ public class PahoEndpoint extends DefaultEndpoint {
return exchange;
}
- // Configuration getters & setters
-
- public String getClientId() {
- return clientId;
- }
-
- /**
- * MQTT client identifier.
- */
- public void setClientId(String clientId) {
- this.clientId = clientId;
- }
-
- public String getBrokerUrl() {
- return brokerUrl;
- }
-
- /**
- * The URL of the MQTT broker.
- */
- public void setBrokerUrl(String brokerUrl) {
- this.brokerUrl = brokerUrl;
- }
-
- public String getTopic() {
- return topic;
- }
-
- /**
- * Name of the topic
- */
- public void setTopic(String topic) {
- this.topic = topic;
- }
-
- public int getQos() {
- return qos;
- }
-
- /**
- * Client quality of service level (0-2).
- */
- public void setQos(int qos) {
- this.qos = qos;
- }
-
- public boolean isRetained() {
- return retained;
- }
-
- /**
- * Retain option
- */
- public void setRetained(boolean retained) {
- this.retained = retained;
- }
-
- // Auto-configuration getters & setters
-
- public PahoPersistence getPersistence() {
- return persistence;
- }
-
- /**
- * Client persistence to be used - memory or file.
- */
- public void setPersistence(PahoPersistence persistence) {
- this.persistence = persistence;
- }
-
- public String getFilePersistenceDirectory() {
- return filePersistenceDirectory;
- }
-
- /**
- * Base directory used by file persistence. Will by default use user directory.
- */
- public void setFilePersistenceDirectory(String filePersistenceDirectory) {
- this.filePersistenceDirectory = filePersistenceDirectory;
+ public PahoConfiguration getConfiguration() {
+ return configuration;
}
public MqttClient getClient() {
@@ -247,65 +159,9 @@ public class PahoEndpoint extends DefaultEndpoint {
}
/**
- * To use the existing MqttClient instance as client.
+ * To use an exiting mqtt client
*/
public void setClient(MqttClient client) {
this.client = client;
}
-
- public MqttConnectOptions getConnectOptions() {
- return connectOptions;
- }
-
- /**
- * Client connection options
- */
- public void setConnectOptions(MqttConnectOptions connOpts) {
- this.connectOptions = connOpts;
- }
-
- public synchronized boolean isAutoReconnect() {
- return autoReconnect;
- }
-
- /**
- * Client will automatically attempt to reconnect to the server if the connection is lost
- */
- public synchronized void setAutoReconnect(boolean autoReconnect) {
- this.autoReconnect = autoReconnect;
- }
-
- public String getUserName() {
- return userName;
- }
-
- /**
- * Username to be used for authentication against the MQTT broker
- */
- public void setUserName(String userName) {
- this.userName = userName;
- }
-
- public String getPassword() {
- return password;
- }
-
- /**
- * Password to be used for authentication against the MQTT broker
- */
- public void setPassword(String password) {
- this.password = password;
- }
-
- public synchronized boolean isResolveMqttConnectOptions() {
- return resolveMqttConnectOptions;
- }
-
- /**
- * Define if you don't want to resolve the MQTT Connect Options from registry
- */
- public synchronized void setResolveMqttConnectOptions(boolean resolveMqttConnectOptions) {
- this.resolveMqttConnectOptions = resolveMqttConnectOptions;
- }
-
}
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
index 29e1854..08428a5 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
@@ -32,8 +32,8 @@ public class PahoProducer extends DefaultProducer {
MqttClient client = getEndpoint().getClient();
String topic = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC, getEndpoint().getTopic(), String.class);
- int qos = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_MSG_QOS, getEndpoint().getQos(), Integer.class);
- boolean retained = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_MSG_RETAINED, getEndpoint().isRetained(), Boolean.class);
+ int qos = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_MSG_QOS, getEndpoint().getConfiguration().getQos(), Integer.class);
+ boolean retained = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_MSG_RETAINED, getEndpoint().getConfiguration().isRetained(), Boolean.class);
byte[] payload = exchange.getIn().getBody(byte[].class);
MqttMessage message = new MqttMessage(payload);
diff --git a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoComponentTest.java b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoComponentTest.java
index 33ae40d..9a48988 100644
--- a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoComponentTest.java
+++ b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoComponentTest.java
@@ -19,23 +19,18 @@ package org.apache.camel.component.paho;
import java.io.UnsupportedEncodingException;
import org.apache.activemq.broker.BrokerService;
-import org.apache.camel.BindToRegistry;
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.AvailablePortFinder;
import org.apache.camel.test.junit4.CamelTestSupport;
-import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.junit.After;
import org.junit.Test;
public class PahoComponentTest extends CamelTestSupport {
- @BindToRegistry("connectOptions")
- MqttConnectOptions connectOptions = new MqttConnectOptions();
-
@EndpointInject("mock:test")
MockEndpoint mock;
@@ -82,8 +77,6 @@ public class PahoComponentTest extends CamelTestSupport {
from("paho:persistenceTest?persistence=FILE&brokerUrl=tcp://localhost:" + mqttPort).to("mock:persistenceTest");
- from("direct:connectOptions").to("paho:registryConnectOptions?connectOptions=#connectOptions&brokerUrl=tcp://localhost:" + mqttPort);
-
from("direct:testCustomizedPaho").to("customizedPaho:testCustomizedPaho?brokerUrl=tcp://localhost:" + mqttPort);
from("paho:testCustomizedPaho?brokerUrl=tcp://localhost:" + mqttPort).to("mock:testCustomizedPaho");
}
@@ -100,10 +93,10 @@ public class PahoComponentTest extends CamelTestSupport {
// Then
assertEquals("/test/topic", endpoint.getTopic());
- assertEquals("sampleClient", endpoint.getClientId());
- assertEquals("tcp://localhost:" + mqttPort, endpoint.getBrokerUrl());
- assertEquals(2, endpoint.getQos());
- assertEquals(PahoPersistence.FILE, endpoint.getPersistence());
+ assertEquals("sampleClient", endpoint.getConfiguration().getClientId());
+ assertEquals("tcp://localhost:" + mqttPort, endpoint.getConfiguration().getBrokerUrl());
+ assertEquals(2, endpoint.getConfiguration().getQos());
+ assertEquals(PahoPersistence.FILE, endpoint.getConfiguration().getPersistence());
}
@Test
@@ -132,25 +125,6 @@ public class PahoComponentTest extends CamelTestSupport {
}
@Test
- public void shouldUseConnectionOptionsFromRegistry() {
- // Given
- PahoEndpoint pahoWithConnectOptionsFromRegistry = getMandatoryEndpoint("paho:registryConnectOptions?connectOptions=#connectOptions&brokerUrl=tcp://localhost:" + mqttPort,
- PahoEndpoint.class);
-
- // Then
- assertSame(connectOptions, pahoWithConnectOptionsFromRegistry.resolveMqttConnectOptions());
- }
-
- @Test
- public void shouldAutomaticallyUseConnectionOptionsFromRegistry() {
- // Given
- PahoEndpoint pahoWithConnectOptionsFromRegistry = getMandatoryEndpoint("paho:registryConnectOptions?brokerUrl=tcp://localhost:" + mqttPort, PahoEndpoint.class);
-
- // Then
- assertSame(connectOptions, pahoWithConnectOptionsFromRegistry.resolveMqttConnectOptions());
- }
-
- @Test
public void shouldKeepDefaultMessageInHeader() throws InterruptedException, UnsupportedEncodingException {
// Given
final String msg = "msg";
diff --git a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java
index 9182566..ef751a4 100644
--- a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java
+++ b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java
@@ -17,19 +17,14 @@
package org.apache.camel.component.paho;
import org.apache.activemq.broker.BrokerService;
-import org.apache.camel.BindToRegistry;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.test.AvailablePortFinder;
import org.apache.camel.test.junit4.CamelTestSupport;
-import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.junit.After;
import org.junit.Test;
public class PahoOverrideTopicTest extends CamelTestSupport {
- @BindToRegistry("connectOptions")
- MqttConnectOptions connectOptions = new MqttConnectOptions();
-
BrokerService broker;
int mqttPort = AvailablePortFinder.getNextAvailable();