SSE implementation in CoffeeScript

sse.coffee

cluster = require('cluster')
http = require('http')
zmq = require('zmq')

numCPUs = require('os').cpus().length
NODES = ['tcp://localhost:5555']

if cluster.isMaster
  publisher = zmq.socket('pub')

  workers = []
  currentWorker = 0
  console.log '( worker count ) ' + numCPUs
  workers.push cluster.fork() for num in [1..numCPUs]

  publisher.bind 'tcp://*:5555', (err) ->
    throw err if err
    console.log "bound"

  server = http.createServer (req, res) ->
    if req.method == 'POST'
      res.writeHead 204
      message = ''
      req.on 'data', (data) ->
        message += data
      req.on 'end', ->
        publisher.send 'channel: ' + req.url + '\nmessage: ' + message
      res.end()
    else
      res.writeHead 200, 'Content-Type':'text/event-stream', 'Cache-Control':'no-cache', 'Connection':'keep-alive'
      workers[currentWorker].send req.url, req.socket
      currentWorker = (currentWorker + 1) % workers.length

  server.listen 8888
else
  connections = {}

  setInterval ->
    totalConnections = 0
    for channel, sockets of connections
      aliveSockets = []
      for socket in sockets
        if not socket.destroyed
          socket.write ':ping\n'
          aliveSockets.push socket
          totalConnections += 1
      connections[channel] = aliveSockets
      #connections[channel] = null if aliveSockets.length == 0
      console.log '( worker ' + process.pid + ' connections ) ' + totalConnections
  , 15000

  process.on 'message', (channel, socket) ->
    connections[channel] ||= []
    connections[channel].push socket

  NODES.forEach (port) ->
    subscriber = zmq.socket('sub')
    subscriber.connect port
    subscriber.subscribe 'channel'
    console.log "subscribed"
    subscriber.on 'message', (data) ->
      str = data.toString()
      [channel, message] = str.split /\n/
      channel = channel.replace 'channel: ', ''
      message = message.replace 'message: ', ''
      console.log "got message"
      if connections[channel]
        console.log "sending message"
        for connection in connections[channel]
          connection.write 'data: ' + message + '\n\n' if not connection.destroyed