services.moleculer.transporter

Class NatsStreamingTransporter

java.lang.Object

services.moleculer.service.MoleculerComponent

services.moleculer.transporter.Transporter

services.moleculer.transporter.NatsStreamingTransporter

All Implemented Interfaces:

io.nats.client.ConnectionListener, io.nats.client.ErrorListener, io.nats.streaming.ConnectionLostHandler, io.nats.streaming.MessageHandler, MetricConstants, MoleculerLifecycle

public class NatsStreamingTransporter
extends Transporter
implements io.nats.client.ConnectionListener, io.nats.client.ErrorListener, io.nats.streaming.ConnectionLostHandler, io.nats.streaming.MessageHandler

NATS-streaming Transporter. NATS Server is a simple, high performance open source messaging system for cloud native applications, IoT messaging, and microservices architectures (website: https://nats.io). Tested with NATS streaming server version 2.1.4. Usage:

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

Required dependency:

// https://mvnrepository.com/artifact/io.nats/java-nats-streaming
compile group: 'io.nats', name: 'java-nats-streaming', version: '2.2.3'

See Also:

TcpTransporter, RedisTransporter, NatsTransporter, MqttTransporter, JmsTransporter, KafkaTransporter, AmqpTransporter

Nested Class Summary

Nested classes/interfaces inherited from interface io.nats.client.ConnectionListener

io.nats.client.ConnectionListener.Events

Field Summary

Fields

Modifier and Type	Field and Description
protected java.time.Duration	ackTimeout
protected io.nats.streaming.StreamingConnection	client Nats connection.
protected java.lang.String	clientId
protected java.lang.String	clusterId
protected java.time.Duration	connectTimeout
protected java.lang.String	discoverPrefix
protected java.lang.Integer	maxPubAcksInFlight
protected java.util.concurrent.atomic.AtomicBoolean	started
protected java.lang.String	url Nats server url.

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 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
NatsStreamingTransporter()
NatsStreamingTransporter(java.lang.String url)

Method Summary

Modifier and Type	Method and Description
void	connect()
void	connectionEvent(io.nats.client.Connection conn, io.nats.client.ConnectionListener.Events type)
void	connectionLost(io.nats.streaming.StreamingConnection conn, java.lang.Exception ex)
protected void	disconnect()
protected void	error(java.lang.Throwable cause) Any I/O error occurred.
void	errorOccurred(io.nats.client.Connection conn, java.lang.String error)
void	exceptionOccurred(io.nats.client.Connection conn, java.lang.Exception exp)
java.time.Duration	getAckTimeout()
java.lang.String	getClientId()
java.lang.String	getClusterId()
java.time.Duration	getConnectTimeout()
java.lang.String	getDiscoverPrefix()
java.lang.Integer	getMaxPubAcksInFlight()
java.time.Duration	getPingInterval()
java.lang.Integer	getPingsMaxOut()
io.nats.streaming.SubscriptionOptions	getSubscriptionOptions()
java.lang.Boolean	getTraceConnection()
java.lang.String	getUrl()
void	onMessage(io.nats.streaming.Message msg)
void	publish(java.lang.String channel, io.datatree.Tree message)
protected void	reconnect()
void	setAckTimeout(java.time.Duration ackTimeout)
void	setClientId(java.lang.String clientId)
void	setClusterId(java.lang.String clusterId)
void	setConnectTimeout(java.time.Duration connectTimeout)
void	setDiscoverPrefix(java.lang.String discoverPrefix)
void	setMaxPubAcksInFlight(java.lang.Integer maxPubAcksInFlight)
void	setPingInterval(java.time.Duration pingInterval)
void	setPingsMaxOut(java.lang.Integer pingsMaxOut)
void	setSubscriptionOptions(io.nats.streaming.SubscriptionOptions subscriptionOptions)
void	setTraceConnection(java.lang.Boolean traceConnection)
void	setUrl(java.lang.String url)
void	slowConsumerDetected(io.nats.client.Connection conn, io.nats.client.Consumer consumer)
void	stopped() Closes transporter.
io.datatree.Promise	subscribe(java.lang.String channel)

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

Methods inherited from interface io.nats.client.ErrorListener

messageDiscarded

Field Detail

url

protected java.lang.String url

Nats server url.

connectTimeout

protected java.time.Duration connectTimeout

ackTimeout

protected java.time.Duration ackTimeout

discoverPrefix

protected java.lang.String discoverPrefix

maxPubAcksInFlight

protected java.lang.Integer maxPubAcksInFlight

clientId

protected java.lang.String clientId

clusterId

protected java.lang.String clusterId

client

protected io.nats.streaming.StreamingConnection client

Nats connection.

started

protected final java.util.concurrent.atomic.AtomicBoolean started

Constructor Detail

NatsStreamingTransporter

public NatsStreamingTransporter()

NatsStreamingTransporter

public NatsStreamingTransporter(java.lang.String url)

Method Detail

connect

public void connect()

Specified by:

connect in class Transporter

disconnect

protected void disconnect()

reconnect

protected void reconnect()

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

onMessage

public void onMessage(io.nats.streaming.Message msg)

Specified by:

onMessage in interface io.nats.streaming.MessageHandler

publish

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

Specified by:

publish in class Transporter

connectionEvent

public void connectionEvent(io.nats.client.Connection conn,
                            io.nats.client.ConnectionListener.Events type)

Specified by:

connectionEvent in interface io.nats.client.ConnectionListener

errorOccurred

public void errorOccurred(io.nats.client.Connection conn,
                          java.lang.String error)

Specified by:

errorOccurred in interface io.nats.client.ErrorListener

exceptionOccurred

public void exceptionOccurred(io.nats.client.Connection conn,
                              java.lang.Exception exp)

Specified by:

exceptionOccurred in interface io.nats.client.ErrorListener

slowConsumerDetected

public void slowConsumerDetected(io.nats.client.Connection conn,
                                 io.nats.client.Consumer consumer)

Specified by:

slowConsumerDetected in interface io.nats.client.ErrorListener

connectionLost

public void connectionLost(io.nats.streaming.StreamingConnection conn,
                           java.lang.Exception ex)

Specified by:

connectionLost in interface io.nats.streaming.ConnectionLostHandler

getUrl

public java.lang.String getUrl()

Returns:

the url

setUrl

public void setUrl(java.lang.String url)

Parameters:

url - the url to set

getConnectTimeout

public java.time.Duration getConnectTimeout()

Returns:

the connectTimeout

setConnectTimeout

public void setConnectTimeout(java.time.Duration connectTimeout)

Parameters:

connectTimeout - the connectTimeout to set

getAckTimeout

public java.time.Duration getAckTimeout()

Returns:

the ackTimeout

setAckTimeout

public void setAckTimeout(java.time.Duration ackTimeout)

Parameters:

ackTimeout - the ackTimeout to set

getDiscoverPrefix

public java.lang.String getDiscoverPrefix()

Returns:

the discoverPrefix

setDiscoverPrefix

public void setDiscoverPrefix(java.lang.String discoverPrefix)

Parameters:

discoverPrefix - the discoverPrefix to set

getMaxPubAcksInFlight

public java.lang.Integer getMaxPubAcksInFlight()

Returns:

the maxPubAcksInFlight

setMaxPubAcksInFlight

public void setMaxPubAcksInFlight(java.lang.Integer maxPubAcksInFlight)

Parameters:

maxPubAcksInFlight - the maxPubAcksInFlight to set

getClientId

public java.lang.String getClientId()

Returns:

the clientId

setClientId

public void setClientId(java.lang.String clientId)

Parameters:

clientId - the clientId to set

getClusterId

public java.lang.String getClusterId()

Returns:

the clusterId

setClusterId

public void setClusterId(java.lang.String clusterId)

Parameters:

clusterId - the clusterId to set

getPingInterval

public java.time.Duration getPingInterval()

Returns:

the pingInterval

setPingInterval

public void setPingInterval(java.time.Duration pingInterval)

Parameters:

pingInterval - the pingInterval to set

getPingsMaxOut

public java.lang.Integer getPingsMaxOut()

Returns:

the pingsMaxOut

setPingsMaxOut

public void setPingsMaxOut(java.lang.Integer pingsMaxOut)

Parameters:

pingsMaxOut - the pingsMaxOut to set

getTraceConnection

public java.lang.Boolean getTraceConnection()

Returns:

the traceConnection

setTraceConnection

public void setTraceConnection(java.lang.Boolean traceConnection)

Parameters:

traceConnection - the traceConnection to set

getSubscriptionOptions

public io.nats.streaming.SubscriptionOptions getSubscriptionOptions()

Returns:

the subscriptionOptions

setSubscriptionOptions

public void setSubscriptionOptions(io.nats.streaming.SubscriptionOptions subscriptionOptions)

Parameters:

subscriptionOptions - the subscriptionOptions to set