xstream producers

producers.js

import fs from 'fs';
import assert from 'assert';
import xs from 'xstream';

function xray (url, classToScrape) {
  return function scrape (callback) {
    // goes and parses internetty stuff

    callback(stuff);
  }
}

xray('https://steveiscool.com', 'h1.name')(function (name) {
  assert.equal(name, 'Steve');
});

xsfromCallback(xray('https://steveiscool.com', 'h1.name'))
  .addListener({
    next (name) {
      assert.equal(name, 'Steve');
    },

    // ...
  })


function xsFromCallback (callbackableFunction) {
  const stream = xs.create();

  callbackableFunction(function (data) {
    stream.shamefullySendNext(data);
  })

  return stream;
}

function xsFromCallback (callbackableFunction) {
  return xs.create({
    start (subject) {
      callbackableFunction(function (data) {
        subject.next(data);
      });
    },

    stop () {
    }
  });
}

function xsFromWebsocket (socket) {
  return xs.create({
    start (subject) {
      socket.onmessage = function (data) {
        subject.next(data);
      };
    },

    stop () {
    }
  });
}

xsFromWebsocket(new WebSocket('ws://localhost:3000'));