How to use zeromq - 10 common examples
To help you get started, we’ve selected a few zeromq examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
rehanift / engine.js / robust.js View on Github 
return task;
};
self.addTask = function(task){
self.taskPool.add(task);
};
self.getTask = function(task_id){
return self.taskPool.get(task_id);
};
var context = require("zeromq");
self.id = robust.util.makeUUID({prefix:"client"});
self.results_socket = context.createSocket("pull");
self.code_socket = context.createSocket("push");
self.results_socket.on("message", function(data){
if (data == robust.constants.HANDSHAKE) {
// do nothing
return true;
} else if (data == robust.constants.READY) {
console.log("Ready handshake");
self.emit("ready");
return true;
} else {
// TODO: need safe JSON parsing
data = JSON.parse(data);
robust.util.info("Receiving results for task " + data['task_id']);
console.log(data);
}};
self.addTask = function(task){
self.taskPool.add(task);
};
self.getTask = function(task_id){
return self.taskPool.get(task_id);
};
var context = require("zeromq");
self.id = robust.util.makeUUID({prefix:"client"});
self.results_socket = context.createSocket("pull");
self.code_socket = context.createSocket("push");
self.results_socket.on("message", function(data){
if (data == robust.constants.HANDSHAKE) {
// do nothing
return true;
} else if (data == robust.constants.READY) {
console.log("Ready handshake");
self.emit("ready");
return true;
} else {
// TODO: need safe JSON parsing
data = JSON.parse(data);
robust.util.info("Receiving results for task " + data['task_id']);
console.log(data);
}connect: (node, settings) => {
if (zmqNodesAmountConnected < maxAmountZmqConnections) {
zmqNodesAmountConnected++;
zmqSockets[node.host] = zmq.socket('sub');
zmqSockets[node.host].connect(`tcp://${node.host}:${node.port}`);
console.log(
Time.Stamp() +
`Connected to ${node.host} | Current ZMQ node connections: ${zmqNodesAmountConnected}`
);
zmqSockets[node.host].subscribe('tx'); // New transactions
zmqSockets[node.host].subscribe('sn'); // New confirmed transactions
zmqSockets[node.host].subscribe('lmhs'); // New milestones
zmqSockets[node.host].on('close', close => {
zmqNodesAmountConnected--;
console.log(
Time.Stamp() +
`Connection close: ${
node.host#!/usr/bin/env node --harmony
'use strict';
const fs = require('fs');
const zmq = require('zeromq');
const filename = process.argv[2];
// Create the publisher endpoint
const publisher = zmq.socket('pub');
fs.watch(filename, () => {
// Send a message to any and all subscribers
publisher.send(JSON.stringify({
type: 'changed',
file: filename,
timestamp: Date.now()
}));
});
// Listen on TCP port 60400
publisher.bind('tcp://*:60400', err => {
if (err) {
throw err;
}
console.log('Listening for zmq subscribers...');
});if (typeof(options.crankshaft) != "undefined") {
crankshaft_endpoint = options.crankshaft;
}
}
self.cylinder_block = context.createSocket("push");
self.cylinder_block.bind(cylinder_block_endpoint, function(err){
if (err) {
throw err;
}
self.emit("cylinder block ready");
self.checkoff("cylinder block", self.handleCheckoff);
});
self.crankshaft = context.createSocket("pull");
self.crankshaft.bind(crankshaft_endpoint, function(err){
if (err) {
throw err;
}
self.emit("crankshaft ready");
self.checkoff("crankshaft", self.handleCheckoff);
});
self.crankshaft.on("message", function(data){
var parsed_data = JSON.parse(data);
var task = self.getRunningTask(parsed_data.running_task_id);
//if (task && task.getCallback()) {
// task.getCallback().call(null, data);
//}#!/usr/bin/env node
'use strict';
var ZMQ = require('zeromq');
var BROKER_REP = '127.0.0.1:5554';
var BROKER_PUB = '127.0.0.1:5555';
function log() {}
//var log = console.log;
//
// broker
//
var brokerListener = ZMQ.createSocket('sub');
var brokerPublisher = ZMQ.createSocket('pub');
var n = 0;
brokerListener.bind('tcp://' + BROKER_REP, function(err) {
if (err) throw err;
brokerListener.subscribe('');
brokerListener.on('message', function(data) {
log('BIN: ' + data.toString('utf8'));
var r = brokerPublisher.send(data);
if (r) {
n++;
console.error('PUB FAILED: ', r, n);
}
});
console.error('Broker is listening to ' + BROKER_REP + '...');
});
brokerPublisher.bind('tcp://' + BROKER_PUB, function(err) {
if (err) throw err;#!/usr/bin/env node
'use strict';
var ZMQ = require('zeromq');
var BROKER_REP = '127.0.0.1:5554';
var BROKER_PUB = '127.0.0.1:5555';
function log() {}
var log = console.log;
//
// broker
//
var brokerListener = ZMQ.createSocket('rep');
var brokerPublisher = ZMQ.createSocket('pub');
var n = 0;
brokerListener.bind('tcp://' + BROKER_REP, function(err) {
if (err) throw err;
brokerListener.on('message', function(data) {
if (!(n % 10000)) console.error('BIN', n, Date.now());
++n;
log('BIN: ' + data.toString('utf8'));
brokerPublisher.send(data);
brokerListener.send('');
});
console.error('Broker is listening to ' + BROKER_REP + '...');
});
brokerPublisher.bind('tcp://' + BROKER_PUB, function(err) {
if (err) throw err;
brokerPublisher.on('message', function(data) {import zmq from 'zeromq'
zmq.Context.setMaxThreads(8)
let getMaxThreads = zmq.Context.getMaxThreads()
let getMaxSockets = zmq.Context.getMaxSockets()
console.log('getMaxThreads', getMaxThreads)
console.log('getMaxSockets', getMaxSockets)
let dealer1 = zmq.socket('dealer')
let dealer2 = zmq.socket('dealer')
let router1 = zmq.socket('router')
let router2 = zmq.socket('router')
// ** BUG scenario
// router1.monitor(10, 0)
// router2.monitor(10, 0)
// dealer1.monitor(10, 0)
// dealer2.monitor(10, 0)import zmq from 'zeromq'
zmq.Context.setMaxThreads(8)
let getMaxThreads = zmq.Context.getMaxThreads()
let getMaxSockets = zmq.Context.getMaxSockets()
console.log('getMaxThreads', getMaxThreads)
console.log('getMaxSockets', getMaxSockets)
let dealer1 = zmq.socket('dealer')
let dealer2 = zmq.socket('dealer')
let router1 = zmq.socket('router')
let router2 = zmq.socket('router')
// ** BUG scenario
// router1.monitor(10, 0)
// router2.monitor(10, 0)new Worker(src, {eval: true}).on("exit", code => {
if (code === 0) {
resolve()
} else {
reject(new Error(`Worker stopped with exit code ${code}`))
}
})
})
})
await Promise.all(workers)
console.log("all workers stopped")
}
/* Queue only 1 incoming message. */
input = new zmq.Pull({receiveHighWaterMark: 1})
output = new zmq.Push()
signal = new zmq.Subscriber()
shift = 13
maxDelay = 2000 /* Average of 1s. */
constructor() {
this.input.connect("inproc://input")
this.output.connect("inproc://output")
this.signal.connect("inproc://signal")
this.signal.subscribe()
const listen = async () => {
for await (const [sig] of this.signal) {
if (sig.toString() === "stop") this.stop()zeromq
Next-generation ZeroMQ bindings for Node.js
