Embed extra operators to Stream

src/extra/buffer.ts

@@ -109,3 +109,13 @@ export default function buffer<T>(s: Stream<any>): (ins: Stream<T>) => Stream<Ar
     return new Stream<Array<T>>(new BufferOperator<T>(s, ins));
   };
 }
+
+declare module '../index' {
+  interface Stream<T> {
+    buffer<T>(s: Stream<any>): Stream<Array<T>>;
+  }
+}
+
+Stream.prototype.buffer = function (s: any): any {
+  return this.compose(buffer(s));
+};

src/extra/dropRepeats.ts

@@ -115,3 +115,13 @@ export default function dropRepeats<T>(isEqual: ((x: T, y: T) => boolean) | unde
     return new Stream<T>(new DropRepeatsOperator<T>(ins, isEqual));
   };
 }
+
+declare module '../index' {
+  interface Stream<T> {
+    dropRepeats<T>(isEqual?: ((x: T, y: T) => boolean) | undefined): Stream<T>;
+  }
+}
+
+Stream.prototype.dropRepeats = function (isEqual: any): any {
+  return this.compose(dropRepeats(isEqual));
+};

src/extra/flattenConcurrently.ts

@@ -91,3 +91,13 @@ export class FlattenConcOperator<T> implements Operator<Stream<T>, T> {
 export default function flattenConcurrently<T>(ins: Stream<Stream<T>>): Stream<T> {
   return new Stream<T>(new FlattenConcOperator(ins));
 }
+
+declare module '../index' {
+  interface Stream<T> {
+    flattenConcurrently<R>(this: Stream<Stream<R>>): T;
+  }
+}
+
+Stream.prototype.flattenConcurrently = function (): any {
+  return this.compose(flattenConcurrently);
+};

src/extra/flattenSequentially.ts

@@ -125,3 +125,13 @@ export class FlattenSeqOperator<T> implements Operator<Stream<T>, T> {
 export default function flattenSequentially<T>(ins: Stream<Stream<T>>): Stream<T> {
   return new Stream<T>(new FlattenSeqOperator(ins));
 }
+
+declare module '../index' {
+  interface Stream<T> {
+    flattenSequentially<R>(this: Stream<Stream<R>>): T;
+  }
+}
+
+Stream.prototype.flattenSequentially = function (): any {
+  return this.compose(flattenSequentially);
+};

tests/extra/buffer.ts

@@ -36,7 +36,7 @@ describe('buffer (extra)', () => {
   it('should flush and complete when source is completed', (done) => {
     const source = xs.of(1, 2).compose(delay(10));
     const separator = xs.empty().compose(delay(50));
-    const buffered = source.compose(buffer(separator));
+    const buffered = source.buffer(separator);
     const expected = [[1, 2]];
 
     let separatorComplete = false;

tests/extra/dropRepeats.ts

@@ -24,7 +24,7 @@ describe('dropRepeats (extra)', () => {
 
   it('should drop consecutive \'duplicate\' strings, with a custom isEqual', (done: any) => {
     const stream = xs.of('a', 'b', 'a', 'A', 'B', 'b')
-      .compose(dropRepeats((x: string, y: string) => x.toLowerCase() === y.toLowerCase()));
+      .dropRepeats((x: string, y: string) => x.toLowerCase() === y.toLowerCase());
     const expected = ['a', 'b', 'a', 'B'];
 
     stream.addListener({

tests/extra/flattenConcurrently.ts

@@ -9,7 +9,7 @@ describe('flattenConcurrently (extra)', () => {
     it('should expand each periodic event with 3 sync events', (done: any) => {
       const stream = xs.periodic(100).take(3)
         .map(i => xs.of(1 + i, 2 + i, 3 + i))
-        .compose(flattenConcurrently);
+        .flattenConcurrently();
       const expected = [1, 2, 3, 2, 3, 4, 3, 4, 5];
 
       stream.addListener({

tests/extra/flattenSequentially.ts

@@ -9,7 +9,7 @@ describe('flattenSequentially (extra)', () => {
     it('should expand each periodic event with 3 sync events', (done: any) => {
       const stream = xs.periodic(100).take(3)
         .map(i => xs.of(1 + i, 2 + i, 3 + i))
-        .compose(flattenSequentially);
+        .flattenSequentially();
       const expected = [1, 2, 3, 2, 3, 4, 3, 4, 5];
       const listener = {
         next: (x: number) => {