yelouafi

// sequence of elements from an array that occur at a fixed interval 
// starting after an initial delay
// yields 1, 2 and 3 each second starting after five seconds
Stream.seq([1,2,3], 5000, 1000)

//stream of DOM events that occur until a promise is completed
// receives click events for 10 seconds
Stream.fromDomEvent(document.body, 'click', delay(10000) )

// Node events can also be thought as (possibly infinite) streams

yelouafi

// undefined equivalent for streams    
const undef = {}

// debounce : (Stream a, () => Promise) => Stream a
Stream.prototype.debounce = function(event, last=undef) {

  return  this.isEmpty || this.isAbort ? 
            (last !== undef ? Stream.Cons(last, this) : this) 
        : this.isCons ? 
            this.tail.debounce(event, this.head) 

yelouafi

// event : () => Promise
// throttle : (Stream a, event) => Stream a
Stream.prototype.throttle = function(event) {

  return  this.isEmpty || this.isAbort ? this 
        : this.isCons ? 
            Stream.Cons(this.head, this.tail.skipUntil(event()).throttle(event)) 
        : Stream.Future(this.promise.then( s => s.throttle(event), Stream.Abort));

};

yelouafi

// concatMap : (Stream a, a -> Stream b) => Stream b
Stream.prototype.concatMap = function(f) {
  return this.map(f).concatAll();
}

// example : log bodies from ajax responses
button.on('click').concatMap( _ => ajaxBody(url) )

yelouafi

// mergeMap : (Stream a, a -> Stream b) => Stream b
Stream.prototype.mergeMap = function(f) {
  return this.map(f).mergeAll();
}

// example : log messages from a chat room
// login user returns a stream of messages from a logged user
loginUser : (name, password) -> Stream String
button.on('click').mergeMap( _ => loginUser(name, password) )

yelouafi

// concatAll : Stream (Stream a) -> Stream a
Stream.prototype.concatAll = function() { 
  return this.flatten( (s1, s2) => s1.concat(s2)  );
}

yelouafi

// mergeAll : Stream (Stream a) -> Stream a
Stream.prototype.mergeAll = function() { 
  return this.flatten( (s1, s2) => s1.merge(s2)  );
}

yelouafi

// flatten : ( Stream (Stream a), (Stream a, Stream a) -> Stream a) -> Stream a
Stream.prototype.flatten = function(f) { 

  return  this.isEmpty || this.isAbort ? this 
        : this.isCons ? f( this.head, this.tail.flatten(f) ) 
        : Stream.Future(this.promise.then(s => s.flatten(f), Stream.Abort));
};

yelouafi

// zip : (Stream a, Stream b) => Stream [a,b]
Stream.prototype.zip = function(s2) { 

  return  this.isEmpty || this.isAbort  ? this 
        : s2.isEmpty || s2.isAbort      ? s2 
        : this.isCons && s2.isCons      ? 
            Stream.Cons([this.head, s2.head], this.tail.zip(s2.tail)) 
        : this.isCons && s2.isFuture    ? 
            Stream.Future(s2.promise.then(s => this.zip(s), Stream.Abort))
        : // this.isFuture && (s2.isCons || s2.isFuture)

yelouafi

// merge : (Stream a, Stream a) => Stream a
Stream.prototype.merge = function(s2) { 

  return  this.isEmpty ? s2 
        : this.isAbort ? this 
        : this.isCons  ? Stream.Cons( this.head, this.tail.merge(s2) ) 
        : ( !s2.isFuture ? 
              s2.merge(this) 
            : Stream.Future(
                Promise.race([