cwharris

var Rx = require('rx'),
    log = console.log.bind(console);
    
Rx.window = Rx; // #hack to make rx-dom work
require('rx-dom'); 

var prop = 'which',
    map = {
      37: 'west',
      38: 'north',

cwharris

Rx.Observable.iterate = function (enumerable, maxConcurrent, workSelector) {
  return Rx.Observable.create(function (o) {

    var i = 0,
        items = [],
        enumerator = enumerable.getEnumerator(),
        done = new Rx.Subject(),
        more = function () {
          if (enumerator.moveNext()) {
            return true;

cwharris

Rx.Observable.prototype.delayMinimumInterval = function (relativeTime, scheduler) {
  if (scheduler === undefined) scheduler = Rx.Scheduler.timeout;

  var latest = scheduler.now();

  return this
    .selectMany(function (x) {
      var now = new Date(scheduler.now());
      var next = new Date(latest);
      next.setMilliseconds(next.getMilliseconds() + relativeTime);

cwharris

var n = Math.floor(Math.random() * 4);
var result = n === 1 ? "one"
           : n === 2 ? "two"
           : n === 3 ? "three"
           : "missing-no";

cwharris

class Program
    {
        static void Main(string[] args)
        {
            Func<int, int, int> add = (a, b) => a + b;

            var obs1 = Observable.Empty<int>().Scan(add);
            var obs2 = Observable.Empty<int>().Scan(0, add);

            var obs3 = Observable.Range(1, 3).Scan(add);

cwharris

public static IObservable<TResult> BatchAsync<T, TResponse, TResult>(
    this IObservable<T> source,
    int count,
    Func<IEnumerable<T>, CancellationToken, Task<TResponse>> process,
    Func<TResponse, IEnumerable<TResult>> resultSelector
    )
{
    return source
        .Buffer(count)
        .SelectMany(batch =>

cwharris

var Rx = require('rx'),
    Rx = require('./rx.helpers')
    ;

var FlowLogic = (function () {

  Rx.Internals.inherits(FlowLogic, Rx.Observable);

  // pendingPours is a behavior subject. 
  // solenoid is an observer. It represents the valve allowing drink flow.

cwharris

var model = new Rx.Subject();
var click = new Rx.Subject();

model.latestOn(click, function (model, e) { return [model, e]; })
  .subscribe(function (x) {
    console.log(x);
  });

cwharris

var Rx = require('rx');

var a = new Rx.Subject();
var b = new Rx.Subject();
var c = new Rx.Subject();

var resA = Rx.Observable
  .when(
    a.and(b)
    .then(function (a, b) { return a + b; })

cwharris

Rx.Observable.prototype.autoPublish = function () {
  var source = this.publish(),
      subscribers = 0,
      connection = null,
      dispose = function () {
        if (--subscribers === 0) { connection.dispose(); }
      };

  return Rx.Observable.createWithDisposable(function (o) {
    if (++subscribers === 1) { connection = source.connect(); }