git ssb

1
2
var sources  = require('./sources')
3
var sinks    = require('./sinks')
4
var throughs = require('./throughs')
5
 
6
for(var k in sources)
7
  exports[k] = Source(sources[k])
8
9
for(var k in throughs)
10
  exports[k] = Through(throughs[k])
11
12
for(var k in sinks)
13
  exports[k] = Sink(sinks[k])
14
15
exports.Duplex  = 
16
exports.Through = exports.pipeable       = Through
17
exports.Source  = exports.pipeableSource = Source
18
exports.Sink    = exports.pipeableSink   = Sink
19
20
function addPipe(read) {
21
  if('function' !== typeof read)
22
    return read
23
24
  read.pipe = read.pipe || function (reader) {
25
    if('function' != typeof reader)
26
      throw new Error('must pipe to reader')
27
    return addPipe(reader(read))
28
  }
29
30
  return read
31
}
32
33
function Source (createRead) {
34
  return function () {
35
    var args = [].slice.call(arguments)
36
    return addPipe(createRead.apply(null, args))
37
  }
38
}
39
40
function Through (createRead) {
41
  return function () {
42
    var args = [].slice.call(arguments)
43
    var piped = []
44
    function reader (read) {
45
      args.unshift(read)
46
      read = createRead.apply(null, args)
47
      while(piped.length)
48
        read = piped.shift()(read)
49
      return read
50
      //pipeing to from this reader should compose...
51
    }
52
    reader.pipe = function (read) {
53
      piped.push(read)
54
      return reader
55
    }
56
    return reader
57
  }
58
}
59
60
function Sink(createReader) {
61
  return function () {
62
    var args = [].slice.call(arguments)
63
    return function (read) {
64
      args.unshift(read)
65
      return createReader.apply(null, args)
66
    }
67
  }
68
}
69
70
/*
71
var destack = function (n) {
72
  var i = 0; n = n || 10, waiting = [], queued = false, ended = false
73
  return function (readable) {
74
    return function (reader) {
75
      return reader(function (end, cb) {
76
        ended = ended || end
77
        if(i ++ < n) {
78
          return readable(end, cb)
79
        } else {
80
          process.nextTick(function () {
81
             i = 0
82
             readable(end, cb)
83
          })
84
        }
85
      })
86
    }
87
  }
88
}
89
*/
90