Support NodeJS Event Emitters in fromEvent extra

src/extra/fromEvent.ts

@@ -1,3 +1,5 @@
+/// <reference path="../../typings/globals/node/index.d.ts" />
+import {EventEmitter} from 'events';
 import {Stream, InternalProducer, InternalListener} from '../core';
 
 export class DOMEventProducer implements InternalProducer<Event> {
@@ -11,29 +13,49 @@ export class DOMEventProducer implements InternalProducer<Event> {
 
   _start(out: InternalListener<Event>) {
     this.listener = (e) => out._n(e);
-    const {node, eventType, useCapture} = this;
-    node.addEventListener(eventType, this.listener, useCapture);
+    this.node.addEventListener(this.eventType, this.listener, this.useCapture);
   }
 
   _stop() {
-    const {node, eventType, listener, useCapture} = this;
-    node.removeEventListener(eventType, listener, useCapture);
+    this.node.removeEventListener(this.eventType, this.listener, this.useCapture);
     this.listener = null;
   }
 }
 
+export class NodeEventProducer implements InternalProducer<any> {
+  public type = 'fromEvent';
+  private listener: Function;
+
+  constructor(private node: EventEmitter, private eventName: string) { }
+
+  _start(out: InternalListener<any>) {
+    this.listener = (e: any) => out._n(e);
+    this.node.addListener(this.eventName, this.listener);
+  }
+
+  _stop() {
+    this.node.removeListener(this.eventName, this.listener);
+    this.listener = null;
+  }
+}
+
+function isEmitter(element: any): boolean {
+    return element.addListener;
+}
+
 /**
- * Creates a stream based on DOM events of type `eventType` from the target
- * node.
+ * Creates a stream based on either:
+ * - DOM events with the name `eventName` from a provided target node
+ * - Events with the name `eventName` from a provided NodeJS EventEmitter
  *
  * Marble diagram:
  *
  * ```text
- *   fromEvent(node, eventType)
+ *   fromEvent(element, eventName)
  * ---ev--ev----ev---------------
  * ```
  *
- * Example:
+ * Examples:
  *
  * ```js
  * import fromEvent from 'xstream/extra/fromEvent'
@@ -54,15 +76,39 @@ export class DOMEventProducer implements InternalProducer<Event> {
  * > 'Button clicked!'
  * ```
  *
- * @param {EventTarget} node The element we want to listen to.
- * @param {string} eventType The type of events we want to listen to.
- * @param {boolean} useCapture An optional boolean that indicates that events of
+ * ```js
+ * import fromEvent from 'xstream/extra/fromEvent'
+ * import {EventEmitter} from 'events'
+ *
+ * const MyEmitter = new EventEmitter()
+ * const stream = fromEvent(MyEmitter, 'foo')
+ *
+ * stream.addListener({
+ *   next: i => console.log(i),
+ *   error: err => console.error(err),
+ *   complete: () => console.log('completed')
+ * })
+ *
+ * MyEmitter.emit('foo', 'bar')
+ * ```
+ *
+ * ```text
+ * > 'bar'
+ * ```
+ *
+ * @param {EventTarget|EventEmitter} element The element upon which to listen.
+ * @param {string} eventName The name of the event for which to listen.
+ * @param {boolean?} useCapture An optional boolean that indicates that events of
  * this type will be dispatched to the registered listener before being
  * dispatched to any EventTarget beneath it in the DOM tree. Defaults to false.
  * @return {Stream}
  */
-export default function fromEvent(node: EventTarget,
-                                  eventType: string,
-                                  useCapture: boolean = false): Stream<Event> {
-  return new Stream<Event>(new DOMEventProducer(node, eventType, useCapture));
+export default function fromEvent(element: EventTarget | EventEmitter,
+                                  eventName: string,
+                                  useCapture: boolean = false): Stream<Event|any> {
+  if (isEmitter(element)) {
+    return new Stream<any>(new NodeEventProducer(<EventEmitter> element, eventName));
+  } else {
+    return new Stream<Event>(new DOMEventProducer(<EventTarget> element, eventName, useCapture));
+  }
 }

tests/extra/fromEvent.ts

@@ -1,6 +1,6 @@
 /// <reference path="../../typings/globals/mocha/index.d.ts" />
 /// <reference path="../../typings/globals/node/index.d.ts" />
-import xs from '../../src/index';
+import {EventEmitter} from 'events';
 import fromEvent from '../../src/extra/fromEvent';
 import * as assert from 'assert';
 function noop() {};
@@ -39,15 +39,47 @@ class FakeEventTarget implements EventTarget {
   }
 };
 
-describe('fromEvent (extra)', () => {
+class FakeEventEmitter extends EventEmitter {
+  public handler: Function;
+  public event: string;
+  public removedEvent: string;
+
+  constructor() {
+    super();
+  }
+
+  emit( eventName: string, ...args: any[] ): any {
+    if (typeof this.handler !== 'function') {
+        return;
+    }
+    this.handler.apply(void 0, args );
+    return true;
+  }
+
+  addListener(e: string, handler: Function): FakeEventEmitter {
+    this.event = e;
+    this.handler = handler;
+    return this;
+  }
+
+  removeListener(e: string, handler: Function): FakeEventEmitter {
+    this.removedEvent = e;
+
+    this.handler = this.event = void 0;
+    return this;
+  }
+
+};
+
+describe('fromEvent (extra) - DOMEvent', () => {
   it('should call addEventListener with expected parameters', () => {
     const target = new FakeEventTarget();
     const stream = fromEvent(target, 'test', true);
 
     stream.addListener({next: noop, error: noop, complete: noop});
 
-    assert.strictEqual('test', target.event);
-    assert.strictEqual(true, target.capture);
+    assert.strictEqual(target.event, 'test');
+    assert.strictEqual(target.capture, true);
   });
 
   it('should call addEventListener with expected parameters', () => {
@@ -56,8 +88,8 @@ describe('fromEvent (extra)', () => {
 
     stream.addListener({next: noop, error: noop, complete: noop});
 
-    assert.strictEqual('test', target.event);
-    assert.strictEqual(false, target.capture);
+    assert.strictEqual(target.event, 'test');
+    assert.strictEqual(target.capture, false);
   });
 
   it('should propagate events', (done) => {
@@ -92,8 +124,8 @@ describe('fromEvent (extra)', () => {
       error: (err: any) => done(err),
       complete() {
         setTimeout(() => {
-          assert.strictEqual('test', target.removedEvent);
-          assert.strictEqual(true, target.removedCapture);
+          assert.strictEqual(target.removedEvent, 'test');
+          assert.strictEqual(target.removedCapture, true);
           done();
         }, 5);
       }
@@ -103,3 +135,56 @@ describe('fromEvent (extra)', () => {
     target.emit(2);
   });
 });
+
+describe('fromEvent (extra) - EventEmitter', () => {
+  it('should call addListener with expected parameters', () => {
+    const target = new FakeEventEmitter();
+    const stream = fromEvent(target, 'test');
+
+    stream.addListener({next: noop, error: noop, complete: noop});
+
+    assert.strictEqual(target.event, 'test');
+  });
+
+  it('should propagate events', (done) => {
+    const target = new FakeEventEmitter();
+    const stream = fromEvent(target, 'test').take(3);
+
+    let expected = [1, 2, 3];
+
+    stream.addListener({
+      next: (x: any) => {
+        assert.strictEqual(x, expected.shift());
+      },
+      error: (err: any) => done(err),
+      complete: () => {
+        assert.strictEqual(expected.length, 0);
+        done();
+      }
+    });
+
+    target.emit( 'test', 1 );
+    target.emit( 'test', 2 );
+    target.emit( 'test', 3 );
+    target.emit( 'test', 4 );
+  });
+
+  it('should call removeListener with expected parameters', (done) => {
+    const target = new FakeEventEmitter();
+    const stream = fromEvent(target, 'test');
+
+    stream.take(1).addListener({
+      next: (x) => {},
+      error: (err: any) => done(err),
+      complete() {
+        setTimeout(() => {
+          assert.strictEqual(target.removedEvent, 'test');
+          done();
+        }, 5);
+      }
+    });
+
+    target.emit( 'test', 1 );
+    target.emit( 'test', 2 );
+  });
+});