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

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

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

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

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

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

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
  .concat(
    [
      Rx.Observable.return('{ "value": 0 }'),
      Rx.Observable.return('{ "value": 5 }'),
      Rx.Observable.return("{")
    ]
  )
  .tagError('Source')
  .tagError('ParseJSON', function (source) {

cwharris

movieFinder.find('Falcon')
  .flatMap(function (movie) {
    return geocoder
      .geocode(movie.locations)
      .take(1)
      .pluck('geometry.location')
      .map(function (location) {
        return {
          location: location,
          movie: movie

cwharris

Keybase proof

I hereby claim:

• I am cwharris on github.
• I am cwharris (https://keybase.io/cwharris) on keybase.
• I have a public key whose fingerprint is 1AF0 0C9A 28CB 9B59 F50A A3BB AD83 0673 F92A 4F0E

To claim this, I am signing this object: