ZMQ Req <--> Router <--> Dealer -- this is the basics of a Paranoid Pirate Pattern in node.js

req_router_dealer.coffee

###
skeleton of a Paranoid Pirate Pattern with a worker that is randomly slow
(needs heartbeating, queuing and retry to be fully fleshed out)
###

zmq = require("zmq")
delay = process.argv[2] or 1000
log = console.log

debugger

handleError = ->
  log 'error ' + argsToString(Array.apply(null, arguments))
  throw 'zmq error'

argsToString = (a) ->
  a.join(', ')

buildSocket = (desc, socketType)->
  log "creating socket: #{argsToString(Array.apply(null, arguments))}"
  socket = zmq.socket(socketType)
  socket.identity = "#{desc}_#{process.pid}"
  socket

qAddr = "tcp://127.0.0.1:8000"

log 'building sockets...'
client = buildSocket('client', 'req')
q = buildSocket('q', 'router')
worker = buildSocket('worker', 'dealer')

q.bind qAddr, (err) ->
  throw err  if err
  log 'q is ready...'
  q.on "message", ->
    log "message @ q: #{argsToString(Array.apply(null, arguments))}"
    if arguments[2]?.toString() == 'do'
      log 'q sending work to worker'
      q.send [ worker.identity, arguments[3] ]
    else if arguments[0].toString() == worker.identity # message from worker, relay to client
      log 'q: sending completed work to client'
      q.send [ client.identity, '',  arguments[1]]
      #else if  arguments[2]?.toString() == 'ack'
      #  log 'q: client got message'
    else
      throw 'no idea what to do!'

  client.connect qAddr
  worker.connect qAddr

q.on 'error', handleError

client.on "message",   ->
  log "message @ client:  #{argsToString(Array.apply(null, arguments))}"
  #client.send ['ack']

worker.on "message",   ->
  log "message @ worker:  #{argsToString(Array.apply(null, arguments))}"
  nv =  +(arguments[0])*2
  if Math.random() > 0.50
    log "worker sending delayed response"
    setTimeout( (-> worker.send(nv) ), 50)
  else
    worker.send(nv)

setTimeout (->
  setInterval (->
    log "\nsending work from client"
    v = Math.floor(Math.random()*100)
    client.send(['do', v])
  ), delay
), 500