last.js

export default function(self) {
  let last = null;
  let isEmpty = true;

  return self
    .forEach(value => {
      last = value;
      isEmpty = false;
    })
    .then(() => {
      if (isEmpty)
        throw new TypeError('[.last] called on a empty stream');

      return last;
    });
}

readable.js

import PromiseCancellable from '../promise-cancellable';

import Readable_static_empty from './static-empty';
import Readable_static_fromPromise from './static-from-promise';
import Readable_static_merge from './static-merge';

import Readable_first from './first';
import Readable_last from './last';
import Readable_single from './single';
import Readable_isEmpty from './is-empty';
import Readable_length from './length';

import Readable_get from './get';
import Readable_distinct from './distinct';
import Readable_until from './until';
import Readable_delay from './delay';
import Readable_accumulate from './accumulate';


export default class Readable extends PromiseCancellable {

  static empty() {
    return Readable_static_empty(this);
  }
  static fromPromise(promise) {
    return Readable_static_fromPromise(this, promise);
  }
  static merge(...streams) {
    // TODO: posponed because we don't know when to pull from the streams
    return Readable_static_merge(this, streams);
  }

  get first() {
    return Readable_first(this);
  }
  get last() {
    return Readable_last(this);
  }
  get single() {
    return Readable_single(this);
  }
  get isEmpty() {
    return Readable_isEmpty(this);
  }
  get length() {
    return Readable_length(this);
  }


  get(index) {
    return Readable_get(this, index);
  }
  distinct(comparator = (first, second) => first === second) {
    return Readable_distinct(this, comparator);
  }
  until(stopper) {
    return Readable_until(this, stopper);
  }
  delay(milliseconds) {
    return Readable_delay(this, milliseconds);
  }
  accumulate(iterator, seed) {
    return Readable_accumulate(this, iterator, seed);
  }
}