org.apache.ignite.stream.mqtt

Class MqttStreamer<K,V>

java.lang.Object

org.apache.ignite.stream.StreamAdapter<org.eclipse.paho.client.mqttv3.MqttMessage,K,V>

org.apache.ignite.stream.mqtt.MqttStreamer<K,V>

All Implemented Interfaces:

org.eclipse.paho.client.mqttv3.MqttCallback

public class MqttStreamer<K,V>
extends StreamAdapter<org.eclipse.paho.client.mqttv3.MqttMessage,K,V>
implements org.eclipse.paho.client.mqttv3.MqttCallback

Streamer that consumes from a MQTT topic and feeds key-value pairs into an IgniteDataStreamer instance, using Eclipse Paho as an MQTT client.

You must also provide a StreamSingleTupleExtractor or a StreamMultipleTupleExtractor to extract cache tuples out of the incoming message.

This Streamer has many features:

• Subscribing to a single topic or multiple topics at once.
• Specifying the subscriber's QoS for a single topic or for multiple topics.
• Allows setting MqttConnectOptions to support features like last will testament, persistent sessions, etc.
• Specifying the client ID. A random one will be generated and maintained throughout reconnections if the user does not provide one.
• (Re-)Connection retries based on the guava-retrying library. Retry wait and retry stop policies can be configured.
• Blocking the start() method until connected for the first time.

Note: features like durable subscriptions, last will testament, etc. can be configured via the setConnectOptions(MqttConnectOptions) setter.

See Also:

guava-retrying library

Field Summary

Fields inherited from class org.apache.ignite.stream.StreamAdapter

ignite

Constructor Summary

Constructors

Constructor and Description
MqttStreamer()

Method Summary

Modifier and Type	Method and Description
void	connectionLost(Throwable throwable) Implements the MqttCallback.connectionLost(Throwable) callback method for the MQTT client to inform the streamer that the connection has been lost.
void	deliveryComplete(org.eclipse.paho.client.mqttv3.IMqttDeliveryToken token) Empty implementation of MqttCallback.deliveryComplete(IMqttDeliveryToken).
String	getBrokerUrl() Gets the broker URL.
String	getClientId() Gets the client ID, either the one set by the user or the automatically generated one.
org.eclipse.paho.client.mqttv3.MqttConnectOptions	getConnectOptions() Gets the currently used MQTT client connect options.
Integer	getDisconnectForciblyTimeout() Gets the timeout if disconnecting forcibly.
Integer	getDisconnectQuiesceTimeout() Gets the disconnect quiesce timeout.
org.eclipse.paho.client.mqttv3.MqttClientPersistence	getPersistence() Gets the currently set persistence mechanism.
List<Integer>	getQualitiesOfService() Gets the qualities of service for multiple topics.
Integer	getQualityOfService() Gets the quality of service set by the user for a single topic consumption.
com.github.rholder.retry.StopStrategy	getRetryStopStrategy() Gets the retry stop strategy.
com.github.rholder.retry.WaitStrategy	getRetryWaitStrategy() Gets the retry wait strategy.
String	getTopic() Gets the subscribed topic.
List<String>	getTopics() Gets the topics subscribed to.
boolean	isBlockUntilConnected() Gets whether to block the start() method until connected for the first time.
boolean	isConnected() Returns whether this streamer is connected by delegating to the underlying MqttClient.isConnected()
boolean	isDisconnectForcibly() Gets whether this MQTT client will disconnect forcibly when shutting down.
boolean	isStopped() Returns whether this streamer is stopped.
void	messageArrived(String topic, org.eclipse.paho.client.mqttv3.MqttMessage message) Implements the MqttCallback.messageArrived(String, MqttMessage) to receive an MQTT message.
void	setBlockUntilConnected(boolean blockUntilConnected) Sets whether to block the start() method until connected for the first time.
void	setBrokerUrl(String brokerUrl) Sets the broker URL (compulsory).
void	setClientId(String clientId) Sets the MQTT client ID (optional).
void	setConnectOptions(org.eclipse.paho.client.mqttv3.MqttConnectOptions connectOptions) Sets the MQTT client connect options, where users can configured the last will and testament, durability, etc.
void	setDisconnectForcibly(boolean disconnectForcibly) Sets whether to disconnect forcibly or not when shutting down.
void	setDisconnectForciblyTimeout(Integer disconnectForciblyTimeout) Sets the timeout if disconnecting forcibly.
void	setDisconnectQuiesceTimeout(Integer disconnectQuiesceTimeout) Sets the quiesce timeout on disconnection.
void	setPersistence(org.eclipse.paho.client.mqttv3.MqttClientPersistence persistence) Sets the persistence mechanism.
void	setQualitiesOfService(List<Integer> qualitiesOfService) Sets the qualities of service to use for multiple topic subscriptions.
void	setQualityOfService(Integer qualityOfService) Sets the quality of service to use for a single topic subscription (optional).
void	setRetryStopStrategy(com.github.rholder.retry.StopStrategy retryStopStrategy) Sets the strategy to determine when to stop retrying to (re-)connect.
void	setRetryWaitStrategy(com.github.rholder.retry.WaitStrategy retryWaitStrategy) Sets the strategy to determine how long to wait between retry attempts.
void	setTopic(String topic) Sets the topic to subscribe to, if a single topic.
void	setTopics(List<String> topics) Sets the topics to subscribe to, if many.
void	start() Starts streamer.
void	stop() Stops streamer.

Methods inherited from class org.apache.ignite.stream.StreamAdapter

addMessage, getIgnite, getMultipleTupleExtractor, getSingleTupleExtractor, getStreamer, getTupleExtractor, setIgnite, setMultipleTupleExtractor, setSingleTupleExtractor, setStreamer, setTupleExtractor

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

MqttStreamer

public MqttStreamer()

Method Detail

start

public void start()
           throws IgniteException

Starts streamer.

Throws:

IgniteException - If failed.

stop

public void stop()
          throws IgniteException

Stops streamer.

Throws:

IgniteException - If failed.

connectionLost

public void connectionLost(Throwable throwable)

Implements the MqttCallback.connectionLost(Throwable) callback method for the MQTT client to inform the streamer that the connection has been lost.

Specified by:

connectionLost in interface org.eclipse.paho.client.mqttv3.MqttCallback

messageArrived

public void messageArrived(String topic,
                           org.eclipse.paho.client.mqttv3.MqttMessage message)
                    throws Exception

Implements the MqttCallback.messageArrived(String, MqttMessage) to receive an MQTT message.

Specified by:

messageArrived in interface org.eclipse.paho.client.mqttv3.MqttCallback

Throws:

Exception

deliveryComplete

public void deliveryComplete(org.eclipse.paho.client.mqttv3.IMqttDeliveryToken token)

Empty implementation of MqttCallback.deliveryComplete(IMqttDeliveryToken). Not required by the streamer as it doesn't produce messages.

Specified by:

deliveryComplete in interface org.eclipse.paho.client.mqttv3.MqttCallback

setBrokerUrl

public void setBrokerUrl(String brokerUrl)

Sets the broker URL (compulsory).

Parameters:

brokerUrl - The Broker URL (compulsory).

getBrokerUrl

public String getBrokerUrl()

Gets the broker URL.

Returns:

The Broker URL.

setTopic

public void setTopic(String topic)

Sets the topic to subscribe to, if a single topic.

Parameters:

topic - The topic to subscribe to.

getTopic

public String getTopic()

Gets the subscribed topic.

Returns:

The subscribed topic.

setQualityOfService

public void setQualityOfService(Integer qualityOfService)

Sets the quality of service to use for a single topic subscription (optional).

Parameters:

qualityOfService - The quality of service.

getQualityOfService

public Integer getQualityOfService()

Gets the quality of service set by the user for a single topic consumption.

Returns:

The quality of service.

setTopics

public void setTopics(List<String> topics)

Sets the topics to subscribe to, if many.

Parameters:

topics - The topics.

getTopics

public List<String> getTopics()

Gets the topics subscribed to.

Returns:

The topics subscribed to.

setQualitiesOfService

public void setQualitiesOfService(List<Integer> qualitiesOfService)

Sets the qualities of service to use for multiple topic subscriptions. If specified, the list must contain the same number of elements as topics.

Parameters:

qualitiesOfService - The qualities of service.

getQualitiesOfService

public List<Integer> getQualitiesOfService()

Gets the qualities of service for multiple topics.

Returns:

The qualities of service.

setClientId

public void setClientId(String clientId)

Sets the MQTT client ID (optional). If one is not provided, the streamer will generate one and will maintain it througout any reconnection attempts.

Parameters:

clientId - The client ID.

getClientId

public String getClientId()

Gets the client ID, either the one set by the user or the automatically generated one.

Returns:

The client ID.

getPersistence

public org.eclipse.paho.client.mqttv3.MqttClientPersistence getPersistence()

Gets the currently set persistence mechanism.

Returns:

The persistence mechanism.

setPersistence

public void setPersistence(org.eclipse.paho.client.mqttv3.MqttClientPersistence persistence)

Sets the persistence mechanism. If not set, Paho will use its default.

Parameters:

persistence - A configurable persistence mechanism.

getConnectOptions

public org.eclipse.paho.client.mqttv3.MqttConnectOptions getConnectOptions()

Gets the currently used MQTT client connect options.

Returns:

The MQTT client connect options.

setConnectOptions

public void setConnectOptions(org.eclipse.paho.client.mqttv3.MqttConnectOptions connectOptions)

Sets the MQTT client connect options, where users can configured the last will and testament, durability, etc.

Parameters:

connectOptions - The MQTT client connect options.

setDisconnectForcibly

public void setDisconnectForcibly(boolean disconnectForcibly)

Sets whether to disconnect forcibly or not when shutting down. By default, it's false.

Parameters:

disconnectForcibly - Whether to disconnect forcibly or not. By default, it's false.

isDisconnectForcibly

public boolean isDisconnectForcibly()

Gets whether this MQTT client will disconnect forcibly when shutting down.

Returns:

Whether to disconnect forcibly or not.

setDisconnectQuiesceTimeout

public void setDisconnectQuiesceTimeout(Integer disconnectQuiesceTimeout)

Sets the quiesce timeout on disconnection. If not provided, this streamer won't use any.

Parameters:

disconnectQuiesceTimeout - The disconnect quiesce timeout.

getDisconnectQuiesceTimeout

public Integer getDisconnectQuiesceTimeout()

Gets the disconnect quiesce timeout.

Returns:

The disconnect quiesce timeout.

setDisconnectForciblyTimeout

public void setDisconnectForciblyTimeout(Integer disconnectForciblyTimeout)

Sets the timeout if disconnecting forcibly. Compulsory in that case.

Parameters:

disconnectForciblyTimeout - The disconnect forcibly timeout.

getDisconnectForciblyTimeout

public Integer getDisconnectForciblyTimeout()

Gets the timeout if disconnecting forcibly.

Returns:

Timeout.

setRetryWaitStrategy

public void setRetryWaitStrategy(com.github.rholder.retry.WaitStrategy retryWaitStrategy)

Sets the strategy to determine how long to wait between retry attempts. By default, this streamer uses a Fibonacci-based strategy.

Parameters:

retryWaitStrategy - The retry wait strategy.

getRetryWaitStrategy

public com.github.rholder.retry.WaitStrategy getRetryWaitStrategy()

Gets the retry wait strategy.

Returns:

The retry wait strategy.

setRetryStopStrategy

public void setRetryStopStrategy(com.github.rholder.retry.StopStrategy retryStopStrategy)

Sets the strategy to determine when to stop retrying to (re-)connect. By default, we never stop.

Parameters:

retryStopStrategy - The retry stop strategy.

getRetryStopStrategy

public com.github.rholder.retry.StopStrategy getRetryStopStrategy()

Gets the retry stop strategy.

Returns:

The retry stop strategy.

setBlockUntilConnected

public void setBlockUntilConnected(boolean blockUntilConnected)

Sets whether to block the start() method until connected for the first time. By default, it's false.

Parameters:

blockUntilConnected - Whether to block or not.

isBlockUntilConnected

public boolean isBlockUntilConnected()

Gets whether to block the start() method until connected for the first time. By default, it's false.

Returns:

true if should connect synchronously in start.

isStopped

public boolean isStopped()

Returns whether this streamer is stopped.

Returns:

true if stopped; false if not.

isConnected

public boolean isConnected()

Returns whether this streamer is connected by delegating to the underlying MqttClient.isConnected()

Returns:

true if connected; false if not.

See Also:

MqttClient.isConnected()