init functionality in stream nodes

lib/streams/core.js

@@ -19,6 +19,7 @@ function Node(identifier, listenType, emitType) {
   this.emitType = emits;
   this.emitDeferred = Promise.defer();
   this.results = null;
+  this.initted = false;
 }
 
 Node.prototype = {
@@ -44,6 +45,10 @@ Node.prototype = {
      * @param   flush     whether to wait until emitReady
      * @returns a promise that resolves when processing is complete
      */
+    if (!this.initted) {
+      this.init();
+      this.initted = true;
+    }
     this.ingest(message);
     if (this.emitReady() || flush) {
       return this.flush();
@@ -83,6 +88,12 @@ Node.prototype = {
      */
     this.results = null;
     this.emitDeferred = Promise.defer();
+  },
+
+  init: function _Node_init() {
+    /*
+     * Optional initialization that runs when consume is called for the first time.
+     */
   }
 }