/*
 * map.js: Stream responsible for mapping events before emitting them.
 *
 * (C) 2012, Nodejitsu Inc.
 *
 */

var util = require('util'),
    ReadWriteStream = require('../common/read-write-stream');

//
// ### function Map (mapFn)
// #### @mapFn {function} Map function to call on each event.
// Constructor function of the Map stream responsible for mapping
// events before emitting them.
//
var Map = module.exports = function (mapFn) {
  if (!mapFn || typeof mapFn !== 'function') {
    throw new Error('map function is required.');
  }

  ReadWriteStream.call(this);
  this.mapFn = mapFn;
  this.async = mapFn.length === 2;
};

//
// Inherit from ReadWriteStream
//
util.inherits(Map, ReadWriteStream);

//
// ### function write (data)
// #### @data {Object} JSON to filter
// Emits data after it is mutating with `this.mapFn`.
//
Map.prototype.write = function (data) {
  var self = this;
  if (!this.async) {
    this.emit('data', this.mapFn(data));
    return;
  }

  this.mapFn(data, function (err, data) {
    return err
      ? self.emit('error', err)
      : self.emit('data', data);
  });
};

@@ -290,16 +290,93 @@
  event.%0A
+// #### @options %7BObject%7D options to pass into reactor to customize behavior%0A
 // Const
@@ -510,16 +510,25 @@
 n (mapFn
+, options
 ) %7B%0A  if
@@ -625,16 +625,44 @@
 );%0A  %7D%0A%0A
+  options = options %7C%7C %7B%7D;%0A%0A
   ReadWr
@@ -740,16 +740,323 @@
  === 2;%0A
+  //%0A  // So by default the behavior is to block the stream when running through%0A  // the async function. This option is here to allow you to do an async%0A  // operation that does not block the stream%0A  // TODO: Think of a better API for this behavior%0A  //%0A  this.passThrough = options.passThrough %7C%7C false;%0A
 %7D;%0A%0A//%0A/
@@ -1313,16 +1313,17 @@
 = this;%0A
+%0A
   if (!t
@@ -1395,16 +1395,52 @@
 %0A  %7D%0A%0A  
+if (!this.passThrough) %7B%0A    return 
 this.map
@@ -1471,24 +1471,26 @@
 data) %7B%0A    
+  
 return err%0A 
@@ -1488,16 +1488,18 @@
 urn err%0A
+  
       ? 
@@ -1528,16 +1528,18 @@
 )%0A      
+  
 : self.e
@@ -1559,15 +1559,133 @@
 ata);%0A  
-%7D
+  %7D);%0A  %7D%0A%0A  this.mapFn(data, function (err) %7B%0A    if (err) %7B self.emit('error', err) %7D%0A  %7D);%0A%0A  this.emit('data', data
 );%0A%7D;%0A

allow map reactor to accept a `passThrough` option which allows you to accomplish async operations triggered by the data stream without blocking it