feat(operator): add materialize. closes #132

spec/Notification-spec.js

@@ -0,0 +1,43 @@
+/* globals describe, it, expect */
+var Rx = require('../dist/cjs/Rx');
+
+var Notification = Rx.Notification;
+
+describe('Notification', function () {
+  it('should exist', function () {
+    expect(typeof Notification).toBe('function');
+  });
+
+  describe('createNext', function () {
+    it('should return a Notification', function () {
+      var n = Notification.createNext('test');
+      expect(n instanceof Notification).toBe(true);
+      expect(n.value).toBe('test');
+      expect(n.kind).toBe('N');
+      expect(typeof n.exception).toBe('undefined');
+      expect(n.hasValue).toBe(true);
+    });
+  });
+
+  describe('createError', function () {
+    it('should return a Notification', function () {
+      var n = Notification.createError('test');
+      expect(n instanceof Notification).toBe(true);
+      expect(typeof n.value).toBe('undefined');
+      expect(n.kind).toBe('E');
+      expect(n.exception).toBe('test');
+      expect(n.hasValue).toBe(false);
+    });
+  });
+
+  describe('createComplete', function () {
+    it('should return a Notification', function () {
+      var n = Notification.createComplete();
+      expect(n instanceof Notification).toBe(true);
+      expect(typeof n.value).toBe('undefined');
+      expect(n.kind).toBe('C');
+      expect(typeof n.exception).toBe('undefined');
+      expect(n.hasValue).toBe(false);
+    });
+  });
+});

spec/operators/materialize-spec.js

@@ -0,0 +1,43 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+var Notification = Rx.Notification;
+
+describe('Observable.prototype.materialize()', function () {
+  it('should materialize a happy stream', function () {
+    var expected = [
+      Notification.createNext(1),
+      Notification.createNext(2),
+      Notification.createNext(3),
+      Notification.createComplete()
+    ];
+
+    Observable.of(1, 2, 3)
+      .materialize()
+      .subscribe(function (n) {
+        expect(n instanceof Notification).toBe(true);
+        expect(n).toEqual(expected.shift());
+      });
+  });
+
+  it('should materialize a sad stream', function () {
+    var expected = [
+      Notification.createNext(1),
+      Notification.createNext(2),
+      Notification.createNext(3),
+      Notification.createError('booooo')
+    ];
+
+    Observable.of(1, 2, 3, 4)
+      .map(function (x) {
+        if (x === 4) {
+          throw 'booooo';
+        }
+        return x;
+      })
+      .materialize()
+      .subscribe(function (n) {
+        expect(n).toEqual(expected.shift());
+      });
+  });
+});

src/Notification.ts

@@ -0,0 +1,67 @@
+import Observer from './Observer';
+import Observable from './Observable';
+import noop from './util/noop';
+
+export default class Notification<T> {
+  hasValue: boolean;
+
+  constructor(public kind: string, public value?: T, public exception?: any) {
+    this.hasValue = kind === 'N';
+  }
+
+  observe(observer: Observer<T>): any {
+    switch (this.kind) {
+      case 'N':
+        return observer.next(this.value);
+      case 'E':
+        return observer.error(this.exception);
+      case 'C':
+        return observer.complete();
+    }
+  }
+
+  do(next: (value: T) => void, error?: (err: any) => void, complete?: () => void): any {
+    const kind = this.kind;
+    switch (kind) {
+      case 'N':
+        return next(this.value);
+      case 'E':
+        return error(this.exception);
+      case 'C':
+        return complete();  
+    }
+  }
+
+  accept(nextOrObserver: Observer<T>|((value: T) => void), error?: (err: any) => void, complete?: () => void) {
+    if (nextOrObserver && typeof (<Observer<T>>nextOrObserver).next === 'function') {
+      return this.observe(<Observer<T>>nextOrObserver);
+    } else {
+      return this.do(<(value: T) => void>nextOrObserver, error, complete);
+    }
+  }
+
+  toObservable(): Observable<T> {
+    const kind = this.kind;
+    const value = this.value;
+    switch (kind) {
+      case 'N':
+        return Observable.value(value);  
+      case 'E':
+        return Observable.throw(value);
+      case 'C':
+        return Observable.empty();  
+    }
+  }
+
+  static createNext<T>(value: T): Notification<T> {
+    return new Notification('N', value);
+  }
+
+  static createError<T>(err: any): Notification<T> {
+    return new Notification('E', undefined, err);
+  }
+
+  static createComplete(): Notification<any> {
+    return new Notification('C');
+  }
+}

src/Observable.ts

@@ -5,6 +5,10 @@ import Scheduler from './Scheduler';
 import Subscriber from './Subscriber';
 import Subscription from './Subscription';
 import ConnectableObservable from './observables/ConnectableObservable';
+// HACK: the Babel part of the build doesn't like this reference.
+// seems to put it in an infinite loop.
+//import Notification from './Notification';
+
 
 import $$observer from './util/Symbol_observer';
 
@@ -131,6 +135,9 @@ export default class Observable<T> {
   partition: (predicate: (x: T) => boolean) => Observable<T>[];
   toPromise: (PromiseCtor: PromiseConstructor) => Promise<T>;
   defaultIfEmpty: <T, R>(defaultValue: R) => Observable<T>|Observable<R>;
+  // HACK: this should be Observable<Notification<T>>, but the build process didn't like it. :(
+  //   this will be fixed when we can move everything to the TypeScript compiler I suspect.
+  materialize: () => Observable<any>;
 
   observeOn: (scheduler: Scheduler, delay?: number) => Observable<T>;
   subscribeOn: (scheduler: Scheduler, delay?: number) => Observable<T>;

src/Rx.ts

@@ -3,6 +3,8 @@ import Scheduler from './Scheduler';
 import Observable from './Observable';
 import Subscriber from './Subscriber';
 import Subscription from './Subscription';
+import Notification from './Notification';
+
 import ReplaySubject from './subjects/ReplaySubject';
 import BehaviorSubject from './subjects/BehaviorSubject';
 import ConnectableObservable from './observables/ConnectableObservable';
@@ -124,10 +126,12 @@ observableProto.subscribeOn = subscribeOn;
 import partition from './operators/partition';
 import toPromise from './operators/toPromise';
 import defaultIfEmpty from './operators/defaultIfEmpty';
+import materialize from './operators/materialize';
 
 observableProto.partition = partition;
 observableProto.toPromise = toPromise;
 observableProto.defaultIfEmpty = defaultIfEmpty;
+observableProto.materialize = materialize;
 
 import _catch from './operators/catch';
 import retryWhen from './operators/retryWhen';
@@ -143,5 +147,6 @@ export default {
     Subscription,
     ReplaySubject,
     BehaviorSubject,
-    ConnectableObservable
+    ConnectableObservable,
+    Notification 
 };

src/operators/materialize.ts

@@ -0,0 +1,36 @@
+import Operator from '../Operator';
+import Observer from '../Observer';
+import Subscriber from '../Subscriber';
+import Notification from '../Notification';
+
+export default function materialize<T>() {
+  return this.lift(new MaterializeOperator());
+}
+
+export class MaterializeOperator<T, R> extends Operator<T, R> {
+  call(observer: Observer<T>): Observer<T> {
+    return new MaterializeSubscriber(observer);
+  }
+}
+
+export class MaterializeSubscriber<T> extends Subscriber<T> {
+  constructor(destination: Observer<T>) {
+    super(destination);
+  }
+
+  _next(value:T) {
+    this.destination.next(Notification.createNext(value));
+  }
+
+  _error(err: any) {
+    const destination = this.destination;
+    destination.next(Notification.createError(err));
+    destination.complete();
+  }
+
+  _complete() {
+    const destination = this.destination;
+    destination.next(Notification.createComplete());
+    destination.complete();
+  }
+}