def on_message(msg)
      # TODO format error
      tag = msg[0].to_s
      entries = msg[1]

      if entries.class == String
        # PackedForward
        es = MessagePackEventStream.new(entries)
        router.emit_stream(tag, es)

      elsif entries.class == Array
        # Forward
        es = MultiEventStream.new
        entries.each {|e|
          record = e[1]
          next if record.nil?
          time = e[0]
          time = (now ||= Engine.now) if time.to_i == 0
          es.add(time, record)
        }
        router.emit_stream(tag, es)

      else
        # Message
        record = msg[2]
        return if record.nil?

        time = msg[1]
        time = Engine.now if time.to_i == 0
        router.emit(tag, time, record)
      end
    end
def on_message(msg)
      # TODO format error
      tag = msg[0].to_s
      entries = msg[1]

      if entries.class == String
        # PackedForward
        es = MessagePackEventStream.new(entries)
        router.emit_stream(tag, es)

      elsif entries.class == Array
        # Forward
        es = MultiEventStream.new
        entries.each {|e|
          record = e[1]
          next if record.nil?
          time = e[0]
          time = (now ||= Engine.now) if time.to_i == 0
          es.add(time, record)
        }
        router.emit_stream(tag, es)

      else
        # Message
        record = msg[2]
        return if record.nil?

        time = msg[1]
        time = Engine.now if time.to_i == 0
        router.emit(tag, time, record)
      end
    end
message Entry {
   1: long time
   2: object record
 }

 message Forward {
   1: string tag
   2: list<Entry> entries
 }

 message PackedForward {
   1: string tag
   2: raw entries  # msgpack stream of Entry
 }

 message Message {
   1: string tag
   2: long? time
   3: object record
 }