[kbn/std] add observable helpers to aid with rxjs 7 upgrade

packages/kbn-optimizer/README.md

@@ -84,9 +84,9 @@ const config = OptimizerConfig.create({
   dist: true
 });
 
-await runOptimizer(config)
-  .pipe(logOptimizerState(log, config))
-  .toPromise();
+await lastValueFrom(
+  runOptimizer(config).pipe(logOptimizerState(log, config))
+);
 ```
 
 This is essentially what we're doing in [`script/build_kibana_platform_plugins`][Cli] and the new [build system task][BuildTask].

packages/kbn-optimizer/package.json

@@ -14,6 +14,7 @@
     "@babel/core": "^7.11.6",
     "@kbn/babel-preset": "1.0.0",
     "@kbn/dev-utils": "1.0.0",
+    "@kbn/std": "1.0.0",
     "@kbn/ui-shared-deps": "1.0.0",
     "autoprefixer": "^9.7.4",
     "babel-loader": "^8.0.6",

packages/kbn-optimizer/src/cli.ts

@@ -22,6 +22,7 @@ import 'source-map-support/register';
 import Path from 'path';
 
 import { REPO_ROOT } from '@kbn/utils';
+import { lastValueFrom } from '@kbn/std';
 import { run, createFlagError, CiStatsReporter } from '@kbn/dev-utils';
 
 import { logOptimizerState } from './log_optimizer_state';
@@ -136,7 +137,7 @@ run(
       update$ = update$.pipe(reportOptimizerStats(reporter, config, log));
     }
 
-    await update$.pipe(logOptimizerState(log, config)).toPromise();
+    await lastValueFrom(update$.pipe(logOptimizerState(log, config)));
 
     if (updateLimits) {
       updateBundleLimits(log, config);

packages/kbn-optimizer/src/common/event_stream_helpers.test.ts

@@ -18,20 +18,21 @@
  */
 
 import * as Rx from 'rxjs';
-import { toArray, take } from 'rxjs/operators';
+import { take } from 'rxjs/operators';
+import { allValuesFrom } from '@kbn/std';
 
 import { summarizeEventStream } from './event_stream_helpers';
 
 it('emits each state with each event, ignoring events when summarizer returns undefined', async () => {
   const event$ = Rx.of(1, 2, 3, 4, 5);
   const initial = 0;
-  const values = await summarizeEventStream(event$, initial, (state, event) => {
-    if (event % 2) {
-      return state + event;
-    }
-  })
-    .pipe(toArray())
-    .toPromise();
+  const values = await allValuesFrom(
+    summarizeEventStream(event$, initial, (state, event) => {
+      if (event % 2) {
+        return state + event;
+      }
+    })
+  );
 
   expect(values).toMatchInlineSnapshot(`
     Array [
@@ -57,15 +58,15 @@ it('emits each state with each event, ignoring events when summarizer returns un
 it('interleaves injected events when source is synchronous', async () => {
   const event$ = Rx.of(1, 7);
   const initial = 0;
-  const values = await summarizeEventStream(event$, initial, (state, event, injectEvent) => {
-    if (event < 5) {
-      injectEvent(event + 2);
-    }
+  const values = await allValuesFrom(
+    summarizeEventStream(event$, initial, (state, event, injectEvent) => {
+      if (event < 5) {
+        injectEvent(event + 2);
+      }
 
-    return state + event;
-  })
-    .pipe(toArray())
-    .toPromise();
+      return state + event;
+    })
+  );
 
   expect(values).toMatchInlineSnapshot(`
     Array [
@@ -95,15 +96,15 @@ it('interleaves injected events when source is synchronous', async () => {
 it('interleaves injected events when source is asynchronous', async () => {
   const event$ = Rx.of(1, 7, Rx.asyncScheduler);
   const initial = 0;
-  const values = await summarizeEventStream(event$, initial, (state, event, injectEvent) => {
-    if (event < 5) {
-      injectEvent(event + 2);
-    }
+  const values = await allValuesFrom(
+    summarizeEventStream(event$, initial, (state, event, injectEvent) => {
+      if (event < 5) {
+        injectEvent(event + 2);
+      }
 
-    return state + event;
-  })
-    .pipe(toArray())
-    .toPromise();
+      return state + event;
+    })
+  );
 
   expect(values).toMatchInlineSnapshot(`
     Array [
@@ -133,17 +134,17 @@ it('interleaves injected events when source is asynchronous', async () => {
 it('interleaves mulitple injected events in order', async () => {
   const event$ = Rx.of(1);
   const initial = 0;
-  const values = await summarizeEventStream(event$, initial, (state, event, injectEvent) => {
-    if (event < 10) {
-      injectEvent(10);
-      injectEvent(20);
-      injectEvent(30);
-    }
-
-    return state + event;
-  })
-    .pipe(toArray())
-    .toPromise();
+  const values = await allValuesFrom(
+    summarizeEventStream(event$, initial, (state, event, injectEvent) => {
+      if (event < 10) {
+        injectEvent(10);
+        injectEvent(20);
+        injectEvent(30);
+      }
+
+      return state + event;
+    })
+  );
 
   expect(values).toMatchInlineSnapshot(`
     Array [
@@ -179,9 +180,9 @@ it('stops an infinite stream when unsubscribed', async () => {
     return prev + event;
   });
 
-  const values = await summarizeEventStream(event$, initial, summarize)
-    .pipe(take(11), toArray())
-    .toPromise();
+  const values = await allValuesFrom(
+    summarizeEventStream(event$, initial, summarize).pipe(take(11))
+  );
 
   expect(values).toMatchInlineSnapshot(`
     Array [

packages/kbn-optimizer/src/common/rxjs_helpers.test.ts

@@ -19,6 +19,7 @@
 
 import * as Rx from 'rxjs';
 import { toArray, map } from 'rxjs/operators';
+import { lastValueFrom } from '@kbn/std';
 
 import { pipeClosure, debounceTimeBuffer, maybeMap, maybe } from './rxjs_helpers';
 
@@ -36,21 +37,21 @@ describe('pipeClosure()', () => {
       toArray()
     );
 
-    await expect(foo$.toPromise()).resolves.toMatchInlineSnapshot(`
+    await expect(lastValueFrom(foo$)).resolves.toMatchInlineSnapshot(`
             Array [
               1,
               2,
               3,
             ]
           `);
-    await expect(foo$.toPromise()).resolves.toMatchInlineSnapshot(`
+    await expect(lastValueFrom(foo$)).resolves.toMatchInlineSnapshot(`
             Array [
               2,
               4,
               6,
             ]
           `);
-    await expect(foo$.toPromise()).resolves.toMatchInlineSnapshot(`
+    await expect(lastValueFrom(foo$)).resolves.toMatchInlineSnapshot(`
             Array [
               3,
               6,
@@ -64,7 +65,7 @@ describe('maybe()', () => {
   it('filters out undefined values from the stream', async () => {
     const foo$ = Rx.of(1, undefined, 2, undefined, 3).pipe(maybe(), toArray());
 
-    await expect(foo$.toPromise()).resolves.toEqual([1, 2, 3]);
+    await expect(lastValueFrom(foo$)).resolves.toEqual([1, 2, 3]);
   });
 });
 
@@ -75,7 +76,7 @@ describe('maybeMap()', () => {
       toArray()
     );
 
-    await expect(foo$.toPromise()).resolves.toEqual([1, 3, 5]);
+    await expect(lastValueFrom(foo$)).resolves.toEqual([1, 3, 5]);
   });
 });
 

packages/kbn-optimizer/src/integration_tests/basic_optimization.test.ts

@@ -24,7 +24,7 @@ import { inspect } from 'util';
 
 import cpy from 'cpy';
 import del from 'del';
-import { toArray, tap, filter } from 'rxjs/operators';
+import { tap, filter } from 'rxjs/operators';
 import { REPO_ROOT } from '@kbn/utils';
 import { ToolingLog } from '@kbn/dev-utils';
 import {
@@ -34,6 +34,7 @@ import {
   logOptimizerState,
   readLimits,
 } from '@kbn/optimizer';
+import { allValuesFrom } from '@kbn/std';
 
 const TMP_DIR = Path.resolve(__dirname, '../__fixtures__/__tmp__');
 const MOCK_REPO_SRC = Path.resolve(__dirname, '../__fixtures__/mock_repo');
@@ -83,13 +84,12 @@ it('builds expected bundles, saves bundle counts to metadata', async () => {
 
   expect(config).toMatchSnapshot('OptimizerConfig');
 
-  const msgs = await runOptimizer(config)
-    .pipe(
+  const msgs = await allValuesFrom(
+    runOptimizer(config).pipe(
       logOptimizerState(log, config),
-      filter((x) => x.event?.type !== 'worker stdio'),
-      toArray()
+      filter((x) => x.event?.type !== 'worker stdio')
     )
-    .toPromise();
+  );
 
   const assert = (statement: string, truth: boolean, altStates?: OptimizerUpdate[]) => {
     if (!truth) {
@@ -208,17 +208,16 @@ it('uses cache on second run and exist cleanly', async () => {
     dist: false,
   });
 
-  const msgs = await runOptimizer(config)
-    .pipe(
+  const msgs = await allValuesFrom(
+    runOptimizer(config).pipe(
       tap((state) => {
         if (state.event?.type === 'worker stdio') {
           // eslint-disable-next-line no-console
           console.log('worker', state.event.stream, state.event.line);
         }
-      }),
-      toArray()
+      })
     )
-    .toPromise();
+  );
 
   expect(msgs.map((m) => m.state.phase)).toMatchInlineSnapshot(`
     Array [
@@ -240,7 +239,7 @@ it('prepares assets for distribution', async () => {
     dist: true,
   });
 
-  await runOptimizer(config).pipe(logOptimizerState(log, config), toArray()).toPromise();
+  await allValuesFrom(runOptimizer(config).pipe(logOptimizerState(log, config)));
 
   expectFileMatchesSnapshotWithCompression('plugins/foo/target/public/foo.plugin.js', 'foo bundle');
   expectFileMatchesSnapshotWithCompression(

packages/kbn-optimizer/src/integration_tests/bundle_cache.test.ts

@@ -21,8 +21,8 @@ import Path from 'path';
 
 import cpy from 'cpy';
 import del from 'del';
-import { toArray } from 'rxjs/operators';
 import { createAbsolutePathSerializer } from '@kbn/dev-utils';
+import { allValuesFrom } from '@kbn/std';
 
 import { getMtimes } from '../optimizer/get_mtimes';
 import { OptimizerConfig } from '../optimizer/optimizer_config';
@@ -78,9 +78,7 @@ it('emits "bundle cached" event when everything is updated', async () => {
     bundleRefExportIds: [],
   });
 
-  const cacheEvents = await getBundleCacheEvent$(config, optimizerCacheKey)
-    .pipe(toArray())
-    .toPromise();
+  const cacheEvents = await allValuesFrom(getBundleCacheEvent$(config, optimizerCacheKey));
 
   expect(cacheEvents).toMatchInlineSnapshot(`
     Array [
@@ -119,9 +117,7 @@ it('emits "bundle not cached" event when cacheKey is up to date but caching is d
     bundleRefExportIds: [],
   });
 
-  const cacheEvents = await getBundleCacheEvent$(config, optimizerCacheKey)
-    .pipe(toArray())
-    .toPromise();
+  const cacheEvents = await allValuesFrom(getBundleCacheEvent$(config, optimizerCacheKey));
 
   expect(cacheEvents).toMatchInlineSnapshot(`
     Array [
@@ -160,9 +156,7 @@ it('emits "bundle not cached" event when optimizerCacheKey is missing', async ()
     bundleRefExportIds: [],
   });
 
-  const cacheEvents = await getBundleCacheEvent$(config, optimizerCacheKey)
-    .pipe(toArray())
-    .toPromise();
+  const cacheEvents = await allValuesFrom(getBundleCacheEvent$(config, optimizerCacheKey));
 
   expect(cacheEvents).toMatchInlineSnapshot(`
     Array [
@@ -201,9 +195,7 @@ it('emits "bundle not cached" event when optimizerCacheKey is outdated, includes
     bundleRefExportIds: [],
   });
 
-  const cacheEvents = await getBundleCacheEvent$(config, optimizerCacheKey)
-    .pipe(toArray())
-    .toPromise();
+  const cacheEvents = await allValuesFrom(getBundleCacheEvent$(config, optimizerCacheKey));
 
   expect(cacheEvents).toMatchInlineSnapshot(`
     Array [
@@ -247,9 +239,7 @@ it('emits "bundle not cached" event when bundleRefExportIds is outdated, include
     bundleRefExportIds: ['plugin/bar/public'],
   });
 
-  const cacheEvents = await getBundleCacheEvent$(config, optimizerCacheKey)
-    .pipe(toArray())
-    .toPromise();
+  const cacheEvents = await allValuesFrom(getBundleCacheEvent$(config, optimizerCacheKey));
 
   expect(cacheEvents).toMatchInlineSnapshot(`
     Array [
@@ -292,9 +282,7 @@ it('emits "bundle not cached" event when cacheKey is missing', async () => {
     bundleRefExportIds: [],
   });
 
-  const cacheEvents = await getBundleCacheEvent$(config, optimizerCacheKey)
-    .pipe(toArray())
-    .toPromise();
+  const cacheEvents = await allValuesFrom(getBundleCacheEvent$(config, optimizerCacheKey));
 
   expect(cacheEvents).toMatchInlineSnapshot(`
     Array [
@@ -333,9 +321,7 @@ it('emits "bundle not cached" event when cacheKey is outdated', async () => {
 
   jest.spyOn(bundle, 'createCacheKey').mockImplementation(() => 'new');
 
-  const cacheEvents = await getBundleCacheEvent$(config, optimizerCacheKey)
-    .pipe(toArray())
-    .toPromise();
+  const cacheEvents = await allValuesFrom(getBundleCacheEvent$(config, optimizerCacheKey));
 
   expect(cacheEvents).toMatchInlineSnapshot(`
     Array [

packages/kbn-optimizer/src/integration_tests/watch_bundles_for_changes.test.ts

@@ -20,6 +20,7 @@
 import * as Rx from 'rxjs';
 import { map } from 'rxjs/operators';
 import ActualWatchpack from 'watchpack';
+import { lastValueFrom } from '@kbn/std';
 
 import { Bundle, ascending } from '../common';
 import { watchBundlesForChanges$ } from '../optimizer/watch_bundles_for_changes';
@@ -78,8 +79,8 @@ afterEach(async () => {
 it('notifies of changes and completes once all bundles have changed', async () => {
   expect.assertions(18);
 
-  const promise = watchBundlesForChanges$(bundleCacheEvent$, Date.now())
-    .pipe(
+  const promise = lastValueFrom(
+    watchBundlesForChanges$(bundleCacheEvent$, Date.now()).pipe(
       map((event, i) => {
         // each time we trigger a change event we get a 'changed detected' event
         if (i === 0 || i === 2 || i === 4 || i === 6) {
@@ -116,7 +117,7 @@ it('notifies of changes and completes once all bundles have changed', async () =
         }
       })
     )
-    .toPromise();
+  );
 
   expect(MockWatchPack.mock.instances).toHaveLength(1);
   const [watcher] = (MockWatchPack.mock.instances as any) as Array<jest.Mocked<ActualWatchpack>>;