1  var util = require('util');
 2  var PassThrough = require('readable-stream/passthrough');
 3  
 4  module.exports = {
 5    Readable: Readable,
 6    Writable: Writable
 7  };
 8  
 9  util.inherits(Readable, PassThrough);
10  util.inherits(Writable, PassThrough);
11  
12  // Patch the given method of instance so that the callback
13  // is executed once, before the actual method is called the
14  // first time.
15  function beforeFirstCall(instance, method, callback) {
16    instance[method] = function() {
17      delete instance[method];
18      callback.apply(this, arguments);
19      return this[method].apply(this, arguments);
20    };
21  }
22  
23  function Readable(fn, options) {
24    if (!(this instanceof Readable))
25      return new Readable(fn, options);
26  
27    PassThrough.call(this, options);
28  
29    beforeFirstCall(this, '_read', function() {
30      var source = fn.call(this, options);
31      var emit = this.emit.bind(this, 'error');
32      source.on('error', emit);
33      source.pipe(this);
34    });
35  
36    this.emit('readable');
37  }
38  
39  function Writable(fn, options) {
40    if (!(this instanceof Writable))
41      return new Writable(fn, options);
42  
43    PassThrough.call(this, options);
44  
45    beforeFirstCall(this, '_write', function() {
46      var destination = fn.call(this, options);
47      var emit = this.emit.bind(this, 'error');
48      destination.on('error', emit);
49      this.pipe(destination);
50    });
51  
52    this.emit('writable');
53  }
54