dump fluentd forward protocol.

fluent-dump.coffee

# $ coffee fluent-dump.coffee [bind] [port]

net = require 'net'
dns = require 'dns'
util = require 'util'
dgram = require 'dgram'

msgpack = require 'msgpack'
msgpack.raw = require 'msgpack-raw'

print_header = (socket, tag, type) ->
   console.log '---------------------- ' + socket.remoteAddress + ' / ' + type
   console.log tag.toString('utf8') if tag?

print_entry = (entry) ->
   entry[0] = new Date entry[0] * 1000
   console.log util.inspect entry, depth: null, colors: true

tcp_server = (family, address, port) ->
   server = net.createServer()

   server.listen port, address, ->
      console.log 'listen TCP on ' + address + ':' + port

   server.on 'error', (e) ->
      console.log 'error'

   server.on 'connection', (socket) ->
      stream = new msgpack.raw.Stream socket
      stream.on 'msg', (msg) ->
         switch
            when msg.length == 3
               print_header socket, msg[0], 'Messaage'
               print_entry [msg[1], msgpack.unpack msgpack.raw.pack msg[2]]
            when typeof msg[0] == 'number'
               print_header socket, null, 'Entry'
               print_entry [msg[0], msgpack.unpack msgpack.raw.pack msg[1]]
            when msg[1] instanceof Buffer
               print_header socket, msg[0], 'PackedForward'
               buf = msg[1]
               while buf.length > 0
                  print_entry msgpack.unpack buf
                  buf = buf.slice buf.length - msgpack.unpack.bytes_remaining, buf.length
            else
               print_header socket, msg[0], 'Forward'
               entries = msgpack.unpack msgpack.raw.pack msg[1]
               for entry in entries
                  print_entry entry

udp_server = (family, address, port) ->
   udp = dgram.createSocket 'udp' + family

   udp.on 'listening', ->
      console.log 'listen UDP on ' + udp.address().address + ':' + udp.address().port

   udp.on 'message', (msg, rinfo) ->
      udp.send msg, 0, msg.length, rinfo.port, rinfo.address

   udp.bind port, address

bind = process.argv[2]
port = parseInt process.argv[3]

for family in [4, 6]
   dns.lookup bind, family, (err, address) ->
      tcp_server family, address, port
      udp_server family, address, port