Networking

To enable communication between nodes (ServiceBrokers), you must configure a transporter. A transporter is Moleculer’s transport layer / messaging backend: it carries requests, responses and events between processes over a message broker (NATS, Redis, Kafka, MQTT, AMQP) or, with the TCP transporter, directly between nodes without any broker. The available transporters typically connect to a central message broker, facilitating reliable message exchange among remote nodes. These message brokers primarily support the publish/subscribe messaging pattern.

Terminology

In these docs ServiceBroker / the broker (as in broker.call() or “broker option”) means Moleculer’s ServiceBroker instance running in your process; message broker means the external messaging server such as NATS or RabbitMQ that a transporter connects to.

Because the transporter is an abstraction, the same service code runs unchanged over any of them — you choose the broker for operational reasons (what you already run, throughput, persistence), not because your code depends on it.

Transporters

The transporter module in Moleculer plays a critical role in facilitating communication between services running on multiple nodes. It manages the transmission of events, request calls, and processing responses among nodes in the network. Note that the transporter only carries the packets; deciding which node receives a request is done by the ServiceBroker’s registry (client-side load balancing) before the packet is handed to the transporter. The exception is the disabled balancer mode, where the message broker’s own queue-group balancing is used.

Abstraction for Seamless Switching

The communication logic is abstracted away from the transporter class itself. This allows for seamless switching between different transporters without requiring any modifications to your service code. You can choose the most suitable transporter for your needs (e.g., TCP, NATS, Redis) without impacting your service logic, offering flexibility and future-proofing for your application.

TCP transporter

This is a no-dependency, zero-configuration TCP transporter. It uses Gossip protocol to disseminate node statuses, service list and heartbeats. It contains an integrated UDP discovery feature to detect new and disconnected nodes on the network.
If the UDP is prohibited on your network, use urls option. It is a list of remote endpoints (host/ip, port, nodeID). It can be a static list in your configuration or a file path which contains the list.

Use TCP transporter with default options

// moleculer.config.js
module.exports = {
    transporter: "TCP"
};

All TCP transporter options with default values

// moleculer.config.js
module.exports = {
    logger: true,
    transporter: {
        type: "TCP",
        options: {
            // Enable UDP discovery
            udpDiscovery: true,
            // Reusing UDP server socket
            udpReuseAddr: true,

            // UDP port
            udpPort: 4445,
            // UDP bind address (if null, bind on all interfaces)
            udpBindAddress: null,
            // UDP sending period (seconds)
            udpPeriod: 30,

            // Multicast address.
            udpMulticast: "239.0.0.0",
            // Multicast TTL setting
            udpMulticastTTL: 1,

            // Send broadcast (Boolean, String, Array<String>)
            udpBroadcast: false,

            // TCP server port. Null or 0 means random port
            port: null,
            // Static remote nodes address list (when UDP discovery is not available)
            urls: null,
            // Use hostname as preffered connection address
            useHostname: true,

            // Gossip sending period in seconds
            gossipPeriod: 2,
            // Maximum enabled outgoing connections. If reach, close the old connections
            maxConnections: 32,
            // Maximum TCP packet size
            maxPacketSize: 1 * 1024 * 1024            
        }
    }
};

TCP transporter with static endpoint list

// moleculer.config.js
module.exports = {
    nodeID: "node-1",
    logger: true,
    transporter: {
        type: "TCP",
        options: {
            udpDiscovery: false,
            urls: [
                "172.17.0.1:6000/node-1",
                "172.17.0.2:6000/node-2",
                "172.17.0.3:6000/node-3"                
            ],
        }
    }
};

You don’t need to set port because it find & parse the self TCP port from URL list.

TCP transporter with shorthand static endpoint list
It needs to start with tcp://.

// moleculer.config.js
module.exports = {
    nodeID: "node-1",
    transporter: "tcp://172.17.0.1:6000/node-1,172.17.0.2:6000/node-2,172.17.0.3:6000/node-3"
};

TCP transporter with static endpoint list file

// moleculer.config.js
module.exports = {
    nodeID: "node-1",
    transporter: {
        type: "TCP",
        options: {
            udpDiscovery: false,
            urls: "file://./nodes.json"
        }
    }
};

// nodes.json
[
    "127.0.0.1:6001/client-1",
    "127.0.0.1:7001/server-1",
    "127.0.0.1:7002/server-2"
]

Serviceless node

Please note, you don’t need to list all remote nodes. It’s enough at least one node which is online. For example, create a “serviceless” gossiper node, which does nothing, just shares other remote nodes addresses by gossip messages. So all nodes must know only the gossiper node address to be able to communicate with all other nodes.

NATS Transporter

Built-in transporter for NATS.

NATS Server is a simple, high performance open source messaging system for cloud-native applications, IoT messaging, and microservices architectures.

// moleculer.config.js
module.exports = {
    nodeID: "server-1",
    transporter: "nats://nats.server:4222"
};

Dependencies

To use this transporter install the nats module with npm install nats --save command.

Examples

Connect to ‘nats://localhost:4222’

// moleculer.config.js
module.exports = {
    transporter: "NATS"
};

Connect to a remote NATS server

// moleculer.config.js
module.exports = {
    transporter: "nats://nats-server:4222"
};

Connect to a remote NATS server with auth

// moleculer.config.js
module.exports = {
    transporter: "nats://user:pass@nats-server:4222"
};

Connect with options

// moleculer.config.js
module.exports = {
    transporter: {
        type: "NATS",
        options: {
            servers: ["nats://localhost:4222"],
            user: "admin",
            pass: "1234"
        }
    }
};

Connect with TLS

// moleculer.config.js
module.exports = {
    transporter: {
        type: "NATS",
        options: {
            servers: ["nats://localhost:4222"]
            // More info: https://github.com/nats-io/node-nats#tls
            tls: {
                key: fs.readFileSync('./client-key.pem'),
                cert: fs.readFileSync('./client-cert.pem'),
                ca: [ fs.readFileSync('./ca.pem') ]
            }
        }
    }
};

Redis Transporter

Built-in transporter for Redis.

// moleculer.config.js
module.exports = {
    nodeID: "server-1",
    transporter: "redis://redis.server:6379"
};

Dependencies

To use this transporter install the ioredis module with npm install ioredis --save command.

Examples

Connect with default settings

// moleculer.config.js
module.exports = {
    transporter: "Redis"
};

Connect with connection string

// moleculer.config.js
module.exports = {
    transporter: "redis://localhost:6379"
};

Connect to a secure Redis server

// moleculer.config.js
module.exports = {
    transporter: "rediss://localhost:6379"
};

Connect with options

// moleculer.config.js
module.exports = {
    transporter: {
        type: "Redis",
        options: {
            host: "redis-server",
            db: 0
        }
    }
};

Connect to Redis cluster

// moleculer.config.js
module.exports = {
    transporter: {
        type: "Redis",
        options: {
            cluster: {
                nodes: [
                    { host: "localhost", port: 6379 },
                    { host: "localhost", port: 6378 }
                ]
            }
        }
    }
};

MQTT Transporter

Built-in transporter for MQTT protocol (e.g.: Mosquitto).

// moleculer.config.js
module.exports = {
    nodeID: "server-1",
    transporter: "mqtt://mqtt-server:1883"
};

Dependencies

To use this transporter install the mqtt module with npm install mqtt --save command.

Examples

Connect with default settings

// moleculer.config.js
module.exports = {
    transporter: "MQTT"
};

Connect with connection string

// moleculer.config.js
module.exports = {
    transporter: "mqtt://mqtt-server:1883"
};

Connect to secure MQTT server

// moleculer.config.js
module.exports = {
    transporter: "mqtts://mqtt-server:1883"
};

Connect with options

// moleculer.config.js
module.exports = {
    transporter: {
        type: "MQTT",
        options: {
            host: "mqtt-server",
            port: 1883,
            qos: 0,
            topicSeparator: "."
        }
    }
};

AMQP (0.9) Transporter

Built-in transporter for AMQP 0.9 protocol (e.g.: RabbitMQ).

// moleculer.config.js
module.exports = {
    nodeID: "server-1",
    transporter: "amqp://rabbitmq-server:5672"
};

Dependencies

To use this transporter install the amqplib module with npm install amqplib --save command.

Transporter options

Options can be passed to amqp.connect() method.

Connect to ‘amqp://guest:guest@localhost:5672’

// moleculer.config.js
module.exports = {
    transporter: "AMQP"
};

Connect to a remote server

// moleculer.config.js
module.exports = {
    transporter: "amqp://rabbitmq-server:5672"
};

Connect to a secure server

// moleculer.config.js
module.exports = {
    transporter: "amqps://rabbitmq-server:5672"
};

Connect to a remote server with options & credentials

// moleculer.config.js
module.exports = {
    transporter: {
        type: "AMQP",
        options: {
            url: "amqp://user:pass@rabbitmq-server:5672",
            eventTimeToLive: 5000,
            prefetch: 1,
            socketOptions: {
                servername: process.env.RABBIT_SERVER_NAME
            }
            // If true, queues will be autodeleted once service is stopped, i.e., queue listener is removed
            autoDeleteQueues: true
        }
    }
};

AMQP 1.0 Transporter

Built-in transporter for AMQP 1.0 protocol (e.g.: ActiveMq or RabbitMQ + rabbitmq-amqp1.0 plugin).

Please note, it is an experimental transporter. Do not use it in production yet!

// moleculer.config.js
module.exports = {
    transporter: "amqp10://activemq-server:5672"
};

Dependencies

To use this transporter install the rhea-promise module with npm install rhea-promise --save command.

Transporter options

Options can be passed to rhea.connection.open() method, the topics, the queues, and the messages themselves.

Connect to ‘amqp10://guest:guest@localhost:5672’

// moleculer.config.js
module.exports = {
    transporter: "AMQP10"
};

Connect to a remote server

// moleculer.config.js
module.exports = {
    transporter: "amqp10://activemq-server:5672"
};

Connect to a remote server with options & credentials

// moleculer.config.js
module.exports = {
    transporter: {
        type: "AMQP10",
        options: {
            url: "amqp10://user:pass@activemq-server:5672",
            eventTimeToLive: 5000,
            heartbeatTimeToLive: 5000,
            connectionOptions: { // rhea connection options https://github.com/amqp/rhea#connectoptions, example:
                ca: "", // (if using tls)
                servername: "", // (if using tls)
                key: "", // (if using tls with client auth)
                cert: "" // (if using tls with client auth)
            },
            queueOptions: {}, // rhea queue options https://github.com/amqp/rhea#open_receiveraddressoptions
            topicOptions: {}, // rhea queue options https://github.com/amqp/rhea#open_receiveraddressoptions
            messageOptions: {}, // rhea message specific options https://github.com/amqp/rhea#message
            topicPrefix: "topic://", // RabbitMq uses '/topic/' instead, 'topic://' is more common
            prefetch: 1
        }
    }
};

Kafka Transporter

Built-in transporter for Kafka.

It is a simple implementation. It transfers Moleculer packets to consumers via pub/sub. There are not implemented offset, replay…etc features.

Dependencies

To use this transporter install the @platformatic/kafka module with npm install @platformatic/kafka --save command. Both @platformatic/kafka 1.x (>=1.18.0) and 2.x are supported (2.x requires Moleculer >=0.15.2).

Connect to Kafka broker

// moleculer.config.js
module.exports = {
    transporter: "kafka://localhost:9092"
};

Connect to Kafka with custom options

// moleculer.config.js
module.exports = {
    transporter: {
        type: "Kafka",
        options: {
            // Client ID for all clients
            clientId: "moleculer-kafka",

            // Bootstrap brokers for connection
            bootstrapBrokers: ["localhost:9092"],

            // Producer options
            producer: {},

            // Consumer options
            consumer: {},

            // Admin options
            admin: {},

            // Advanced options for `send`
            publish: {},

            // Advanced message options for `send`
            publishMessage: {
                partition: 0
            }
        }
    }
};

Custom transporter

Custom transporter module can be created. We recommend to copy the source of NatsTransporter and implement the connect, disconnect, subscribe and send methods.

Create custom transporter

const BaseTransporter = require("moleculer").Transporters.Base;

class MyTransporter extends BaseTransporter {
    connect() { /*...*/ }
    disconnect() { /*...*/ }
    subscribe() { /*...*/ }
    send() { /*...*/ }
}

Use custom transporter

// moleculer.config.js
const MyTransporter = require("./my-transporter");

module.exports = {
    transporter: new MyTransporter()
};

Disabled balancer

Some transporter servers have built-in balancer solution. E.g.: RabbitMQ, NATS. If you want to use the transporter balancer instead of Moleculer balancer, set the disableBalancer broker option to true.

Only the NATS, AMQP (0.9) and AMQP 1.0 transporters have a built-in balancer (hasBuiltInBalancer). If you set disableBalancer: true with any other transporter (TCP, Redis, MQTT, Kafka), the broker logs a warning (The ... has no built-in balancer. Broker balancer is ENABLED.) and forces disableBalancer back to false.

When the balancer is disabled, the broker no longer selects the target node: the registry strategies and the preferLocal option are not applied, the request is published to the message broker and its queue-group mechanism picks a consumer.

Example

// moleculer.config.js
module.exports = {
    disableBalancer: true,
    transporter: "nats://some-server:4222"
};

Please note

If you disable the built-in Moleculer balancer, all requests & events will be transferred via transporter (including local requests). E.g. you have a local math service and you call math.add locally, the request will be sent via transporter.

Serialization

Transporter needs a serializer module which serializes & deserializes the transferred packets. The default serializer is the JSONSerializer but there are several built-in serializer.

Note that certain data types (e.g., Date, Map, BigInt) cannot be serialized with native JSON serializer. If you are working with this kind of data consider using JSONExt or Notepack serializers.

Example

// moleculer.config.js
module.exports = {
    nodeID: "server-1",
    transporter: "NATS",
};

JSON serializer

This is the default serializer. It serializes the packets to JSON string and deserializes the received data to packet.

// moleculer.config.js
module.exports = {
    serializer: "JSON" // not necessary to set, because it is the default
};

JSON Extended serializer

We implemented a new JSON serializer which (unlike the native JSON serializer) is able to serializer Buffer, BigInt, Date, Map, Set and RegExp classes.

Example

// moleculer.config.js
module.exports = {
    serializer: "JSONExt"
}

Custom extensions

You can extend the serializer with custom types.

Example to extend with a custom class serializing/deserializing

class MyClass {
  constructor(a, b) {
    this.a = a;
    this.b = b;
  }
}

// moleculer.config.js
module.exports = {
  serializer: {
    type: "JSONExt",
    options: {
      customs: [
        {
          // This is the identifier of the custom type
          prefix: "AB",

          // This function checks the type of JSON value
          check: (v) => v instanceof MyClass,

          // Serialize the custom class properties to a String
          serialize: (v) => v.a + "|" + v.b,

          // Deserialize the JSON string to custom class instance and set properties
          deserialize: (v) => {
            const [a, b] = v.split("|");
            return new MyClass(parseInt(a), b);
          },
        },
      ],
    },
  },
};

MsgPack serializer

Built-in MsgPack serializer.

// moleculer.config.js
module.exports = {
    serializer: "MsgPack"
};

Dependencies

To use this serializer install the msgpack5 module with npm install msgpack5 --save command.

Notepack serializer

Built-in Notepack serializer.

// moleculer.config.js
module.exports = {
    serializer: "Notepack"
};

Dependencies

To use this serializer install the notepack module with npm install notepack.io --save command.

CBOR serializer

CBOR (cbor-x) is the fastest than any other serializers.

Example

// moleculer.config.js
module.exports = {
    logger: true,
    serializer: "CBOR"
};

Custom serializer

Custom serializer module can be created. We recommend to copy the source of JSONSerializer and implement the serialize and deserialize methods.

Create custom serializer

const BaseSerializer = require("moleculer").Serializers.Base;

class MySerializer extends BaseSerializer {
    serialize(obj, type) { /*...*/ }
    deserialize(buf, type) { /*...*/ }
}

Use custom serializer

// moleculer.config.js
const MySerializer = require("./my-serializer");

module.exports = {
    serializer: new MySerializer()
};