k$ -> observable.pipe

example_plugins/baz/src/BazService.ts

@@ -1,4 +1,4 @@
-import { k$, Observable, $combineLatest, map, first, toPromise } from 'kbn-observable';
+import { Observable, $combineLatest, map, first, toPromise } from 'kbn-observable';
 
 import { ElasticsearchService, KibanaConfig, KibanaRequest } from 'kbn-types';
 
@@ -13,7 +13,7 @@ export class BazService {
     const { page = 1, perPage = 20, type } = options;
 
     const [kibanaIndex, adminCluster] = await latestValues(
-      k$(this.kibanaConfig$)(map(config => config.index)),
+      this.kibanaConfig$.pipe(map(config => config.index)),
       this.elasticsearchService.getClusterOfType$('admin')
     );
 
@@ -64,7 +64,5 @@ function latestValues<A, B, C, D>(
   d: Observable<D>
 ): Promise<[A, B, C, D]>;
 function latestValues(...values: Observable<any>[]) {
-  return k$($combineLatest(values))(
-    first(),
-    toPromise());
+  return $combineLatest(values).pipe(first(), toPromise());
 }

packages/kbn-internal-native-observable/__tests__/index.test.js

@@ -0,0 +1,34 @@
+import { Observable } from '../';
+
+const first = () => source =>
+  new Observable(observer =>
+    source.subscribe({
+      next(value) {
+        observer.next(value);
+        observer.complete();
+      }
+    })
+  );
+
+const plus = x => source =>
+  new Observable(observer =>
+    source.subscribe({
+      next(value) {
+        observer.next(value + x);
+      },
+      complete() {
+        observer.complete();
+      }
+    })
+  );
+
+test('can pipe values', () => {
+  const observable = Observable.of(1, 2, 3).pipe(plus(10), first());
+
+  let value;
+  observable.subscribe(x => {
+    value = x;
+  });
+
+  expect(value).toEqual(11);
+});

packages/kbn-internal-native-observable/index.d.ts

@@ -6,6 +6,8 @@ declare global {
   }
 }
 
+type UnaryFunction<T, R> = (source: T) => R;
+
 // These types are based on the Observable proposal readme, see
 // https://github.com/tc39/proposal-observable#api, with the addition of using
 // generics to define the type of the `value`.
@@ -109,6 +111,70 @@ declare namespace Observable {
       onComplete?: () => void
     ): Subscription;
 
+    // pipe
+    pipe(): Observable<T>;
+    pipe<A>(op1: UnaryFunction<Observable<T>, A>): A;
+    pipe<A, B>(
+      op1: UnaryFunction<Observable<T>, Observable<A>>,
+      op2: UnaryFunction<Observable<A>, B>
+    ): B;
+    pipe<A, B, C>(
+      op1: UnaryFunction<Observable<T>, Observable<A>>,
+      op2: UnaryFunction<Observable<A>, Observable<B>>,
+      op3: UnaryFunction<Observable<B>, C>
+    ): C;
+    pipe<A, B, C, D>(
+      op1: UnaryFunction<Observable<T>, Observable<A>>,
+      op2: UnaryFunction<Observable<A>, Observable<B>>,
+      op3: UnaryFunction<Observable<B>, Observable<C>>,
+      op4: UnaryFunction<Observable<C>, D>
+    ): D;
+    pipe<A, B, C, D, E>(
+      op1: UnaryFunction<Observable<T>, Observable<A>>,
+      op2: UnaryFunction<Observable<A>, Observable<B>>,
+      op3: UnaryFunction<Observable<B>, Observable<C>>,
+      op4: UnaryFunction<Observable<C>, Observable<D>>,
+      op5: UnaryFunction<Observable<D>, E>
+    ): E;
+    pipe<A, B, C, D, E, F>(
+      op1: UnaryFunction<Observable<T>, Observable<A>>,
+      op2: UnaryFunction<Observable<A>, Observable<B>>,
+      op3: UnaryFunction<Observable<B>, Observable<C>>,
+      op4: UnaryFunction<Observable<C>, Observable<D>>,
+      op5: UnaryFunction<Observable<D>, Observable<E>>,
+      op6: UnaryFunction<Observable<E>, F>
+    ): F;
+    pipe<A, B, C, D, E, F, G>(
+      op1: UnaryFunction<Observable<T>, Observable<A>>,
+      op2: UnaryFunction<Observable<A>, Observable<B>>,
+      op3: UnaryFunction<Observable<B>, Observable<C>>,
+      op4: UnaryFunction<Observable<C>, Observable<D>>,
+      op5: UnaryFunction<Observable<D>, Observable<E>>,
+      op6: UnaryFunction<Observable<E>, Observable<F>>,
+      op7: UnaryFunction<Observable<F>, G>
+    ): G;
+    pipe<A, B, C, D, E, F, G, H>(
+      op1: UnaryFunction<Observable<T>, Observable<A>>,
+      op2: UnaryFunction<Observable<A>, Observable<B>>,
+      op3: UnaryFunction<Observable<B>, Observable<C>>,
+      op4: UnaryFunction<Observable<C>, Observable<D>>,
+      op5: UnaryFunction<Observable<D>, Observable<E>>,
+      op6: UnaryFunction<Observable<E>, Observable<F>>,
+      op7: UnaryFunction<Observable<F>, Observable<G>>,
+      op8: UnaryFunction<Observable<G>, H>
+    ): H;
+    pipe<A, B, C, D, E, F, G, H, I>(
+      op1: UnaryFunction<Observable<T>, Observable<A>>,
+      op2: UnaryFunction<Observable<A>, Observable<B>>,
+      op3: UnaryFunction<Observable<B>, Observable<C>>,
+      op4: UnaryFunction<Observable<C>, Observable<D>>,
+      op5: UnaryFunction<Observable<D>, Observable<E>>,
+      op6: UnaryFunction<Observable<E>, Observable<F>>,
+      op7: UnaryFunction<Observable<F>, Observable<G>>,
+      op8: UnaryFunction<Observable<G>, Observable<H>>,
+      op9: UnaryFunction<Observable<H>, I>
+    ): I;
+
     // Returns itself
     [Symbol.observable](): Observable<T>;
 

packages/kbn-internal-native-observable/index.js

@@ -255,6 +255,14 @@ export class Observable {
 
     return new Subscription(observer, this._subscriber);
   }
+
+  pipe(...operations) {
+    if (operations.length === 0) {
+      return this;
+    }
+
+    return operations.reduce((prev, fn) => fn(prev), this);
+  }
 
   [symbolObservable]() { return this }
 

packages/kbn-observable/package.json

@@ -11,6 +11,6 @@
     "rxjs": "5.4.3"
   },
   "devDependencies": {
-    "typescript": "^2.5.3"
+    "typescript": "2.5.3"
   }
 }

packages/kbn-observable/src/__tests__/Subject.test.ts

@@ -1,6 +1,5 @@
 import { Observable } from '../Observable';
 import { Subject } from '../Subject';
-import { k$ } from '../k$';
 import { first } from '../operators';
 
 const noop = () => {};
@@ -340,7 +339,7 @@ test('can use subject in $k', async () => {
   const complete = jest.fn();
   const error = jest.fn();
 
-  k$(values$)(first()).subscribe({
+  values$.pipe(first()).subscribe({
     next,
     error,
     complete

packages/kbn-observable/src/index.ts

@@ -1,5 +1,3 @@
-export { k$ } from './k$';
-
 export * from './Observable';
 export { Subject } from './Subject';
 export { BehaviorSubject } from './BehaviorSubject';