services.moleculer.transporter

Class MqttTransporter

java.lang.Object

services.moleculer.service.MoleculerComponent

services.moleculer.transporter.Transporter

services.moleculer.transporter.MqttTransporter

All Implemented Interfaces:

net.sf.xenqtt.client.AsyncClientListener, net.sf.xenqtt.client.MqttClientListener, MetricConstants, MoleculerLifecycle

public class MqttTransporter
extends Transporter
implements net.sf.xenqtt.client.AsyncClientListener

MQTT Transporter (eg. for Mosquitto MQTT Server or ActiveMQ Server). MQTT is a machine-to-machine (M2M)/"Internet of Things" connectivity protocol. It was designed as an extremely lightweight publish/subscribe messaging transport (website: http://mqtt.org). Usage:

 ServiceBroker broker = ServiceBroker.builder().nodeID("node1").transporter(new MqttTransporter("localhost")).build();
 

Required dependency:

// https://mvnrepository.com/artifact/net.sf.xenqtt/xenqtt
compile group: 'net.sf.xenqtt', name: 'xenqtt', version: '0.9.7'

See Also:

TcpTransporter, RedisTransporter, NatsTransporter, NatsStreamingTransporter, JmsTransporter, KafkaTransporter, AmqpTransporter

Field Summary

Fields

Modifier and Type	Field and Description
protected int	blockingTimeoutSeconds
protected boolean	cleanSession
protected net.sf.xenqtt.client.AsyncMqttClient	client
protected int	connectTimeoutSeconds
protected short	keepAliveSeconds
protected int	maxInFlightMessages
protected int	messageResendIntervalSeconds
protected java.lang.String	password
protected net.sf.xenqtt.message.QoS	qos
protected java.util.concurrent.ConcurrentHashMap<java.lang.String,io.datatree.Promise>	subscriptions
protected java.lang.String	url
protected java.lang.String	username

Fields inherited from class services.moleculer.transporter.Transporter

checkTimeoutTimer, counterTransporterPacketsReceivedBytes, counterTransporterPacketsReceivedTotal, counterTransporterPacketsSentBytes, counterTransporterPacketsSentTotal, debug, debugHeartbeats, disconnectChannel, discoverBroadcastChannel, discoverChannel, eventbus, eventChannel, executor, gaugeTransitConnected, heartbeatChannel, heartbeatInterval, heartbeatTimeout, heartBeatTimer, infoBroadcastChannel, infoChannel, infoScheduled, infoScheduledAt, instanceID, lastReceivedMessageAt, metrics, monitor, namespace, nodeID, nodes, offlineTimeout, PACKET_DISCONNECT, PACKET_DISCOVER, PACKET_EVENT, PACKET_HEARTBEAT, PACKET_INFO, PACKET_PING, PACKET_PONG, PACKET_REQUEST, PACKET_RESPONSE, pingChannel, pongChannel, preferHostname, prefix, protocolVersion, registry, requestChannel, responseChannel, scheduler, serializer, serviceInvoker, subscriptionTimeout, uidGenerator, usingJsonSerializer

Fields inherited from class services.moleculer.service.MoleculerComponent

broker, logger, name

Fields inherited from interface net.sf.xenqtt.client.AsyncClientListener

NULL_LISTENER

Fields inherited from interface services.moleculer.metrics.MetricConstants

MOLECULER_CACHER_CLEAN_TIME, MOLECULER_CACHER_CLEAN_TIME_DESC, MOLECULER_CACHER_CLEAN_TOTAL, MOLECULER_CACHER_CLEAN_TOTAL_DESC, MOLECULER_CACHER_DEL_TIME, MOLECULER_CACHER_DEL_TIME_DESC, MOLECULER_CACHER_DEL_TOTAL, MOLECULER_CACHER_DEL_TOTAL_DESC, MOLECULER_CACHER_EXPIRED_TOTAL, MOLECULER_CACHER_EXPIRED_TOTAL_DESC, MOLECULER_CACHER_FOUND_TOTAL, MOLECULER_CACHER_FOUND_TOTAL_DESC, MOLECULER_CACHER_GET_TIME, MOLECULER_CACHER_GET_TIME_DESC, MOLECULER_CACHER_GET_TOTAL, MOLECULER_CACHER_GET_TOTAL_DESC, MOLECULER_CACHER_SET_TIME, MOLECULER_CACHER_SET_TIME_DESC, MOLECULER_CACHER_SET_TOTAL, MOLECULER_CACHER_SET_TOTAL_DESC, MOLECULER_CIRCUIT_BREAKER_HALF_OPENED_ACTIVE, MOLECULER_CIRCUIT_BREAKER_HALF_OPENED_ACTIVE_DESC, MOLECULER_CIRCUIT_BREAKER_OPENED_ACTIVE, MOLECULER_CIRCUIT_BREAKER_OPENED_ACTIVE_DESC, MOLECULER_CIRCUIT_BREAKER_OPENED_TOTAL, MOLECULER_CIRCUIT_BREAKER_OPENED_TOTAL_DESC, MOLECULER_EVENT_BROADCAST_TOTAL, MOLECULER_EVENT_BROADCAST_TOTAL_DESC, MOLECULER_EVENT_BROADCASTLOCAL_TOTAL, MOLECULER_EVENT_BROADCASTLOCAL_TOTAL_DESC, MOLECULER_EVENT_EMIT_TOTAL, MOLECULER_EVENT_EMIT_TOTAL_DESC, MOLECULER_EVENT_RECEIVED_ACTIVE, MOLECULER_EVENT_RECEIVED_ACTIVE_DESC, MOLECULER_EVENT_RECEIVED_ERROR_TOTAL, MOLECULER_EVENT_RECEIVED_ERROR_TOTAL_DESC, MOLECULER_EVENT_RECEIVED_TIME, MOLECULER_EVENT_RECEIVED_TIME_DESC, MOLECULER_EVENT_RECEIVED_TOTAL, MOLECULER_EVENT_RECEIVED_TOTAL_DESC, MOLECULER_EXECUTOR, MOLECULER_REQUEST_ACTIVE, MOLECULER_REQUEST_ACTIVE_DESC, MOLECULER_REQUEST_ERROR_TOTAL, MOLECULER_REQUEST_ERROR_TOTAL_DESC, MOLECULER_REQUEST_LEVELS, MOLECULER_REQUEST_LEVELS_DESC, MOLECULER_REQUEST_RETRY_ATTEMPTS_TOTAL, MOLECULER_REQUEST_RETRY_ATTEMPTS_TOTAL_DESC, MOLECULER_REQUEST_TIME, MOLECULER_REQUEST_TIME_DESC, MOLECULER_REQUEST_TIMEOUT_TOTAL, MOLECULER_REQUEST_TIMEOUT_TOTAL_DESC, MOLECULER_REQUEST_TOTAL, MOLECULER_REQUEST_TOTAL_DESC, MOLECULER_SCHEDULER, MOLECULER_TRANSIT_CONNECTED, MOLECULER_TRANSIT_CONNECTED_DESC, MOLECULER_TRANSIT_ORPHAN_RESPONSE_TOTAL, MOLECULER_TRANSIT_ORPHAN_RESPONSE_TOTAL_DESC, MOLECULER_TRANSIT_REQUESTS_ACTIVE, MOLECULER_TRANSIT_REQUESTS_ACTIVE_DESC, MOLECULER_TRANSIT_STREAMS_RECEIVE_ACTIVE, MOLECULER_TRANSIT_STREAMS_RECEIVE_ACTIVE_DESC, MOLECULER_TRANSPORTER_PACKETS_RECEIVED_BYTES, MOLECULER_TRANSPORTER_PACKETS_RECEIVED_BYTES_DESC, MOLECULER_TRANSPORTER_PACKETS_RECEIVED_TOTAL, MOLECULER_TRANSPORTER_PACKETS_RECEIVED_TOTAL_DESC, MOLECULER_TRANSPORTER_PACKETS_SENT_BYTES, MOLECULER_TRANSPORTER_PACKETS_SENT_BYTES_DESC, MOLECULER_TRANSPORTER_PACKETS_SENT_TOTAL, MOLECULER_TRANSPORTER_PACKETS_SENT_TOTAL_DESC

Constructor Summary

Constructors

Constructor and Description
MqttTransporter()
MqttTransporter(java.lang.String url)
MqttTransporter(java.lang.String username, java.lang.String password, java.lang.String url)

Method Summary

Modifier and Type	Method and Description
void	connect()
void	connected(net.sf.xenqtt.client.MqttClient client, net.sf.xenqtt.message.ConnectReturnCode returnCode)
protected void	disconnect()
void	disconnected(net.sf.xenqtt.client.MqttClient client, java.lang.Throwable cause, boolean reconnecting)
protected void	error(java.lang.Throwable cause) Any I/O error occurred.
int	getBlockingTimeoutSeconds()
int	getConnectTimeoutSeconds()
short	getKeepAliveSeconds()
int	getMaxInFlightMessages()
int	getMessageResendIntervalSeconds()
java.lang.String	getPassword()
net.sf.xenqtt.message.QoS	getQos()
java.lang.String	getUrl()
java.lang.String	getUsername()
boolean	isCleanSession()
void	publish(java.lang.String channel, io.datatree.Tree message)
void	published(net.sf.xenqtt.client.MqttClient client, net.sf.xenqtt.client.PublishMessage message)
void	publishReceived(net.sf.xenqtt.client.MqttClient client, net.sf.xenqtt.client.PublishMessage message)
protected void	reconnect(java.lang.Throwable cause)
void	setBlockingTimeoutSeconds(int blockingTimeoutSeconds)
void	setCleanSession(boolean cleanSession)
void	setConnectTimeoutSeconds(int connectTimeoutSeconds)
void	setKeepAliveSeconds(short keepAliveInterval)
void	setMaxInFlightMessages(int maxInFlightMessages)
void	setMessageResendIntervalSeconds(int messageResendIntervalSeconds)
void	setPassword(java.lang.String password)
void	setQos(net.sf.xenqtt.message.QoS qos)
void	setUrl(java.lang.String url)
void	setUsername(java.lang.String username)
void	stopped() Closes transporter.
io.datatree.Promise	subscribe(java.lang.String channel)
void	subscribed(net.sf.xenqtt.client.MqttClient client, net.sf.xenqtt.client.Subscription[] requestedSubscriptions, net.sf.xenqtt.client.Subscription[] grantedSubscriptions, boolean requestsGranted)
void	unsubscribed(net.sf.xenqtt.client.MqttClient client, java.lang.String[] topics)

Methods inherited from class services.moleculer.transporter.Transporter

broadcastInfoPacket, broadcastNodeConnected, broadcastNodeDisconnected, broadcastNodeUpdated, broadcastTransporterConnected, broadcastTransporterDisconnected, channel, checkTimeouts, connected, connected, createPingPacket, getAddress, getAllNodeIDs, getCpuUsage, getDescriptor, getHeartbeatInterval, getHeartbeatTimeout, getInstanceID, getLastHeartbeatTime, getNamespace, getOfflineTimeout, getPrefix, getSerializer, getSubscriptionTimeout, isDebug, isDebugHeartbeats, isOnline, isPreferHostname, processReceivedMessage, processReceivedMessage, publish, received, scheduleInfoPacket, sendClosePacket, sendDataPacket, sendDisconnectPacket, sendDiscoverPacket, sendErrorPacket, sendEventPacket, sendHeartbeatPacket, sendInfoPacket, sendPongPacket, sendRequestPacket, setDebug, setDebugHeartbeats, setHeartbeatInterval, setHeartbeatTimeout, setNamespace, setOfflineTimeout, setPreferHostname, setPrefix, setSerializer, setSubscriptionTimeout, started, subscribe, updateNodeInfo

Methods inherited from class services.moleculer.service.MoleculerComponent

getBroker, getLogger, getName

Methods inherited from class java.lang.Object

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

Field Detail

username

protected java.lang.String username

password

protected java.lang.String password

url

protected java.lang.String url

cleanSession

protected boolean cleanSession

keepAliveSeconds

protected short keepAliveSeconds

connectTimeoutSeconds

protected int connectTimeoutSeconds

messageResendIntervalSeconds

protected int messageResendIntervalSeconds

blockingTimeoutSeconds

protected int blockingTimeoutSeconds

maxInFlightMessages

protected int maxInFlightMessages

qos

protected net.sf.xenqtt.message.QoS qos

subscriptions

protected final java.util.concurrent.ConcurrentHashMap<java.lang.String,io.datatree.Promise> subscriptions

client

protected net.sf.xenqtt.client.AsyncMqttClient client

Constructor Detail

MqttTransporter

public MqttTransporter()

MqttTransporter

public MqttTransporter(java.lang.String url)

MqttTransporter

public MqttTransporter(java.lang.String username,
                       java.lang.String password,
                       java.lang.String url)

Method Detail

connect

public void connect()

Specified by:

connect in class Transporter

connected

public void connected(net.sf.xenqtt.client.MqttClient client,
                      net.sf.xenqtt.message.ConnectReturnCode returnCode)

Specified by:

connected in interface net.sf.xenqtt.client.AsyncClientListener

disconnected

public void disconnected(net.sf.xenqtt.client.MqttClient client,
                         java.lang.Throwable cause,
                         boolean reconnecting)

Specified by:

disconnected in interface net.sf.xenqtt.client.MqttClientListener

disconnect

protected void disconnect()

reconnect

protected void reconnect(java.lang.Throwable cause)

error

protected void error(java.lang.Throwable cause)

Description copied from class: Transporter

Any I/O error occurred. Implementation-specific error handling goes here (reconnection, etc.).

Overrides:

error in class Transporter

Parameters:

cause - I/O error

stopped

public void stopped()

Closes transporter.

Specified by:

stopped in interface MoleculerLifecycle

Overrides:

stopped in class Transporter

subscribe

public io.datatree.Promise subscribe(java.lang.String channel)

Specified by:

subscribe in class Transporter

subscribed

public void subscribed(net.sf.xenqtt.client.MqttClient client,
                       net.sf.xenqtt.client.Subscription[] requestedSubscriptions,
                       net.sf.xenqtt.client.Subscription[] grantedSubscriptions,
                       boolean requestsGranted)

Specified by:

subscribed in interface net.sf.xenqtt.client.AsyncClientListener

unsubscribed

public void unsubscribed(net.sf.xenqtt.client.MqttClient client,
                         java.lang.String[] topics)

Specified by:

unsubscribed in interface net.sf.xenqtt.client.AsyncClientListener

publish

public void publish(java.lang.String channel,
                    io.datatree.Tree message)

Specified by:

publish in class Transporter

published

public void published(net.sf.xenqtt.client.MqttClient client,
                      net.sf.xenqtt.client.PublishMessage message)

Specified by:

published in interface net.sf.xenqtt.client.AsyncClientListener

publishReceived

public void publishReceived(net.sf.xenqtt.client.MqttClient client,
                            net.sf.xenqtt.client.PublishMessage message)

Specified by:

publishReceived in interface net.sf.xenqtt.client.MqttClientListener

getUrl

public java.lang.String getUrl()

setUrl

public void setUrl(java.lang.String url)

getUsername

public java.lang.String getUsername()

setUsername

public void setUsername(java.lang.String username)

getPassword

public java.lang.String getPassword()

setPassword

public void setPassword(java.lang.String password)

isCleanSession

public boolean isCleanSession()

setCleanSession

public void setCleanSession(boolean cleanSession)

getKeepAliveSeconds

public short getKeepAliveSeconds()

setKeepAliveSeconds

public void setKeepAliveSeconds(short keepAliveInterval)

getConnectTimeoutSeconds

public int getConnectTimeoutSeconds()

setConnectTimeoutSeconds

public void setConnectTimeoutSeconds(int connectTimeoutSeconds)

getMessageResendIntervalSeconds

public int getMessageResendIntervalSeconds()

setMessageResendIntervalSeconds

public void setMessageResendIntervalSeconds(int messageResendIntervalSeconds)

getBlockingTimeoutSeconds

public int getBlockingTimeoutSeconds()

setBlockingTimeoutSeconds

public void setBlockingTimeoutSeconds(int blockingTimeoutSeconds)

getMaxInFlightMessages

public int getMaxInFlightMessages()

setMaxInFlightMessages

public void setMaxInFlightMessages(int maxInFlightMessages)

getQos

public net.sf.xenqtt.message.QoS getQos()

setQos

public void setQos(net.sf.xenqtt.message.QoS qos)