Upgrade RxJS to 7

examples/search_examples/public/search/app.tsx

@@ -32,6 +32,7 @@ import {
   EuiTabbedContentTab,
 } from '@elastic/eui';
 
+import { lastValueFrom } from 'rxjs';
 import { CoreStart } from '../../../../src/core/public';
 import { mountReactNode } from '../../../../src/core/public/utils';
 import { NavigationPublicPluginStart } from '../../../../src/plugins/navigation/public';
@@ -303,9 +304,9 @@ export const SearchExamplesApp = ({
       const abortController = new AbortController();
       setAbortController(abortController);
       setIsLoading(true);
-      const { rawResponse: res } = await searchSource
-        .fetch$({ abortSignal: abortController.signal })
-        .toPromise();
+      const { rawResponse: res } = await lastValueFrom(
+        searchSource.fetch$({ abortSignal: abortController.signal })
+      );
       setRawResponse(res);
 
       const message = <EuiText>Searched {res.hits.total} documents.</EuiText>;

examples/search_examples/public/search_sessions/app.tsx

@@ -30,7 +30,7 @@ import {
   EuiTitle,
 } from '@elastic/eui';
 import { catchError, map, tap } from 'rxjs/operators';
-import { of } from 'rxjs';
+import { firstValueFrom, of } from 'rxjs';
 
 import { CoreStart } from '../../../../src/core/public';
 import { mountReactNode } from '../../../../src/core/public/utils';
@@ -690,9 +690,8 @@ function doSearch(
   const startTs = performance.now();
 
   // Submit the search request using the `data.search` service.
-  return data.search
-    .search(req, { sessionId })
-    .pipe(
+  return firstValueFrom(
+    data.search.search(req, { sessionId }).pipe(
       tap((res) => {
         if (isCompleteResponse(res)) {
           const avgResult: number | undefined = res.rawResponse.aggregations
@@ -721,7 +720,7 @@ function doSearch(
         return of({ request: req, response: e });
       })
     )
-    .toPromise();
+  );
 }
 
 function getNumeric(fields?: DataViewField[]) {

package.json

@@ -376,7 +376,7 @@
     "redux-actions": "^2.6.5",
     "redux-devtools-extension": "^2.13.8",
     "redux-logger": "^3.0.6",
-    "redux-observable": "^1.2.0",
+    "redux-observable": "2.0.0",
     "redux-saga": "^1.1.3",
     "redux-thunk": "^2.3.0",
     "redux-thunks": "^1.0.0",
@@ -387,7 +387,7 @@
     "reselect": "^4.0.0",
     "resize-observer-polyfill": "^1.5.1",
     "rison-node": "1.0.2",
-    "rxjs": "^6.5.5",
+    "rxjs": "^7.5.5",
     "safe-squel": "^5.12.5",
     "seedrandom": "^3.0.5",
     "semver": "^7.3.2",

packages/kbn-cli-dev-mode/src/dev_server.ts

@@ -6,8 +6,6 @@
  * Side Public License, v 1.
  */
 
-import { EventEmitter } from 'events';
-
 import * as Rx from 'rxjs';
 import {
   map,
@@ -60,9 +58,9 @@ export class DevServer {
     this.script = options.script;
     this.argv = options.argv;
     this.gracefulTimeout = options.gracefulTimeout;
-    this.processExit$ = options.processExit$ ?? Rx.fromEvent(process as EventEmitter, 'exit');
-    this.sigint$ = options.sigint$ ?? Rx.fromEvent(process as EventEmitter, 'SIGINT');
-    this.sigterm$ = options.sigterm$ ?? Rx.fromEvent(process as EventEmitter, 'SIGTERM');
+    this.processExit$ = options.processExit$ ?? Rx.fromEvent<void>(process, 'exit');
+    this.sigint$ = options.sigint$ ?? Rx.fromEvent<void>(process, 'SIGINT');
+    this.sigterm$ = options.sigterm$ ?? Rx.fromEvent<void>(process, 'SIGTERM');
     this.mapLogLine = options.mapLogLine;
   }
 
@@ -117,7 +115,7 @@ export class DevServer {
    */
   run$ = new Rx.Observable<void>((subscriber) => {
     // listen for SIGINT and forward to process if it's running, otherwise unsub
-    const gracefulShutdown$ = new Rx.Subject();
+    const gracefulShutdown$ = new Rx.Subject<void>();
     subscriber.add(
       this.sigint$
         .pipe(

packages/kbn-config/src/config_service.ts

@@ -9,8 +9,8 @@
 import type { PublicMethodsOf } from '@kbn/utility-types';
 import { Type } from '@kbn/config-schema';
 import { isEqual } from 'lodash';
-import { BehaviorSubject, combineLatest, Observable } from 'rxjs';
-import { distinctUntilChanged, first, map, shareReplay, take, tap } from 'rxjs/operators';
+import { BehaviorSubject, combineLatest, firstValueFrom, Observable } from 'rxjs';
+import { distinctUntilChanged, first, map, shareReplay, tap } from 'rxjs/operators';
 import { Logger, LoggerFactory } from '@kbn/logging';
 
 import { Config, ConfigPath, Env } from '.';
@@ -170,7 +170,7 @@ export class ConfigService {
     const namespace = pathToString(path);
     const hasSchema = this.schemas.has(namespace);
 
-    const config = await this.config$.pipe(first()).toPromise();
+    const config = await firstValueFrom(this.config$);
     if (!hasSchema && config.has(path)) {
       // Throw if there is no schema, but a config exists at the path.
       throw new Error(`No validation schema has been defined for [${namespace}]`);
@@ -195,13 +195,13 @@ export class ConfigService {
   }
 
   public async getUnusedPaths() {
-    const config = await this.config$.pipe(first()).toPromise();
+    const config = await firstValueFrom(this.config$);
     const handledPaths = [...this.handledPaths.values()].map(pathToString);
     return config.getFlattenedPaths().filter((path) => !isPathHandled(path, handledPaths));
   }
 
   public async getUsedPaths() {
-    const config = await this.config$.pipe(first()).toPromise();
+    const config = await firstValueFrom(this.config$);
     const handledPaths = [...this.handledPaths.values()].map(pathToString);
     return config.getFlattenedPaths().filter((path) => isPathHandled(path, handledPaths));
   }
@@ -211,8 +211,8 @@ export class ConfigService {
   }
 
   private async logDeprecation() {
-    const rawConfig = await this.rawConfigProvider.getConfig$().pipe(take(1)).toPromise();
-    const deprecations = await this.deprecations.pipe(take(1)).toPromise();
+    const rawConfig = await firstValueFrom(this.rawConfigProvider.getConfig$());
+    const deprecations = await firstValueFrom(this.deprecations);
     const deprecationMessages: string[] = [];
     const createAddDeprecation = (domainId: string) => (context: DeprecatedConfigDetails) => {
       if (!context.silent) {

packages/kbn-config/src/raw/raw_config_service.test.ts

@@ -8,7 +8,7 @@
 
 import { mockGetConfigFromFiles } from './raw_config_service.test.mocks';
 
-import { first } from 'rxjs/operators';
+import { firstValueFrom } from 'rxjs';
 import { RawConfigService } from './raw_config_service';
 
 const configFile = '/config/kibana.yml';
@@ -72,7 +72,7 @@ test('returns config at path as observable', async () => {
 
   configService.loadConfig();
 
-  const exampleConfig = await configService.getConfig$().pipe(first()).toPromise();
+  const exampleConfig = await firstValueFrom(configService.getConfig$());
 
   expect(exampleConfig.key).toEqual('value');
   expect(Object.keys(exampleConfig)).toEqual(['key']);

packages/kbn-dev-utils/src/tooling_log/tooling_log.test.ts

@@ -14,6 +14,7 @@ import { Writer } from './writer';
 import { ToolingLogTextWriter } from './tooling_log_text_writer';
 import { ToolingLogCollectingWriter } from './tooling_log_collecting_writer';
 import { createStripAnsiSerializer } from '../serializers/strip_ansi_serializer';
+import { lastValueFrom } from 'rxjs';
 
 expect.addSnapshotSerializer(createStripAnsiSerializer());
 
@@ -146,8 +147,8 @@ describe('#getWritten$()', () => {
     const log = new ToolingLog();
     log.setWriters(writers);
 
-    const done$ = new Rx.Subject();
-    const promise = log.getWritten$().pipe(takeUntil(done$), toArray()).toPromise();
+    const done$ = new Rx.Subject<void>();
+    const promise = lastValueFrom(log.getWritten$().pipe(takeUntil(done$), toArray()));
 
     log.debug('foo');
     log.info('bar');

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

@@ -70,12 +70,13 @@ describe('maybeMap()', () => {
 });
 
 describe('debounceTimeBuffer()', () => {
+  const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
   beforeEach(() => {
-    jest.useFakeTimers();
+    jest.useRealTimers();
   });
 
   afterEach(() => {
-    jest.useRealTimers();
+    jest.useFakeTimers();
   });
 
   it('buffers items until there is n milliseconds of silence, then flushes buffer to stream', async () => {
@@ -91,18 +92,18 @@ describe('debounceTimeBuffer()', () => {
     foo$.next(1);
     expect(dest.getValue()).toBe(undefined);
 
-    // only wait 99 milliseconds before sending the next value
-    jest.advanceTimersByTime(99);
+    // only wait 50 milliseconds before sending the next value
+    await delay(50);
     foo$.next(1);
     expect(dest.getValue()).toBe(undefined);
 
     // only wait 99 milliseconds before sending the next value
-    jest.advanceTimersByTime(99);
+    await delay(99);
     foo$.next(1);
     expect(dest.getValue()).toBe(undefined);
 
     // send the next value after 100 milliseconds and observe that it was forwarded
-    jest.advanceTimersByTime(100);
+    await delay(500);
     foo$.next(1);
     expect(dest.getValue()).toBe(3);
 

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

@@ -8,6 +8,7 @@
 
 import * as Rx from 'rxjs';
 import { map } from 'rxjs/operators';
+import { fakeSchedulers } from 'rxjs-marbles/jest';
 import ActualWatchpack from 'watchpack';
 import { lastValueFrom } from '@kbn/std';
 
@@ -68,66 +69,69 @@ afterEach(async () => {
 it('notifies of changes and completes once all bundles have changed', async () => {
   expect.assertions(18);
 
-  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) {
-          expect(event).toHaveProperty('type', 'changes detected');
-          return;
-        }
-
-        expect(event).toHaveProperty('type', 'changes');
-        // to teach TS what we're doing
-        if (event.type !== 'changes') {
-          return;
-        }
-
-        // first we change foo and bar, after 1 second that change comes though
-        if (i === 1) {
-          expect(event.bundles).toHaveLength(2);
-          const [bar, foo] = event.bundles.sort(ascending((b) => b.id));
-          expect(bar).toHaveProperty('id', 'bar');
-          expect(foo).toHaveProperty('id', 'foo');
-        }
-
-        // next we change just the baz package and it's represented on its own
-        if (i === 3) {
-          expect(event.bundles).toHaveLength(1);
-          expect(event.bundles[0]).toHaveProperty('id', 'baz');
-        }
-
-        // finally we change box and car together
-        if (i === 5) {
-          expect(event.bundles).toHaveLength(2);
-          const [bar, foo] = event.bundles.sort(ascending((b) => b.id));
-          expect(bar).toHaveProperty('id', 'box');
-          expect(foo).toHaveProperty('id', 'car');
-        }
-      })
-    )
-  );
-
-  expect(MockWatchPack.mock.instances).toHaveLength(1);
-  const [watcher] = MockWatchPack.mock.instances as any as Array<jest.Mocked<ActualWatchpack>>;
-  expect(watcher.on).toHaveBeenCalledTimes(1);
-  expect(watcher.on).toHaveBeenCalledWith('change', expect.any(Function));
-  const [, changeListener] = watcher.on.mock.calls[0];
-
-  // foo and bar are changes without 1sec so they are batched
-  changeListener(bundleEntryPath(FOO_BUNDLE), 'modified');
-  jest.advanceTimersByTime(900);
-  changeListener(bundleEntryPath(BAR_BUNDLE), 'modified');
-  jest.advanceTimersByTime(1000);
-
-  // baz is the only change in 1sec so it is on its own
-  changeListener(bundleEntryPath(BAZ_BUNDLE), 'modified');
-  jest.advanceTimersByTime(1000);
-
-  // finish by changing box and car
-  changeListener(bundleEntryPath(BOX_BUNDLE), 'deleted');
-  changeListener(bundleEntryPath(CAR_BUNDLE), 'deleted');
-  jest.advanceTimersByTime(1000);
-
+  const promise = fakeSchedulers((advance) => {
+    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) {
+            expect(event).toHaveProperty('type', 'changes detected');
+            return;
+          }
+
+          expect(event).toHaveProperty('type', 'changes');
+          // to teach TS what we're doing
+          if (event.type !== 'changes') {
+            return;
+          }
+
+          // first we change foo and bar, after 1 second that change comes though
+          if (i === 1) {
+            expect(event.bundles).toHaveLength(2);
+            const [bar, foo] = event.bundles.sort(ascending((b) => b.id));
+            expect(bar).toHaveProperty('id', 'bar');
+            expect(foo).toHaveProperty('id', 'foo');
+          }
+
+          // next we change just the baz package and it's represented on its own
+          if (i === 3) {
+            expect(event.bundles).toHaveLength(1);
+            expect(event.bundles[0]).toHaveProperty('id', 'baz');
+          }
+
+          // finally we change box and car together
+          if (i === 5) {
+            expect(event.bundles).toHaveLength(2);
+            const [bar, foo] = event.bundles.sort(ascending((b) => b.id));
+            expect(bar).toHaveProperty('id', 'box');
+            expect(foo).toHaveProperty('id', 'car');
+          }
+        })
+      )
+    );
+
+    expect(MockWatchPack.mock.instances).toHaveLength(1);
+    const [watcher] = MockWatchPack.mock.instances as any as Array<jest.Mocked<ActualWatchpack>>;
+    expect(watcher.on).toHaveBeenCalledTimes(1);
+    expect(watcher.on).toHaveBeenCalledWith('change', expect.any(Function));
+    const [, changeListener] = watcher.on.mock.calls[0];
+
+    // foo and bar are changes without 1sec so they are batched
+    changeListener(bundleEntryPath(FOO_BUNDLE), 'modified');
+    advance(900);
+    changeListener(bundleEntryPath(BAR_BUNDLE), 'modified');
+    advance(1000);
+
+    // baz is the only change in 1sec so it is on its own
+    changeListener(bundleEntryPath(BAZ_BUNDLE), 'modified');
+    advance(1000);
+
+    // finish by changing box and car
+    changeListener(bundleEntryPath(BOX_BUNDLE), 'deleted');
+    changeListener(bundleEntryPath(CAR_BUNDLE), 'deleted');
+    advance(1000);
+
+    return _promise;
+  })();
   await expect(promise).resolves.toEqual(undefined);
 });

packages/kbn-optimizer/src/optimizer/observe_worker.ts

@@ -159,7 +159,7 @@ export function observeWorker(
     let lastMsg: WorkerMsg;
     const worker$: Rx.Observable<WorkerMsg | WorkerStatus> = Rx.merge(
       Rx.of({
-        type: 'worker started',
+        type: 'worker started' as const,
         bundles,
       }),
       // TypeScript note: As long as the proc stdio[1] is 'pipe', then stdout will not be null

packages/kbn-optimizer/src/worker/run_compilers.ts

@@ -27,7 +27,7 @@ const observeCompiler = (
   compiler: webpack.Compiler
 ): Rx.Observable<CompilerMsg> => {
   const compilerMsgs = new CompilerMsgs(bundle.id);
-  const done$ = new Rx.Subject();
+  const done$ = new Rx.Subject<void>();
   const { beforeRun, watchRun, done } = compiler.hooks;
 
   /**