services.moleculer.transporter

Class TcpTransporter

java.lang.Object

services.moleculer.service.MoleculerComponent

services.moleculer.transporter.Transporter

services.moleculer.transporter.TcpTransporter

All Implemented Interfaces:

MetricConstants, MoleculerLifecycle

public class TcpTransporter
extends Transporter

TCP Transporter with optional UDP discovery ("zero configuration") module.

TCP Transporter uses fault tolerant and peer-to-peer Gossip Protocol to discover location and service information about the other nodes participating in a Moleculer Cluster. In Moleculer's P2P architecture all nodes are equal, there is no "leader" or "controller" node, so the cluster is truly horizontally scalable. This transporter aims to run on top of an infrastructure of hundreds of nodes.

Nodes can explore each other in two ways. With or without UDP packets. If the URLs of all nodes specified in startup parameters (in "tcp://host:port/nodeID" format), TCP Transporter will work without UDP. Otherwise the TCP Transporter starts an UDP server, and the Moleculer nodes detect each other with UDP packets. Events and function calls always go through TCP channels.

TCP Transporter provides the highest speed data transfer between the nodes (eg. hundred thousand packets per second can be transmitted from one node to another over a high-speed LAN). Usage:

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

// *

See Also:

RedisTransporter, NatsTransporter, NatsStreamingTransporter, MqttTransporter, JmsTransporter, KafkaTransporter, AmqpTransporter

Field Summary

Fields

Modifier and Type	Field and Description
protected NodeDescriptor	cachedDescriptor Current node descriptor
protected byte[]	cachedHelloMessage
protected int	currentPort Current TCP port
protected java.util.concurrent.ScheduledFuture<?>	gossiperTimer Cancelable timer of gossiper
protected int	gossipPeriod Gossiping period time, in SECONDS.
protected UDPLocator	locator UDP broadcaster
protected int	maxConnections Max number of keep-alive connections (-1 = unlimited, 0 = disable keep-alive connections).
protected int	maxPacketSize Max enable packet size (BYTES).
protected static byte	PACKET_EVENT_ID
protected static byte	PACKET_GOSSIP_HELLO_ID
protected static byte	PACKET_GOSSIP_REQ_ID
protected static byte	PACKET_GOSSIP_RSP_ID
protected static byte	PACKET_PING_ID
protected static byte	PACKET_PONG_ID
protected static byte	PACKET_REQUEST_ID
protected static byte	PACKET_RESPONSE_ID
protected int	port TCP port (used by the Transporter and Gossiper services).
protected TcpReader	reader Socket reader
protected java.util.Random	rnd Random generator.
protected java.lang.String	startupHost The local hostname specified in the "urls" parameter.
protected java.util.concurrent.atomic.AtomicLong	timestamp Current node descriptor's timestamp
protected java.lang.String	udpBindAddress UDP bind address (null = autodetect)
protected boolean	udpBroadcast Use UDP broadcast WITH UDP multicast (false = use UDP multicast only)
protected int	udpMaxDiscovery Maximum number of outgoing multicast packets (0 = runs forever)
protected java.lang.String	udpMulticast UDP multicast address of automatic discovery service.
protected int	udpMulticastTTL TTL of UDP multicast packets
protected int	udpPeriod UDP broadcast/multicast period in SECONDS
protected int	udpPort UDP broadcast/multicast port
protected boolean	udpReuseAddr Resuse addresses
protected java.lang.String[]	urls List of URLs ("tcp://host:port/nodeID" or "host:port/nodeID" or "host/nodeID"), when UDP discovery is disabled.
protected boolean	useHostname Use hostnames instead of IP addresses As the DHCP environment is dynamic, any later attempt to use IPs instead hostnames would most likely yield false results.
protected TcpWriter	writer Socket writer

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
TcpTransporter() Start TCP Transporter in "zero config" mode, with automatic UDP service discovery.
TcpTransporter(java.lang.String... urls) Start TCP Transporter in full TCP mode, without UDP discovery.
TcpTransporter(java.net.URL urlList) Start TCP Transporter in full TCP mode, without UDP discovery.

Method Summary

Modifier and Type	Method and Description
void	broadcastInfoPacket()
void	connect()
protected void	disconnect()
byte[]	generateGossipHello() Create Gossip HELLO packet.
int	getCurrentPort()
NodeDescriptor	getDescriptor()
int	getGossipPeriod()
int	getMaxConnections()
int	getMaxPacketSize()
int	getPort()
java.lang.String	getUdpBindAddress()
int	getUdpMaxDiscovery()
java.lang.String	getUdpMulticast()
int	getUdpMulticastTTL()
int	getUdpPeriod()
int	getUdpPort()
java.lang.String[]	getUrls()
boolean	isUdpBroadcast()
boolean	isUdpReuseAddr()
boolean	isUseHostname()
protected void	processGossipHello(io.datatree.Tree data)
protected io.datatree.Tree	processGossipRequest(io.datatree.Tree data)
protected void	processGossipResponse(io.datatree.Tree data)
void	publish(java.lang.String channel, io.datatree.Tree message)
void	received(byte packetID, byte[] packet)
protected void	reconnect()
protected void	registerAsNewNode(java.lang.String sender, java.lang.String host, int port)
protected void	sendDisconnectPacket()
protected io.datatree.Tree	sendGossipRequest() Create and send a Gossip request packet.
protected void	sendGossipToRandomEndpoint(java.lang.String[] endpoints, int size, byte[] packet, io.datatree.Tree message)
protected byte[]	serialize(byte packetID, io.datatree.Tree message)
void	setGossipPeriod(int gossipPeriod)
void	setHeartbeatInterval(int heartbeatInterval)
void	setHeartbeatTimeout(int heartbeatTimeout)
void	setMaxConnections(int maxConnections)
void	setMaxPacketSize(int maxPacketSize)
void	setPort(int port)
void	setUdpBindAddress(java.lang.String udpBindAddress)
void	setUdpBroadcast(boolean udpBroadcast)
void	setUdpMaxDiscovery(int udpMaxDiscovery)
void	setUdpMulticast(java.lang.String udpMulticast)
void	setUdpMulticastTTL(int udpMulticastTTL)
void	setUdpPeriod(int udpPeriod)
void	setUdpPort(int udpPort)
void	setUdpReuseAddr(boolean udpReuseAddr)
void	setUrls(java.lang.String... urls)
void	setUseHostname(boolean preferHostname)
void	stopped() Closes TCP Transporter.
io.datatree.Promise	subscribe(java.lang.String channel)
void	udpPacketReceived(java.lang.String sender, java.lang.String host, int port)
void	unableToSend(java.lang.String nodeID, java.util.LinkedList<byte[]> packets, java.lang.Throwable cause)

Methods inherited from class services.moleculer.transporter.Transporter

broadcastNodeConnected, broadcastNodeDisconnected, broadcastNodeUpdated, broadcastTransporterConnected, broadcastTransporterDisconnected, channel, checkTimeouts, connected, connected, createPingPacket, error, getAddress, getAllNodeIDs, getCpuUsage, getDescriptor, getHeartbeatInterval, getHeartbeatTimeout, getInstanceID, getLastHeartbeatTime, getNamespace, getOfflineTimeout, getPrefix, getSerializer, getSubscriptionTimeout, isDebug, isDebugHeartbeats, isOnline, isPreferHostname, processReceivedMessage, processReceivedMessage, publish, received, scheduleInfoPacket, sendClosePacket, sendDataPacket, sendDiscoverPacket, sendErrorPacket, sendEventPacket, sendHeartbeatPacket, sendInfoPacket, sendPongPacket, sendRequestPacket, setDebug, setDebugHeartbeats, 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

PACKET_EVENT_ID

protected static final byte PACKET_EVENT_ID

See Also:

Constant Field Values

PACKET_REQUEST_ID

protected static final byte PACKET_REQUEST_ID

See Also:

Constant Field Values

PACKET_RESPONSE_ID

protected static final byte PACKET_RESPONSE_ID

See Also:

Constant Field Values

PACKET_PING_ID

protected static final byte PACKET_PING_ID

See Also:

Constant Field Values

PACKET_PONG_ID

protected static final byte PACKET_PONG_ID

See Also:

Constant Field Values

PACKET_GOSSIP_REQ_ID

protected static final byte PACKET_GOSSIP_REQ_ID

See Also:

Constant Field Values

PACKET_GOSSIP_RSP_ID

protected static final byte PACKET_GOSSIP_RSP_ID

See Also:

Constant Field Values

PACKET_GOSSIP_HELLO_ID

protected static final byte PACKET_GOSSIP_HELLO_ID

See Also:

Constant Field Values

port

protected int port

TCP port (used by the Transporter and Gossiper services). A port number of zero will let the system pick up an ephemeral port in a bind operation.

gossipPeriod

protected int gossipPeriod

Gossiping period time, in SECONDS.

maxConnections

protected int maxConnections

Max number of keep-alive connections (-1 = unlimited, 0 = disable keep-alive connections).

maxPacketSize

protected int maxPacketSize

Max enable packet size (BYTES).

urls

protected java.lang.String[] urls

List of URLs ("tcp://host:port/nodeID" or "host:port/nodeID" or "host/nodeID"), when UDP discovery is disabled.

useHostname

protected boolean useHostname

Use hostnames instead of IP addresses As the DHCP environment is dynamic, any later attempt to use IPs instead hostnames would most likely yield false results. Therefore, use hostnames if you are using DHCP.

udpPort

protected int udpPort

UDP broadcast/multicast port

udpBindAddress

protected java.lang.String udpBindAddress

UDP bind address (null = autodetect)

udpPeriod

protected int udpPeriod

UDP broadcast/multicast period in SECONDS

udpReuseAddr

protected boolean udpReuseAddr

Resuse addresses

udpMaxDiscovery

protected int udpMaxDiscovery

Maximum number of outgoing multicast packets (0 = runs forever)

udpMulticast

protected java.lang.String udpMulticast

UDP multicast address of automatic discovery service.

udpMulticastTTL

protected int udpMulticastTTL

TTL of UDP multicast packets

udpBroadcast

protected boolean udpBroadcast

Use UDP broadcast WITH UDP multicast (false = use UDP multicast only)

rnd

protected final java.util.Random rnd

Random generator.

gossiperTimer

protected volatile java.util.concurrent.ScheduledFuture<?> gossiperTimer

Cancelable timer of gossiper

reader

protected TcpReader reader

Socket reader

writer

protected TcpWriter writer

Socket writer

locator

protected UDPLocator locator

UDP broadcaster

currentPort

protected int currentPort

Current TCP port

startupHost

protected java.lang.String startupHost

The local hostname specified in the "urls" parameter.

cachedDescriptor

protected NodeDescriptor cachedDescriptor

Current node descriptor

timestamp

protected java.util.concurrent.atomic.AtomicLong timestamp

Current node descriptor's timestamp

cachedHelloMessage

protected byte[] cachedHelloMessage

Constructor Detail

TcpTransporter

public TcpTransporter()

Start TCP Transporter in "zero config" mode, with automatic UDP service discovery.

TcpTransporter

public TcpTransporter(java.lang.String... urls)

Start TCP Transporter in full TCP mode, without UDP discovery. Valid URL syntax is "tcp://host:port/nodeID" or "host:port/nodeID".

Parameters:

urls - list of urls of ALL nodes

TcpTransporter

public TcpTransporter(java.net.URL urlList)
               throws java.lang.Exception

Start TCP Transporter in full TCP mode, without UDP discovery. Loads node list from an URL (as an http or file resource in JSON/XML/YAML format). Sample JSON file:

{
"nodes":[
"tcp://host1:port1/nodeID1",
"tcp://host2:port2/nodeID2",
"tcp://host3:port3/nodeID3"
]
}

Parameters:

urlList - an URL, where the peer configuration's JSON is located

Throws:

java.lang.Exception - URL format exception or any I/O error

Method Detail

connect

public void connect()

Specified by:

connect in class Transporter

disconnect

protected void disconnect()

reconnect

protected void reconnect()

stopped

public void stopped()

Closes TCP Transporter.

Specified by:

stopped in interface MoleculerLifecycle

Overrides:

stopped in class Transporter

received

public void received(byte packetID,
                     byte[] packet)

unableToSend

public void unableToSend(java.lang.String nodeID,
                         java.util.LinkedList<byte[]> packets,
                         java.lang.Throwable cause)

sendDisconnectPacket

protected void sendDisconnectPacket()

Overrides:

sendDisconnectPacket in class Transporter

subscribe

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

Specified by:

subscribe in class Transporter

publish

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

Specified by:

publish in class Transporter

serialize

protected byte[] serialize(byte packetID,
                           io.datatree.Tree message)
                    throws java.lang.Exception

Throws:

java.lang.Exception

getDescriptor

public NodeDescriptor getDescriptor()

udpPacketReceived

public void udpPacketReceived(java.lang.String sender,
                              java.lang.String host,
                              int port)

processGossipHello

protected void processGossipHello(io.datatree.Tree data)

registerAsNewNode

protected void registerAsNewNode(java.lang.String sender,
                                 java.lang.String host,
                                 int port)

sendGossipRequest

protected io.datatree.Tree sendGossipRequest()

Create and send a Gossip request packet.

Returns:

created Gossip request (used for testing)

sendGossipToRandomEndpoint

protected void sendGossipToRandomEndpoint(java.lang.String[] endpoints,
                                          int size,
                                          byte[] packet,
                                          io.datatree.Tree message)
                                   throws java.lang.Exception

Throws:

java.lang.Exception

processGossipRequest

protected io.datatree.Tree processGossipRequest(io.datatree.Tree data)
                                         throws java.lang.Exception

Throws:

java.lang.Exception

processGossipResponse

protected void processGossipResponse(io.datatree.Tree data)
                              throws java.lang.Exception

Throws:

java.lang.Exception

generateGossipHello

public byte[] generateGossipHello()

Create Gossip HELLO packet. Hello message is invariable, so we can cache it.

Returns:

created "hello" request

broadcastInfoPacket

public void broadcastInfoPacket()

Overrides:

broadcastInfoPacket in class Transporter

setHeartbeatInterval

public void setHeartbeatInterval(int heartbeatInterval)

Overrides:

setHeartbeatInterval in class Transporter

setHeartbeatTimeout

public void setHeartbeatTimeout(int heartbeatTimeout)

Overrides:

setHeartbeatTimeout in class Transporter

getUrls

public java.lang.String[] getUrls()

setUrls

public void setUrls(java.lang.String... urls)

getPort

public int getPort()

setPort

public void setPort(int port)

getGossipPeriod

public int getGossipPeriod()

setGossipPeriod

public void setGossipPeriod(int gossipPeriod)

getMaxConnections

public int getMaxConnections()

setMaxConnections

public void setMaxConnections(int maxConnections)

getMaxPacketSize

public int getMaxPacketSize()

setMaxPacketSize

public void setMaxPacketSize(int maxPacketSize)

getCurrentPort

public int getCurrentPort()

isUseHostname

public boolean isUseHostname()

setUseHostname

public void setUseHostname(boolean preferHostname)

isUdpBroadcast

public boolean isUdpBroadcast()

setUdpBroadcast

public void setUdpBroadcast(boolean udpBroadcast)

getUdpMaxDiscovery

public int getUdpMaxDiscovery()

setUdpMaxDiscovery

public void setUdpMaxDiscovery(int udpMaxDiscovery)

getUdpPeriod

public int getUdpPeriod()

setUdpPeriod

public void setUdpPeriod(int udpPeriod)

isUdpReuseAddr

public boolean isUdpReuseAddr()

setUdpReuseAddr

public void setUdpReuseAddr(boolean udpReuseAddr)

getUdpPort

public int getUdpPort()

setUdpPort

public void setUdpPort(int udpPort)

getUdpBindAddress

public java.lang.String getUdpBindAddress()

setUdpBindAddress

public void setUdpBindAddress(java.lang.String udpBindAddress)

getUdpMulticast

public java.lang.String getUdpMulticast()

setUdpMulticast

public void setUdpMulticast(java.lang.String udpMulticast)

getUdpMulticastTTL

public int getUdpMulticastTTL()

setUdpMulticastTTL

public void setUdpMulticastTTL(int udpMulticastTTL)