jooyunghan

import { Observable } from "rxjs";

Observable.prototype[Symbol.asyncIterator] = createAsyncIterator;

async function* createAsyncIterator() {
  const promise = [];

  const values = [];
  let done = false;
  let error = null;

jooyunghan

const { Observable } = require('rxjs');

Observable.prototype[Symbol.asyncIterator] = createAsyncIterator;

function createAsyncIterator() {
  const promises = [];
  const values = [];
  let done = false;
  let error = null;
  const subscription = this.subscribe({

jooyunghan

async function consume2(obs) {
  const ai = obs[Symbol.asyncIterator]();
  ai.next().then(r => console.log(0, r), e => console.log(0, e.message));
  ai.next().then(r => console.log(1, r), e => console.log(1, e.message));
  ai.next().then(r => console.log(2, r), e => console.log(2, e.message));
  ai.next().then(r => console.log(3, r), e => console.log(3, e.message));
}
consume2(Observable.interval(100).do(n => console.assert(n < 1)));

jooyunghan

function createAsyncIterator() {
  const promises = [];
  const values = [];
  let done = false;
  let error = null;
  const subscription = this.subscribe({
    next(value) {
      if (promises.length > 0) {
        promises.shift().resolve({ value, done: false });
      } else {

jooyunghan

async function consume(obs) {
  try {
    for await (const n of obs) {
      console.log(n);
    }
  } catch (e) {
    console.log('got it', e.message);
  }
  console.log('done');  
}

jooyunghan

function createAsyncIterator() {
  const promises = [];
  const values = [];
  let done = false;

  const subscription = this.subscribe({
    next(value) {
      if (promises.length > 0) {
        promises.shift().resolve({ value, done: false });
      } else {

jooyunghan

function createAsyncIterator() {
  const promises = [];
  const values = [];

  this.subscribe({
    next(value) {
      if (promises.length > 0) {
        promises.shift().resolve({ value, done: false });
      } else {
        values.push(value);

jooyunghan

async function consume(obs) {
  for await (const n of obs) {
    console.log(n);
    if (n >= 5) break;
  }
}
consume(Observable.interval(100));

jooyunghan

function createAsyncIterator() {
  const promises = [];
  this.subscribe({
    next(value) {
      promises.shift().resolve({ value, done: false });
    }
  });
  return {
    next() {
      return new Promise((resolve, reject) => {

jooyunghan

var noop = [];

function stream(arr){
    if(arr.length === 0) return noop;
    return cons(function (){return arr[0]}, function (){ return stream(arr.slice(1))})
}

function cons(head, tail){
    return [head, tail];
}