Test Async Queue + Streams

test1.js

S = require("./support/jsftp/support/streamer/core");

function queue() {
   var next;
   var buffer = Array.prototype.slice.call(arguments)

   function stream($, stop) { next = $; stream._update() }

   stream._update = function _update() {
    buffer.push.apply(buffer, arguments)
    if (next && buffer.length) {
      if (false !== next(buffer.shift())) _update()
    }
  }
  return stream
}

function enqueue(stream, element) {
   stream._update.apply(null, Array.prototype.slice.call(arguments, 1))
}

var pasvReqs = queue();

!function recur() {
    // Take first element from `pasvReqs` and process it via `processElement` black box that will call
   // `recur` once it's done to process next element from the `pasvReqs` queue.
    S.head(pasvReqs)(function(element) {
       processElement(element, recur)
    })
}()


var processElement = function(el, f) {
    var cb = function() {
        el.callback();
        f();
    }
    setTimeout(cb, parseInt(Math.random() * 1000));
}

enqueue(pasvReqs, { callback: function() { console.log("1..."); }});
enqueue(pasvReqs, { callback: function() { console.log("2..."); }});
enqueue(pasvReqs, { callback: function() { console.log("3..."); }});

https://gist.github.com/1200696

…

On Wed, Sep 7, 2011 at 6:04 PM, Irakli Gozalishvili ***@***.*** wrote: So reason you need that line is that reader stops reading stream by returning false from  `next`. That's also what `head` does, but since `enqueue` calls `_update` which calls next if  `next && buffer.length` this code misbehaves, each call to `enqueue` will call next. Setting `next` to `null` fixes an issue as if condition will be false until next reader starts reading. Hope this helps! ## Reply to this email directly or view it on GitHub: https://gist.github.com/1200696

Dude your comment is empty!

Heh. It worked great, but now I have to deal with an architecture fuckup in my implementation :) Thanks!

…

On Thu, Sep 8, 2011 at 10:17 AM, Irakli Gozalishvili ***@***.*** wrote: Dude your comment is empty! ## Reply to this email directly or view it on GitHub: https://gist.github.com/1200696