fix(operators): improve *type* metadata for operators with fusion

flatten, flattenConcurrently and filter+map should show whether the its map or mapTo in the type
metadata field

src/core.ts

@@ -780,12 +780,13 @@ class MFCIL<R> implements InternalListener<R> {
 }
 
 export class MapFlattenConcOperator<T, R> implements Operator<T, R> {
-  public type = 'map+flattenConcurrently';
+  public type: string;
   public ins: Stream<T>;
   private active: number = 1; // number of outers and inners that have not yet ended
   private out: Stream<R> = null;
 
   constructor(public mapOp: MapOperator<T, Stream<R>>) {
+    this.type = `${mapOp.type}+flattenConcurrently`;
     this.ins = mapOp.ins;
   }
 
@@ -850,14 +851,15 @@ class MFIL<R> implements InternalListener<R> {
 }
 
 export class MapFlattenOperator<T, R> implements Operator<T, R> {
-  public type = 'map+flatten';
+  public type: string;
   public ins: Stream<T>;
   public inner: Stream<R> = null; // Current inner Stream
   private il: InternalListener<R> = null; // Current inner InternalListener
   private open: boolean = true;
   private out: Stream<R> = null;
 
   constructor(public mapOp: MapOperator<T, Stream<R>>) {
+    this.type = `${mapOp.type}+flatten`;
     this.ins = mapOp.ins;
   }
 

tests/operator/filter.ts

@@ -78,4 +78,24 @@ describe('Stream.prototype.filter', () => {
       }
     });
   });
+
+  it('should should have filter+map fusion metadata', (done) => {
+    const isEven = (x: number) => x % 2 === 0;
+    const stream = xs.of(1, 2, 3, 4, 5, 6, 7, 8)
+      .filter(isEven)
+      .map(x => 10 * x);
+
+    assert.strictEqual(stream['_prod']['type'], 'filter+map');
+    done();
+  });
+
+  it('should should have filter+mapTo fusion metadata', (done) => {
+    const isEven = (x: number) => x % 2 === 0;
+    const stream = xs.of(1, 2, 3, 4, 5, 6, 7, 8)
+      .filter(isEven)
+      .mapTo(10);
+
+    assert.strictEqual(stream['_prod']['type'], 'filter+mapTo');
+    done();
+  });
 });

tests/operator/flatten.ts

@@ -139,4 +139,14 @@ describe('Stream.prototype.flatten', () => {
       });
     });
   });
+
+  describe('with mapTo', () => {
+    it('should have the correct \'type\' metadata on the operator producer', (done) => {
+      const source: Stream<Stream<number>> = xs.periodic(100).take(3)
+        .mapTo(xs.of(1, 2, 3));
+      const stream: Stream<number> = source.flatten();
+      assert.strictEqual(stream['_prod']['type'], 'mapTo+flatten');
+      done();
+    });
+  });
 });

tests/operator/flattenConcurrently.ts

@@ -140,4 +140,14 @@ describe('Stream.prototype.flattenConcurrently', () => {
       });
     });
   });
+
+  describe('with mapTo', () => {
+    it('should have the correct \'type\' metadata on the operator producer', (done) => {
+      const source: Stream<Stream<number>> = xs.periodic(100).take(3)
+        .mapTo(xs.of(1, 2, 3));
+      const stream: Stream<number> = source.flattenConcurrently();
+      assert.strictEqual(stream['_prod']['type'], 'mapTo+flattenConcurrently');
+      done();
+    });
+  });
 });