yelouafi

// aBoolean : Object (JavaScript view of truth)
// filter : (Stream a, a -> aBoolean) => Stream a
Stream.prototype.filter = function(pred) {

  return  this.isEmpty || this.isAbort ? this
        : this.isCons ?
            (pred(this.head) ?
                Stream.Cons(this.head, this.tail.filter(pred)) 
              : this.tail.filter(pred))
        : Stream.Future(

yelouafi

span :: (a -> Bool) -> [a] -> ([a],[a])
span _ this@[] = (this, this)
span p this@(x:xs)
         | p x          =  let (ys,zs) = span p xs in (x:ys,zs)
         | otherwise    =  ([],this)

yelouafi

// span : (Stream a, a -> aBoolean) -> [Stream a, Stream a]
Stream.prototype.span = function(p) { 
  let s1, s2, futs;
 
  return  this.isEmpty || this.isAbort ? [this, this]
        : this.isCons ? 
           (p(this.head) ?
              ([s1, s2] = this.tail.span(p), [Stream.Cons(this.head, s1), s2])
            : [Stream.Empty, this])
        : (futs = this.promise.then(s => s.span(p), Stream.Abort),

yelouafi

// groupBy : (Stream a, (a, a) -> aBoolean) => Stream (Stream a)
Stream.prototype.groupBy = function(p) {
  var s1, s2;

  return  this.isEmpty || this.isAbort ? this 
        : this.isCons ?
          ( [s1, s2] = this.tail.span( x => p(this.head, x) ), 
            Stream.Cons(
              Stream.Cons(this.head, s1),
              s2.groupBy(p))) 

yelouafi

// group : Stream a => Stream (Stream a)
Stream.prototype.group = function() {
  return this.groupBy( (x1, x2) => x1 === x2  );
}

yelouafi

// reduce : ( Stream a, ((b, a) -> a), b  ) -> Promise b
Stream.prototype.reduce = function(f, seed) { 
  return  this.isEmpty  ? Promise.resolve(seed) 
        : this.isAbort  ? Promise.reject(this.error) 
        : this.isCons   ? this.tail.reduce( f, f(seed, this.head) ) 
        : /* isFuture */  this.promise.then( s => s.reduce(f, seed), Promise.reject )  
}

Stream.seq([1,2,3,4,5], 0, 1000)
  .reduce((x, y) => x + y, 0)

yelouafi

// length : Stream a -> Promise Number
Stream.prototype.length = function() { 
    return  this.reduce( (n, _) =>  n + 1, 0 )
}

yelouafi

// all : ( Stream a, a -> aBoolean ) -> Promise Boolean
Stream.prototype.all = function(pred)  {
  return this.reduce( (prev, cur) => prev && !!pred(cur), true );
}

Stream.seq([1,2,3,4,5], 0, 1000).all( x => x % 2) // => false

yelouafi

// any : ( Stream a, a -> aBoolean ) -> Promise Boolean
Stream.prototype.any = function(pred)  {

  return  this.isEmpty ? Promise.resolve(false) 
        : this.isAbort ? Promise.reject(this.err) 
        : this.isCons  ? 
            (pred(this.head) ? Promise.resolve(true) : this.tail.any(pred) ) 
        : /* isFuture */ 
            this.promise.then( s => s.any(pred) );
}

yelouafi

// takeUntil  : (Stream a, Promise) => Stream a
Stream.prototype.takeUntil = function(untilP) { 

  return  this.isEmpty || this.isAbort ?  this 
        : this.isCons ? Stream.Cons( this.head, this.tail.takeUntil(untilP) ) 
        : Stream.Future(
          // can you spot a problem here ?
          Promise.race([
            untilP.then( _ => Stream.Empty, Stream.Abort ),
            this.promise.then( s => s.takeUntil(untilP), Stream.Abort )