lazy.js

   1 var EventEmitter = require('events').EventEmitter;
   2
   3 Lazy.prototype = new EventEmitter;
   4 module.exports = Lazy;
   5 function Lazy (em, opts) {
   6     if (!(this instanceof Lazy)) return new Lazy(em, opts);
   7     var self = this;
   8     if (em) {
   9         if (!em._events) em._events = {};
  10         self._events = em._events;
  11     }
  12
  13     self.once = function (name, f) {
  14         self.on(name, function g () {
  15             self.removeListener(name, g);
  16             f.apply(this, arguments);
  17         });
  18     }
  19
  20     if (!opts) opts = {};
  21     var dataName = opts.data || 'data';
  22     var pipeName = opts.pipe || 'pipe';
  23     var endName = opts.pipe || 'end';
  24
  25     if (pipeName != endName) {
  26         var piped = false;
  27         self.once(pipeName, function () { piped = true });
  28         self.once(endName, function () {
  29             if (!piped) self.emit(pipeName);
  30         });
  31     }
  32
  33     self.push = function (x) {
  34         self.emit(dataName, x);
  35     }
  36
  37     self.end = function () {
  38         self.emit(endName);
  39     }
  40
  41     function newLazy (g, h) {
  42         if (!g) g = function () { return true };
  43         if (!h) h = function (x) { return x };
  44         var lazy = new Lazy(null, opts);
  45         self.on(dataName, function (x) {
  46             if (g.call(lazy, x)) lazy.emit(dataName, h(x));
  47         });
  48         self.once(pipeName, function () {
  49             lazy.emit(pipeName)
  50         });
  51         return lazy;
  52     }
  53
  54     self.filter = function (f) {
  55         return newLazy(function (x) {
  56             return f(x);
  57         });
  58     }
  59
  60     self.forEach = function (f) {
  61         return newLazy(function (x) {
  62             f(x);
  63             return true;
  64         });
  65     }
  66
  67     self.map = function (f) {
  68         return newLazy(
  69             function () { return true },
  70             function (x) { return f(x) }
  71         );
  72     }
  73
  74     self.head = function (f) {
  75         var lazy = newLazy();
  76         lazy.on(dataName, function g (x) {
  77             f(x)
  78             lazy.removeListener(dataName, g)
  79         })
  80     }
  81
  82     self.tail = function () {
  83         var skip = true;
  84         return newLazy(function () {
  85             if (skip) {
  86                 skip = false;
  87                 return false;
  88             }
  89             return true;
  90         });
  91     }
  92
  93     self.skip = function (n) {
  94         return newLazy(function () {
  95             if (n > 0) {
  96                 n--;
  97                 return false;
  98             }
  99             return true;
 100         });
 101     }
 102
 103     self.take = function (n) {
 104         return newLazy(function () {
 105             if (n == 0) self.emit(pipeName);
 106             return n-- > 0;
 107         });
 108     }
 109
 110     self.takeWhile = function (f) {
 111         var cond = true;
 112         return newLazy(function (x) {
 113             if (cond && f(x)) return true;
 114             cond = false;
 115             self.emit(pipeName);
 116             return false;
 117         });
 118     }
 119
 120     self.foldr = function (op, i, f) {
 121         var acc = i;
 122         var lazy = newLazy();
 123         lazy.on(dataName, function g (x) {
 124             acc = op(x, acc);
 125         });
 126         lazy.once(pipeName, function () {
 127             f(acc);
 128         });
 129     }
 130
 131     self.sum = function (f) {
 132         return self.foldr(function (x, acc) { return x + acc }, 0, f);
 133     }
 134
 135     self.product = function (f) {
 136         return self.foldr(function (x, acc) { return x*acc }, 1, f);
 137     }
 138
 139     self.join = function (f) {
 140         var data = []
 141         var lazy = newLazy(function (x) {
 142             data.push(x);
 143             return true;
 144         });
 145         lazy.once(pipeName, function () { f(data) });
 146         return self;
 147     }
 148
 149     self.bucket = function (init, f) {
 150         var lazy = new Lazy(null, opts);
 151         var yield = function (x) {
 152             lazy.emit(dataName, x);
 153         };
 154
 155         var acc = init;
 156
 157         self.on(dataName, function (x) {
 158             acc = f.call(yield, acc, x);
 159         });
 160
 161         self.once(pipeName, function () {
 162             lazy.emit(pipeName)
 163         });
 164
 165         // flush on end event
 166         self.once(endName, function () {
 167           var finalBuffer = mergeBuffers(acc);
 168           if(finalBuffer) yield(finalBuffer);
 169         });
 170
 171         return lazy;
 172     }
 173
 174     self.spilt = function(c){
 175         return self.bucket([], function (chunkArray, chunk) {
 176             var newline = c.charCodeAt(0), lastNewLineIndex = 0;
 177             if (typeof chunk === 'string') chunk = new Buffer(chunk);
 178
 179             for (var i = 0; i < chunk.length; i++) {
 180               if (chunk[i] === newline) {
 181                 // If we have content from the current chunk to append to our buffers, do it.
 182                 if(i>0) chunkArray.push(chunk.slice(lastNewLineIndex, i));
 183
 184                 // Wrap all our buffers and emit it.
 185                 this(mergeBuffers(chunkArray));
 186                 lastNewLineIndex = i + 1;
 187               }
 188             }
 189
 190             if(lastNewLineIndex>0) {
 191               // New line found in the chunk, push the remaining part of the buffer.
 192               if(lastNewLineIndex < chunk.length) chunkArray.push(chunk.slice(lastNewLineIndex));
 193             } else {
 194               // No new line found, push the whole buffer.
 195               if(chunk.length) chunkArray.push(chunk);
 196             }
 197             return chunkArray;
 198         });
 199     }
 200
 201     // Streams that use this should emit strings or buffers only
 202     self.__defineGetter__('lines', function () {
 203         return self.spilt("\n");
 204     });
 205 }
 206
 207 Lazy.range = function () {
 208     var args = arguments;
 209     var step = 1;
 210     var infinite = false;
 211
 212     if (args.length == 1 && typeof args[0] == 'number') {
 213         var i = 0, j = args[0];
 214     }
 215     else if (args.length == 1 && typeof args[0] == 'string') { // 'start[,next]..[end]'
 216         var arg = args[0];
 217         var startOpen = false, endClosed = false;
 218         if (arg[0] == '(' || arg[0] == '[') {
 219             if (arg[0] == '(') startOpen = true;
 220             arg = arg.slice(1);
 221         }
 222         if (arg.slice(-1) == ']') endClosed = true;
 223
 224         var parts = arg.split('..');
 225         if (parts.length != 2)
 226             throw new Error("single argument range takes 'start..' or 'start..end' or 'start,next..end'");
 227
 228         if (parts[1] == '') { // 'start..'
 229             var i = parts[0];
 230             infinite = true;
 231         }
 232         else { // 'start[,next]..end'
 233             var progression = parts[0].split(',');
 234             if (progression.length == 1) { // start..end
 235                 var i = parts[0], j = parts[1];
 236             }
 237             else { // 'start,next..end'
 238                 var i = progression[0], j = parts[1];
 239                 step = Math.abs(progression[1]-i);
 240             }
 241         }
 242
 243         i = parseInt(i, 10);
 244         j = parseInt(j, 10);
 245
 246         if (startOpen) {
 247             if (infinite || i < j) i++;
 248             else i--;
 249         }
 250
 251         if (endClosed) {
 252             if (i < j) j++;
 253             else j--;
 254         }
 255     }
 256     else if (args.length == 2 || args.length == 3) { // start, end[, step]
 257         var i = args[0], j = args[1];
 258         if (args.length == 3) {
 259             var step = args[2];
 260         }
 261     }
 262     else {
 263         throw new Error("range takes 1, 2 or 3 arguments");
 264     }
 265     var lazy = new Lazy;
 266     var stopInfinite = false;
 267     lazy.on('pipe', function () {
 268         stopInfinite = true;
 269     });
 270     if (infinite) {
 271         process.nextTick(function g () {
 272             if (stopInfinite) return;
 273             lazy.emit('data', i++);
 274             process.nextTick(g);
 275         });
 276     }
 277     else {
 278         process.nextTick(function () {
 279             if (i < j) {
 280                 for (; i<j; i+=step) {
 281                     lazy.emit('data', i)
 282                 }
 283             }
 284             else {
 285                 for (; i>j; i-=step) {
 286                     lazy.emit('data', i)
 287                 }
 288             }
 289             lazy.emit('end');
 290         });
 291     }
 292     return lazy;
 293 }
 294
 295 var mergeBuffers = function mergeBuffers(buffers) {
 296   // We expect buffers to be a non-empty Array
 297   if (!buffers || !Array.isArray(buffers) || !buffers.length) return;
 298
 299   var finalBufferLength, finalBuffer, currentBuffer, currentSize = 0;
 300
 301   // Sum all the buffers lengths
 302   finalBufferLength = buffers.reduce(function(left, right) { return (left.length||left) + (right.length||right); }, 0);
 303   finalBuffer = new Buffer(finalBufferLength);
 304   while(buffers.length) {
 305     currentBuffer = buffers.shift();
 306     currentBuffer.copy(finalBuffer, currentSize);
 307     currentSize += currentBuffer.length;
 308   }
 309
 310   return finalBuffer;
 311 }