Progressive microservices framework for Node.js
MIT License
Published by icebob over 6 years ago
Thanks for @rmccallum81, ServiceBroker supports graceful shutdown. You can enable it with trackContext
broker option. If you enable it, all services wait for all running contexts before shutdowning. You can also define a timeout value with gracefulStopTimeout
broker option.
const broker = new ServiceBroker({
trackContext: true,
gracefulStopTimeout: 5 * 1000 // waiting max 5 sec
});
This timeout can be overwrite in service settings with $gracefulStopTimeout
property.
Published by icebob over 6 years ago
mixins: []
).Published by icebob over 6 years ago
This strategy selects a node which has the lowest latency, measured by periodic PING
. Notice that the strategy only ping one of nodes from a single host. Due to the node list can be very long, it gets samples and selects the host with the lowest latency from only samples instead of the whole node list.
Usage
let broker = new ServiceBroker({
registry: {
strategy: "Latency"
}
});
Strategy options
Name | Type | Default | Description |
---|---|---|---|
sampleCount |
Number |
5 |
the number of samples. If you have a lot of hosts/nodes, it's recommended to increase the value. |
lowLatency |
Number |
10 |
the low latency (ms). The node which has lower latency than this value is selected immediately. |
collectCount |
Number |
5 |
the number of measured latency per host to keep in order to calculate the average latency. |
pingInterval |
Number |
10 |
ping interval (s). If you have a lot of host/nodes, it's recommended to increase the value. |
Usage with custom options
let broker = new ServiceBroker({
registry: {
strategy: "Latency",
strategyOptions: {
sampleCount: 15,
lowLatency: 20,
collectCount: 10,
pingInterval: 15
}
}
});
Thanks for @zllovesuki!
There is a new Moleculer Runner option --mask
to define filemask when load all services from folders.
Example
$ moleculer-runner.js -r --mask **/user*.service.js examples
Example to load Typescript services
$ node -r ts-node/register node_modules/moleculer/bin/moleculer-runner --hot --repl --mask **/*.service.ts services
d.ts
issuesgroup
handling in mixins (#217)mergeSchemas
from utils
to Service
static method. It can be overwritten in a custom ServiceFactoryd.ts
prefix
option in Redis Cacher (223)nanomatch
dependency, use own implementationrequire("moleculer").Utils
mergeSchemas
static method in Service
class.SERVICES
& SERVICEDIR
env vars overwrites the paths in CLI arguments.Published by icebob over 6 years ago
This version contains the most changes in the history of Moleculer! More than 200 commits with 17k additions and a lot of new features.
The Github organization name (Ice Services) has been renamed to MoleculerJS. Please update your bookmarks.
To support #188, mixin merging logic is changed at actions
. Now it uses defaultsDeep
for merging. It means you can extend the actions definition of mixins, no need to redeclare the handler
.
Add extra action properties but handler
is untouched
// mixin.service.js
module.exports = {
actions: {
create(ctx) {
// Action handler without `params`
}
}
};
// my.service.js
module.exports = {
mixins: [MixinService]
actions: {
create: {
// Add only `params` property to the `create` action
// The handler is merged from mixin
params: {
name: "string"
}
}
}
};
If you are using transporter options, you will need to migrate them. The transporter specific wrapper has been removed from options (nats
, redis
, mqtt
, amqp
).
Before
// NATS transporter
const broker = new ServiceBroker({
transporter: {
type: "NATS",
options: {
nats: {
user: "admin",
pass: "1234"
}
}
}
});
// Redis transporter
const broker = new ServiceBroker({
transporter: {
type: "Redis",
options: {
redis: {
port: 6379,
db: 0
}
}
}
});
// MQTT transporter
const broker = new ServiceBroker({
transporter: {
type: "MQTT",
options: {
mqtt: {
user: "admin",
pass: "1234"
}
}
}
});
// AMQP transporter
const broker = new ServiceBroker({
transporter: {
type: "AMQP",
options: {
amqp: {
prefetch: 1
}
}
}
});
After
// NATS transporter
const broker = new ServiceBroker({
transporter: {
type: "NATS",
options: {
user: "admin",
pass: "1234"
}
}
});
// Redis transporter
const broker = new ServiceBroker({
transporter: {
type: "Redis",
options: {
port: 6379,
db: 0
}
}
});
// MQTT transporter
const broker = new ServiceBroker({
transporter: {
type: "MQTT",
options: {
user: "admin",
pass: "1234"
}
}
});
// AMQP transporter
const broker = new ServiceBroker({
transporter: {
type: "AMQP",
options: {
prefetch: 1
}
}
});
nodeID
generator changedWhen nodeID
didn't define in broker options, the broker generated it from hostname (os.hostname()
). It could cause problem for new users when they tried to start multiple instances on the same computer. Therefore, the broker generates nodeID
from hostname and process PID. The newly generated nodeID looks like server-6874
where server
is the hostname and 6874
is the PID.
The transport protocol is changed. The new version is 3
. Check the changes.
It means, the >=0.12.x versions can't communicate with old <=0.11 versions.
Changes:
RESPONSE
packet has a new field meta
.EVENT
packet has a new field broadcast
.port
field is removed from INFO
packet.INFO
packet has a new field hostname
.There are some new properties in ServiceBroker option: middlewares
, created
, started
, stopped
.
They can be useful when you use broker config file and start your project with Moleculer Runner.
// moleculer.config.js
module.exports = {
logger: true,
// Add middlewares
middlewares: [myMiddleware()],
// Fired when the broker created
created(broker) {
},
// Fired when the broker started
started(broker) {
// You can return Promise
return broker.Promise.resolve();
},
// Fired when the broker stopped
stopped(broker) {
// You can return Promise
return broker.Promise.resolve();
}
};
The broker.broadcast
function has a third groups
argument similar to broker.emit
.
// Send to all "mail" service instances
broker.broadcast("user.created", { user }, "mail");
// Send to all "user" & "purchase" service instances.
broker.broadcast("user.created", { user }, ["user", "purchase"]);
There is a new CpuUsageStrategy
strategy. It selects a node which has the lowest CPU usage.
Due to the node list can be very long, it gets samples and selects the node with the lowest CPU usage from only samples instead of the whole node list.
There are 2 options for the strategy:
sampleCount
: the number of samples. Default: 3
lowCpuUsage
: the low CPU usage percent. The node which has lower CPU usage than this value is selected immediately. Default: 10
Usage:
const broker = new ServiceBroker({
registry: {
strategy: "CpuUsage"
}
});
Usage with custom options
const broker = new ServiceBroker({
registry: {
strategy: "CpuUsage",
strategyOptions: {
sampleCount: 3,
lowCpuUsage: 10
}
}
});
The broker & services starting logic has been changed.
Previous logic: the broker starts transporter connecting. When it's done, it starts all services (calls service started
handlers). It has a disadvantage because other nodes can send requests to these services, while they are still starting and not ready yet.
New logic: the broker starts transporter connecting but it doesn't publish the local service list to remote nodes. When it's done, it starts all services (calls service started
handlers). Once all services start successfully, broker publishes the local service list to remote nodes. Hence other nodes send requests only after all local service started properly.
Please note: you can make dead-locks when two services wait for each other. E.g.:
users
service hasdependencies: [posts]
andposts
service hasdependencies: [users]
. To avoid it remove the concerned service fromdependencies
and usewaitForServices
method out ofstarted
handler instead.
At requests, ctx.meta
is sent back to the caller service. You can use it to send extra meta information back to the caller.
E.g.: send response headers back to API gateway or set resolved logged in user to metadata.
Export & download a file with API gateway:
// Export data
export(ctx) {
const rows = this.adapter.find({});
// Set response headers to download it as a file
ctx.meta.headers = {
"Content-Type": "application/json; charset=utf-8",
"Content-Disposition": 'attachment; filename=\"book.json\"'
}
return rows;
}
Authenticate:
auth(ctx) {
let user = this.getUserByJWT(ctx.params.token);
if (ctx.meta.user) {
ctx.meta.user = user;
return true;
}
throw new Forbidden();
}
If you like better ES6 classes than Moleculer service schema, you can write your services in ES6 classes.
There are two ways to do it:
Native ES6 classes with schema parsing
Define actions
and events
handlers as class methods. Call the parseServiceSchema
method in constructor with schema definition where the handlers pointed to these class methods.
const Service = require("moleculer").Service;
class GreeterService extends Service {
constructor(broker) {
super(broker);
this.parseServiceSchema({
name: "greeter",
version: "v2",
meta: {
scalable: true
},
dependencies: [
"auth",
"users"
],
settings: {
upperCase: true
},
actions: {
hello: this.hello,
welcome: {
cache: {
keys: ["name"]
},
params: {
name: "string"
},
handler: this.welcome
}
},
events: {
"user.created": this.userCreated
},
created: this.serviceCreated,
started: this.serviceStarted,
stopped: this.serviceStopped,
});
}
// Action handler
hello() {
return "Hello Moleculer";
}
// Action handler
welcome(ctx) {
return this.sayWelcome(ctx.params.name);
}
// Private method
sayWelcome(name) {
this.logger.info("Say hello to", name);
return `Welcome, ${this.settings.upperCase ? name.toUpperCase() : name}`;
}
// Event handler
userCreated(user) {
this.broker.call("mail.send", { user });
}
serviceCreated() {
this.logger.info("ES6 Service created.");
}
serviceStarted() {
this.logger.info("ES6 Service started.");
}
serviceStopped() {
this.logger.info("ES6 Service stopped.");
}
}
module.exports = GreeterService;
Use decorators
Thanks for @ColonelBundy, you can use ES7/TS decorators as well: moleculer-decorators
Please note, you need to use Typescript or Babel to compile decorators.
Example service
const moleculer = require('moleculer');
const { Service, Action, Event, Method } = require('moleculer-decorators');
const web = require('moleculer-web');
const broker = new moleculer.ServiceBroker({
logger: console,
logLevel: "debug",
});
@Service({
mixins: [web],
settings: {
port: 3000,
routes: [
...
]
}
})
class ServiceName {
@Action()
Login(ctx) {
...
}
// With options
@Action({
cache: false,
params: {
a: "number",
b: "number"
}
})
Login2(ctx) {
...
}
@Event
'event.name'(payload, sender, eventName) {
...
}
@Method
authorize(ctx, route, req, res) {
...
}
hello() { // Private
...
}
started() { // Reserved for moleculer, fired when started
...
}
created() { // Reserved for moleculer, fired when created
...
}
stopped() { // Reserved for moleculer, fired when stopped
...
}
}
broker.createService(ServiceName);
broker.start();
The broker groups the event listeners by group name. The group name is the name of the service where your event handler is declared. You can change it in the event definition.
module.export = {
name: "payment",
events: {
"order.created": {
// Register handler to "other" group instead of "payment" group.
group: "other",
handler(payload) {
// ...
}
}
}
}
There is a new built-in zero-config TCP transporter. It uses Gossip protocol to disseminate node info, service list and heartbeats. It has an integrated UDP discovery to detect new nodes on the network. It uses multicast discovery messages.
If the UDP is prohibited on your network, you can 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.
Please note you don't need to list all remote nodes. It's enough at least one node which is online. For example, you can create a "serviceless" gossiper node, which does nothing, just shares remote nodes addresses by gossip messages. So all nodes need to know only the gossiper node address to be able to detect all other nodes.
Use TCP transporter with default options
const broker = new ServiceBroker({
transporter: "TCP"
});
Use TCP transporter with static node list
const broker = new ServiceBroker({
transporter: "tcp://172.17.0.1:6000/node-1,172.17.0.2:6000/node-2"
});
or
const broker = new ServiceBroker({
nodeID: "node-1",
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"
]
}
}
});
All TCP transporter options with default values
const broker = new ServiceBroker({
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
}
}
});
There is a new transporter for Kafka. It is a very simple implementation. It transfers Moleculer packets to consumers via pub/sub. There are not implemented offset, replay...etc features.
Please note, it is an experimental transporter. Do not use it in production yet!
To use it, install
kafka-node
withnpm install kafka-node --save
command.
Connect to Zookeeper
const broker = new ServiceBroker({
logger: true,
transporter: "kafka://192.168.51.29:2181"
});
Connect to Zookeeper with custom options
const broker = new ServiceBroker({
logger: true,
transporter: {
type: "kafka",
options: {
host: "192.168.51.29:2181",
// KafkaClient options. More info: https://github.com/SOHU-Co/kafka-node#clientconnectionstring-clientid-zkoptions-noackbatchoptions-ssloptions
client: {
zkOptions: undefined,
noAckBatchOptions: undefined,
sslOptions: undefined
},
// KafkaProducer options. More info: https://github.com/SOHU-Co/kafka-node#producerclient-options-custompartitioner
producer: {},
customPartitioner: undefined,
// ConsumerGroup options. More info: https://github.com/SOHU-Co/kafka-node#consumergroupoptions-topics
consumer: {
},
// Advanced options for `send`. More info: https://github.com/SOHU-Co/kafka-node#sendpayloads-cb
publish: {
partition: 0,
attributes: 0
}
}
}
});
There is a new transporter for NATS Streaming. It is a very simple implementation. It transfers Moleculer packets to consumers via pub/sub. There are not implemented offset, replay...etc features.
Please note, it is an experimental transporter. Do not use it in production yet!
To use it, install
node-nats-streaming
withnpm install node-nats-streaming --save
command.
Connect to NATS Streaming server
// Shorthand to local server
const broker = new ServiceBroker({
logger: true,
transporter: "STAN"
});
// Shorthand
const broker = new ServiceBroker({
logger: true,
transporter: "stan://192.168.0.120:4222"
});
// Shorthand with options
const broker = new ServiceBroker({
logger: true,
transporter: {
type: "STAN",
options: {
url: "stan://127.0.0.1:4222",
clusterID: "my-cluster"
}
}
});
You can define your custom REPL commands in broker options to extend Moleculer REPL commands.
const broker = new ServiceBroker({
logger: true,
replCommands: [
{
command: "hello <name>",
description: "Call the greeter.hello service with name",
alias: "hi",
options: [
{ option: "-u, --uppercase", description: "Uppercase the name" }
],
types: {
string: ["name"],
boolean: ["u", "uppercase"]
},
//parse(command, args) {},
//validate(args) {},
//help(args) {},
allowUnknownOptions: true,
action(broker, args) {
const name = args.options.uppercase ? args.name.toUpperCase() : args.name;
return broker.call("greeter.hello", { name }).then(console.log);
}
}
]
});
broker.repl();
broker.loadServices
file mask is changed from *.service.js
to **/*.service.js
in order to load all services from subfolders too.ServiceNotFoundError
and ServiceNotAvailableError
errors are retryable errors.Strategy.select
method gets only available endpoint list.HEARTBEAT
packet is working properly in Windows too.$node.*
) loading.broker.getAction
deprecated method is removed.PROTOCOL_VERSION
constant is available via broker as ServiceBroker.PROTOCOL_VERSION
or broker.PROTOCOL_VERSION
ctx.broadcast
shortcut method is created to send broadcast events from action handler.broker.started
property is created to indicate broker starting state.dependencies
value in service schema #164
null
,Published by icebob almost 7 years ago
By @tinchoz49 Moleculer Runner has a new built-in clustering function. With it, you can start multiple instances from your broker.
Example to start all services from the services
folder in 4 instances.
$ moleculer-runner --instances 4 services
Please note, the
nodeID
will be suffixed with the worker ID.
By @dani8art you can set that the broker put some ctx.meta
and ctx.params
fields to the metrics events.
You can define it in the action definition:
module.exports = {
name: "test",
actions: {
import: {
cache: true,
metrics: {
// Disable to add `ctx.params` to metrics payload. Default: false
params: false,
// Enable to add `ctx.meta` to metrics payload. Default: true
meta: true
},
handler(ctx) {
// ...
}
}
}
}
If the value is true
, it adds all fields. If Array
, it adds the specified fields. If Function
, it calls with params
or meta
and you need to return with an Object
.
Published by icebob almost 7 years ago
ServiceBroker can resolve the strategy
from a string.
let broker = new ServiceBroker({
registry: {
strategy: "Random"
// strategy: "RoundRobin"
}
});
You can set it via env variables as well, if you are using the Moleculer Runner:
$ REGISTRY_STRATEGY=random
Moleculer runner can load .env
file at starting. There are two new cli options to load env file:
-e, --env
- Load envorinment variables from the '.env' file from the current folder.-E, --envfile <filename>
- Load envorinment variables from the specified file.Example
# Load the default .env file from current directory
$ moleculer-runner --env
# Load the specified .my-env file
$ moleculer-runner --envfile .my-env
Published by icebob almost 7 years ago
In action cache keys you can use meta keys with #
prefix.
broker.createService({
name: "posts",
actions: {
list: {
cache: {
// Cache key: "limit" & "offset" from ctx.params, "user.id" from ctx.meta
keys: ["limit", "offset", "#user.id"],
ttl: 5
},
handler(ctx) {...}
}
}
});
You can override the cacher default TTL setting in action definition.
let broker = new ServiceBroker({
cacher: {
type: "memory",
options: {
ttl: 30 // 30 seconds
}
}
});
broker.createService({
name: "posts",
actions: {
list: {
cache: {
// This cache entries will be expired after 5 seconds instead of 30.
ttl: 5
},
handler(ctx) {...}
}
}
});
You can change the built-in cacher keygen function to your own one.
let broker = new ServiceBroker({
cacher: {
type: "memory",
options: {
keygen(name, params, meta, keys) {
// Generate a cache key
return ...;
}
}
}
});
d.ts
has been improved by @rmccallum81
Published by icebob about 7 years ago
ack
handling has been fixed in AMQP transporter.Published by icebob about 7 years ago
The Service
schema has a new dependencies
property. The serice can wait for other dependening ones when it starts. This way you don't need to call waitForServices
in started
any longer.
module.exports = {
name: "posts",
settings: {
$dependencyTimeout: 30000 // Default: 0 - no timeout
},
dependencies: [
"likes", // shorthand w/o version
{ name: "users", version: 2 }, // with numeric version
{ name: "comments", version: "staging" } // with string version
],
started() {
this.logger.info("Service started after the dependent services available.");
}
....
}
The started
service handler is called once the likes
, users
and comments
services are registered (on the local or remote nodes).
The ServiceBroker
has a new maxQueueSize
option under transit
key. The broker protects the process to avoid crash during a high load with it. The maxQueueSize
default value is 50,000. If pending request queue size reaches it, broker rejects the request with a QueueIsFull
(retryable) error.
let broker = new ServiceBroker({
transporter: "NATS",
transit: {
maxQueueSize: 10 * 1000
}
}
waitForServices
method supports service versions #112
By @imatefx, the waitForServices
broker & service methods support service versions. Use the following formats to define version in a dependency:
module.exports = {
name: "test",
dependencies: { name: "users", version: 2 }
};
Published by icebob about 7 years ago
The Service
schema has a new metadata
property. The Moleculer modules doesn't use it, so you can use it whatever you want.
broker.createService({
name: "posts",
settings: {},
metadata: {
scalable: true,
priority: 5
},
actions: { ... }
});
The
metadata
is transferred between nodes, you can access it via$node.services
. Or inside service withthis.metadata
like settings.
The NATS transporter has been changed. It supports to use the NATS built-in balancer instead of Moleculer balancer. In this case every call
& emit
will be transferred through NATS message broker.
let broker = new ServiceBroker({
transporter: "NATS",
disableBalancer: true
});
broker.sendPing
instead of broker.transit.sendPing
.index.d.ts
updated to v0.112
to 1
in broker.fatal
. Reason: 2
is reserved by Bash for builtin misuse. More info
Published by icebob about 7 years ago
The Moleculer transportation protocol has been changed. It means, the new (>= v0.11) versions can't communicate with the old (<= v0.10.x) ones.
You can find more information about changes in #86 issue.
The whole event handling has been rewritten. By now Moleculer supports event driven architecture. It means that event emits are balanced like action calls are.
For example, you have 2 main services: users
& payments
. Both subscribe to the user.created
event. You start 3 instances from users
service and 2 instances from payments
service. If you emit the event with broker.emit('user.created')
, broker groups & balances the event, so only one users
and one payments
service receive the event.
You can also send broadcast events with the broker.broadcast('user.created
) command. This way every service instance on every node receives the event.
The broker.broadcastLocal('user.created')
command sends events only to the local services.
Every internal event name starts with '$'. These events are not transferred to remote nodes.
Renamed events:
node.connected
-> $node.connected
node.updated
-> $node.updated
node.disconnected
-> $node.disconnected
services.changed
-> $services.changed
. It is called if local or remote service list is changed.circuit-breaker.closed
-> $circuit-breaker.closed
circuit-breaker.opened
-> $circuit-breaker.opened
circuit-breaker.half-opened
-> $circuit-breaker.half-opened
New events:
metrics.circuit-breaker.closed
, metrics.circuit-breaker.opened
, metrics.circuit-breaker.half-opened
The built-in Moleculer load balancer is switchable. You can turn it off, if the transporter has internal balancer (currently AMQP has it).
const broker = new ServiceBroker({
disableBalancer: false
});
Please note! If built-in balancer is disabled, every call & emit (including local ones too) are transferred via transporter.
Some internal broker methods have been removed or renamed.
broker.bus
has been removed.broker.on
has been removed. Use events
in service schema instead.broker.once
has been removed.broker.off
has been removed.broker.getService
has been renamed to broker.getLocalService
broker.hasService
has been removed.broker.hasAction
has been removed.broker.getAction
has been deprecated.broker.isActionAvailable
has been removed.Internal action ($node.list
, $node.services
, $node.actions
, $node.health
) responses are changed. New internal action ($node.events
) to list event subscriptiion is added.
heartbeatInterval
default value is changed from 10
to 5
.heartbeatTimeout
default value is changed from 30
to 15
.circuitBreaker.maxFailures
default value is changed from 5
to 3
.logFormatter
accepts string. The simple
value is a new formatter to show only log level & log messages.New PING & PONG feature has been implemented. Ping remite nodes to measure the network latency and system time differences.
broker.createService({
name: "test",
events: {
"$node.pong"({ nodeID, elapsedTime, timeDiff }) {
this.logger.info(`Pong received from '${nodeID}' - Time: ${elapsedTime}ms, System time difference: ${timeDiff}ms`);
}
}
});
broker.start().then(() => broker.transit.sendPing(/*nodeID*/));
The Validator in ServiceBroker is plugable. So you can change the built-in fastest-validator
to a slower one :) Example Joi validator
If your services depend on other ones, use the waitForService
method to make services wait until dependencies start.
let svc = broker.createService({
name: "seed",
started() {
return this.waitForServices(["posts", "users"]).then(() => {
// Do work...
});
}
});
Signature:
this.waitForServices(serviceNames: String|Array<String>, timeout: Number/*milliseconds*/, interval: Number/*milliseconds*/): Promise
We added some new Moleculer error classes.
MoleculerRetryableError
- Common Retryable error. Caller retries the request if retryCount > 0
.MoleculerServerError
- Common server error (5xx).MoleculerClientError
- Common client/request error (4xx).ServiceNotAvailable
- Raises if the service is registered but isn't available (no live nodes or CB disabled them).ProtocolVersionMismatchError
- Raises if connect a node with an older client (<= v0.10.0)).Published by icebob about 7 years ago
nodeID == null
in local stuffIn all core modules removed the nullable nodeID
. Every places (context, events, $node.* results) the nodeID contains a valid (local or remote) nodeID. On local nodes it equals with broker.nodeID
.
Migration guide
Before:
if (ctx.nodeID == null) { ... }
// ---------
events: {
"users.created"(payload, sender) {
if (sender == null) { ... }
}
}
After:
if (ctx.nodeID == ctx.broker.nodeID) { ... }
// ---------
events: {
"users.created"(payload, sender) {
if (sender == this.broker.nodeID) { ... }
}
}
internalActions
is renamed to internalServices
The internalActions
broker option is renamed to internalServices
.
broker.createNewContext
methodThe createNewContext
broker method is moved to Context
class as a static method.
Migration guide:
Before:
let ctx = broker.createNewContext(action, nodeID, params, opts);
After:
let ctx = Context.create(broker, action, nodeID, params, opts);
// or better
let ctx = broker.ContextFactory.create(broker, action, nodeID, params, opts);
LOCAL_NODE_ID
constantThe recently added LOCAL_NODE_ID
constant is removed. If you want to check the nodeID is local, please use the if (nodeID == broker.nodeID)
syntax.
By @WoLfulus, the service registry balancer strategy is now pluggable.
New syntax:
let Strategies = require("moleculer").Strategies;
let broker = new ServiceBroker({
registry: {
strategy: new Strategies.RoundRobin()
}
});
Custom strategy
You can create you custom strategy.
let BaseStrategy = require("moleculer").Strategies.Base;
class CustomStrategy extends BaseStrategy {
select(list) {
return list[0];
}
};
let broker = new ServiceBroker({
registry: {
strategy: new CustomStrategy()
}
});
The metrics payload contains remoteCall
and callerNodeID
properties. The remoteCall
is true if the request is called from a remote node. In this case the callerNodeID
contains the caller nodeID.
metrics.trace.span.start
:
{
"action": {
"name": "users.get"
},
"id": "123123123",
"level": 1,
"parent": 123,
"remoteCall": true,
"requestID": "abcdef",
"startTime": 123456789,
"nodeID": "node-1",
"callerNodeID": "node-2"
}
metrics.trace.span.start
:
{
"action": {
"name": "users.get"
},
"duration": 45,
"id": "123123123",
"parent": 123,
"requestID": "abcdef",
"startTime": 123456789,
"endTime": 123456795,
"fromCache": false,
"level": 1,
"remoteCall": true,
"nodeID": "node-1",
"callerNodeID": "node-2"
}
The ServiceBroker supports hot reloading services. If you enable it broker will watch file changes. If you modify service file, broker will reload it on-the-fly.
Demo video
Note: Hot reloading is only working with Moleculer Runner or if you load your services with
broker.loadService
orbroker.loadServices
.
Usage
let broker = new ServiceBroker({
logger: console,
hotReload: true
});
broker.loadService("./services/test.service.js");
Usage with Moleculer Runner
Turn it on with --hot
or -H
flags.
$ moleculer-runner --hot ./services/test.service.js
Moleculer protocol documentation is available in docs/PROTOCOL.md file.
By @Nathan-Schwartz, AMQP (for RabbitMQ) transporter added to Moleculer project.
let broker = new ServiceBroker({
transporter: "amqp://guest:guest@rabbitmq-server:5672"
});
let broker = new ServiceBroker({
transporter: new AmqpTransporter({
amqp: {
url: "amqp://guest:guest@localhost:5672",
eventTimeToLive: 5000,
prefetch: 1
}
});
});
Published by icebob about 7 years ago
prefix
options #57
The broker has a new namespace
option to segment your services. For example, you are running development & production services (or more production services) on the same transporter. If you are using different namespace
you can avoid collisions between different environments.
You can reach it in your services as
this.broker.namespace
.
Thereupon the prefix
option in transporters & cachers is removed.
Example
const broker = new ServiceBroker({
logger: console,
namespace: "DEV",
transporter: "NATS",
cacher: "Redis"
});
In this case the transporter & cacher prefix will be MOL-DEV
.
The useVersionPrefix
is renamed to $noVersionPrefix
. The serviceNamePrefix
is renamed to $noServiceNamePrefix
. Both settings logical state is changed.
The cache
setting is renamed to $cache
.
Before
broker.createService({
name: "test",
settings: {
useVersionPrefix: false,
serviceNamePrefix: false,
cache: true
}
});
After
broker.createService({
name: "test",
settings: {
$noVersionPrefix: true,
$noServiceNamePrefix: true,
$cache: true
}
});
Based on #58 if service version is a String
, the version in action names won't be prefixed with v
, expect if it is a Number
.
Example
broker.createService({
name: "test",
version: 3,
actions: {
hello(ctx) {}
}
});
broker.call("v3.test.hello");
broker.createService({
name: "test",
version: "staging",
actions: {
hello(ctx) {}
}
});
broker.call("staging.test.hello");
The module log level is not supported. The logLevel
option can be only String
. It is used if the logger is the console
. In case of external loggers you have to handle log levels.
The whole Moleculer logger is rewritten. It supports better the external loggers. The built-in log message format is also changed.
console
loggerconst broker = createBroker({
logger: console,
logLevel: "info"
});
New console output:
With custom logFormatter
const broker = new ServiceBroker({
logger: console,
logFormatter(level, args, bindings) {
return level.toUpperCase() + " " + bindings.nodeID + ": " + args.join(" ");
}
});
broker.logger.warn("Warn message");
broker.logger.error("Error message");
Output:
WARN dev-pc: Warn message
ERROR dev-pc: Error message
const pino = require("pino")({ level: "info" });
const broker = new ServiceBroker({
logger: bindings => pino.child(bindings)
});
Sample output:
const bunyan = require("bunyan");
const logger = bunyan.createLogger({ name: "moleculer", level: "info" });
const broker = new ServiceBroker({
logger: bindings => logger.child(bindings)
});
Sample output:
const broker = new ServiceBroker({
logger: bindings => new winston.Logger({
transports: [
new (winston.transports.Console)({
timestamp: true,
colorize: true,
prettyPrint: true
})
]
})
});
const WinstonContext = require("winston-context");
const winston = require("winston");
const broker = createBroker({
logger: bindings => new WinstonContext(winston, "", bindings)
});
Please note! Some external loggers have not
trace
&fatal
log methods (e.g.: winston). In this case you have to extend your logger.
const WinstonContext = require("winston-context");
const winston = require("winston");
const { extend } = require("moleculer").Logger;
const broker = createBroker({
logger: bindings => extend(new WinstonContext(winston, "", bindings))
});
The bindings
contains the following properties:
ns
- namespacenodeID
- nodeIDmod
- type of core module: broker
, cacher
, transit
, transporter
svc
- service namever
- service versionPlease avoid to use these property names when you log an Object
. For example: the broker.logger.error({ mod: "peanut" })
overrides the original mod
value!
Available to load & destroy services after the broker started. For example you can hot-reload your services in runtime. The remote nodes will be notified about changes and the broker will emit a services.changed
event locally.
Example
broker.start().then(() => {
setTimeout(() => {
// Create a new service after 5s
broker.createService({
name: "math",
actions: {
add(ctx) {
return Number(ctx.params.a) + Number(ctx.params.b);
},
}
});
}, 5000);
setTimeout(() => {
// Destroy a created service after 10s
let svc = broker.getService("math");
broker.destroyService(svc);
}, 10000);
});
With broker.mcall
method you can call multiple actions (in parallel).
Example with Array
broker.mcall([
{ action: "posts.find", params: {limit: 5, offset: 0}, options: { timeout: 500 } },
{ action: "users.find", params: {limit: 5, sort: "username"} }
]).then(results => {
let posts = results[0];
let users = results[1];
})
Example with Object
broker.mcall({
posts: { action: "posts.find", params: {limit: 5, offset: 0}, options: { timeout: 500 } },
users: { action: "users.find", params: {limit: 5, sort: "username"} }
}).then(results => {
let posts = results.posts;
let users = results.users;
})
Published by icebob about 7 years ago
Calling error! TypeError : Cannot read property 'requestID' of undefined
error when you call a local action from other one directly.Published by icebob about 7 years ago
You can remove an existing action when mixing a service.
broker.createService({
name: "test",
mixins: [OtherService],
actions: {
dangerAction: false
}
});
In the test
service the dangerAction
action won't be registered.
moleculer-runner
You can load services from NPM module in moleculer-runner
.
With CLI arguments
$ moleculer-runner -r npm:moleculer-fake npm:moleculer-twilio
With env
$ SERVICES=posts,users,npm:moleculer-fale,npm:moleculer-twilio
$ moleculer-runner
Published by icebob over 7 years ago
fixed Redis cacher option resolver in ServiceBroker. Now it accepts connection string.
let broker = new ServiceBroker({
cacher: "redis://localhost"
});
The fastest-validator is updated to v0.5.0. It supports multi rules & custom validators.
Published by icebob over 7 years ago
The mixins merge logic is handle better events & lifecycle events. If you have a created
, started
, stopped
lifecycle event or any other service event handler in your services, but your mixin has the same event, Moleculer will call all of them in your service and in mixins.
Published by icebob over 7 years ago
There is a new Moleculer project runner script in the bin
folder.
You can use it if you want to create small repos for services. In this case you needn't to create a ServiceBroker with options. Just create a moleculer.config.js
or moleculer.config.json
file in the root of repo fill it with your options and call the moleculer-runner
within the NPM scripts.
Other solution is that you don't put options to file, instead put it to the environment variables.
There are implemented some new resolvers in broker options to support shorthand configurations. This feature is enabled to load broker options easily from a JSON file or load from environment variables.
Usage for transporters
// Connect to the NATS default (localhost) server
let broker = new ServiceBroker({
transporter: "NATS"
});
// Connect to a NATS server with connection string
let broker = new ServiceBroker({
transporter: "nats://nats-server:4222"
});
// Connect to a NATS server with transporter options
let broker = new ServiceBroker({
transporter: {
type: "NATS",
options: {
prefix: "TEST",
nats: {
host: "nats-server",
user: "admin",
pass: "nats-pass"
}
}
}
});
Usage for cachers
// Use a memory cacher
let broker = new ServiceBroker({
cacher: true
// or
// cacher: "Memory"
});
// Use a Redis cacher with default options
let broker = new ServiceBroker({
cacher: "Redis"
});
// Use a Redis cacher with options
let broker = new ServiceBroker({
cacher: {
type: "Redis",
options: {
ttl: 100
}
}
});
Usage for serializers
// Use the Avro serializer
let broker = new ServiceBroker({
serializers: "Avro"
});
// Use the Protocol Buffer serializer
let broker = new ServiceBroker({
serializers: {
type: "ProtoBuf"
}
});
Implemented better circuit breaker solution. Now every calls (local and remote) are protected with the built-in circuit breaker.
You only need to enable it in broker options.
Usage
let broker = new ServiceBroker({
circuitBreaker: {
enabled: true, // Enable this feature
maxFailures: 5, // Trip breaker on 5 failures
halfOpenTime: 10 * 1000 // 10 sec to switch to `half-open` state
failureOnTimeout: true // Failure if request timed out
failureOnReject: true // Failure if request rejected with error code >= 500
}
});
nodeUnavailable
method is dropped.
Created a built-in Service Registry module. It handles actions of services on nodes, circuit breaker logic...etc. In the future it will be perhaps pluggable.
Via broker options you can change the load balancing strategies of Service Registry.
Example
const { STRATEGY_ROUND_ROBIN, STRATEGY_RANDOM } = require("moleculer");
let broker = new ServiceBroker({
registry: {
strategy: STRATEGY_ROUND_ROBIN, // Load balancing strategy
preferLocal: true // First call local service if available
}
});
Broker has an interactive REPL mode. You can load services, call actions, emit events, subscribe & unsubscribe events from your console. You can list registered nodes & actions.
To use REPL mode please install the moleculer-repl module with
npm install moleculer-repl --save
command.
Start REPL mode
let broker = new ServiceBroker({ logger: console });
// Start REPL
broker.repl();
Commands
Commands:
help [command...] Provides help for a given command.
exit Exits application.
q Exit application
call <actionName> [params] Call an action
dcall <nodeID> <actionName> [params] Call a direct action
emit <eventName> [payload] Emit an event
load <servicePath> Load a service from file
loadFolder <serviceFolder> [fileMask] Load all service from folder
subscribe <eventName> Subscribe to an event
unsubscribe <eventName> Unsubscribe from an event
actions [options] List of actions
nodes List of nodes
info Information from broker
List nodes
mol $ nodes
List services
mol $ services
List actions
mol $ actions
Show common informations
mol $ info
Call an action
mol $ call "test.hello"
Call an action with params
mol $ call "math.add" '{"a": 5, "b": 4}'
Direct call
mol $ dcall server-2 "$node.health"
Emit an event
mol $ emit "user.created"
Subscribe to an event
mol $ subscribe "user.created"
Unsubscribe from an event
mol $ unsubscribe "user.created"
Load a service
mol $ load "./math.service.js"
Load services from folder
mol $ load "./services"
There is available to call an action directly on a specified node. For use, you need to set nodeID
in options of call.
Example
broker.call("user.create", {}, { timeout: 5000, nodeID: "server-12" });
createService
Now there is a second parameter of broker.createService
. With it you can override the schema properties. You can use it to use a built-in service & override some props.
Example
broker.createService(apiGwService, {
settings: {
// Change port setting
port: 8080
},
actions: {
myAction() {
// Add a new action to apiGwService service
}
},
created() {
// Overwrite apiGwService.created handler
}
});
Or you can merge it manually with mergeSchemas
method.
let mergedSchema = broker.mergeSchemas(origSchema, modifications);
broker.createService(mergedSchema);
Similar as mergeable schemas, the service can contain any mixin schemas. The constructor of Service will merge these mixins with the schema of Service. Use it to reuse an other Service in your service. Or you can extend an other Service.
Examples
const ApiGwService = require("moleculer-web");
module.exports = {
name: "api",
mixins: [ApiGwService]
settings: {
// Change port setting
port: 8080
},
actions: {
myAction() {
// Add a new action to apiGwService service
}
}
}
You can protect your app against calling loop with the new maxCallLevel
option. If the ctx.level
value reaches this limit, will be thrown a MaxCallLevelError
error.
let broker = new ServiceBroker({
maxCallLevel: 100
});
There is a new useVersionPrefix
option in settings of Service. If false, Moleculer can't use the version number of service as prefix for action names. The name of service will be users.find
instead of v2.users.find
. The default is true
.
node.reconnected
and node.broken
events (breaking)We merged the node.connected
and node.reconnected
events. The payload is changed:
{
node: {...},
reconnected: false // it indicates the node is connected or reconnected
}
We merged also the node.disconnected
and node.broken
events. The payload is changed:
{
node: {...},
unexpected: true // True: broken, not coming heart-beat, False: received "DISCONNECT" packet
}
Moleculer doesn't contain dependencies for NATS, Redis, MQTT, MsgPack, Avro and Protobuf. So it need install manually in your project.
If you want to create a Moleculer project which communicates via NATS and your Redis cacher, you have to install npm install moleculer nats redis --save
The code of ServiceNotFoundError
is changed from 501
to 404
. More info
Memory cacher is using nanomatch instead of micromatch. The nanomatch
is ~10x faster.
metricsSendInterval
option #24
The metricsSendInterval
option is removed from broker options. If you want to access statistics & health info, call the $node.health
and $node.stats
actions.
The metrics & statistics features separated. You can use just metrics or just statistics.
Metrics events contains two nodeID properties.
nodeID
: the "caller" nodeIDtargetNodeID
: in case of remote call this is the remote nodeIDIf an action responses an error on a remote node, the transporter will send back the error to the caller with the stack traces.
// It will print the original error stack trace.
broker.call("account.deposit").catch(err => console.log(err.stack));
The CustomError
class renamed to MoleculerError
. It got a type
new property. You can store here a custom error type. E.g if you have a Validation error sometimes you don't enough the name & code. With type
the client can handle the cause of error programmatically.
Example
const ERR_MISSING_ID = "ERR_MISSING_ID";
const ERR_ENTITY_NOT_FOUND = "ERR_ENTITY_NOT_FOUND";
broker.createService({
actions: {
get(ctx) {
if (ctx.params.id) {
const entity = this.searchEntity(ctx.params.id);
if (entity)
return entity;
else
return Promise.reject(new ValidationError("Not found entity!", ERR_ENTITY_NOT_FOUND));
} else
return Promise.reject(new ValidationError("Please set the ID field!", ERR_MISSING_ID));
}
}
});
appendServiceName
settings to serviceNamePrefix
in Service schemaThe ServiceBroker
has a new fatal
method. If you call it, broker will log the message with fatal
level and exit the process with code 2
.
broker.fatal(message, err, needExit = true)
If you are running your app in containers and it has restart policy, you can use it to restart your app.
Usage
try {
// Do something dangerous
} catch(err) {
broker.fatal("Dangerous thing is happened!", err, true);
}
$node.actions
and $node.services
INFO
& DISCOVER
changed the actions
property to services
and now it contains all services with actions of nodebroker.registerService
to registerLocalService
and registerRemoteService
broker.unregisterServicesByNode
. It will be called when a node disconnected