         for own key, value of opts
           options[key] = value
 
+    @inPorts.tag.on "data", (@tag) =>
+
     @inPorts.in.on "connect", =>
       count++
       @cache = []
       @groups = []
+
+      if @tag?
+        @cache[@tag] ?= []
+        @groups.push @tag
 
     @inPorts.in.on "begingroup", (group) =>
       here = @locate()

    @inPorts.in.on "disconnect", =>
      @outPorts.out.disconnect() if @outPorts.out.isAttached()

      log.push @cache
      count--
      flush() if count is 0

  entire stream is complete."

  constructor: ->
    @tag = null

    @inPorts =
      in: new noflo.Port
      options: new noflo.Port
      tag: new noflo.Port
    @outPorts =
      out: new noflo.Port

