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';
@@ -306,9 +307,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 { lastValueFrom, of } from 'rxjs';
 
 import { CoreStart } from '../../../../src/core/public';
 import { mountReactNode } from '../../../../src/core/public/utils';
@@ -693,9 +693,8 @@ function doSearch(
   const startTs = performance.now();
 
   // Submit the search request using the `data.search` service.
-  return data.search
-    .search(req, { sessionId })
-    .pipe(
+  return lastValueFrom(
+    data.search.search(req, { sessionId }).pipe(
       tap((res) => {
         if (isCompleteResponse(res)) {
           const avgResult: number | undefined = res.rawResponse.aggregations
@@ -724,7 +723,7 @@ function doSearch(
         return of({ request: req, response: e });
       })
     )
-    .toPromise();
+  );
 }
 
 function getNumeric(fields?: DataViewField[]) {

package.json

@@ -383,7 +383,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",
@@ -394,7 +394,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/elastic-analytics/src/analytics_client/analytics_client.test.ts

@@ -8,7 +8,7 @@
 
 // eslint-disable-next-line max-classes-per-file
 import type { Observable } from 'rxjs';
-import { BehaviorSubject, Subject } from 'rxjs';
+import { BehaviorSubject, lastValueFrom, Subject } from 'rxjs';
 import type { MockedLogger } from '@kbn/logging-mocks';
 import { loggerMock } from '@kbn/logging-mocks';
 import { AnalyticsClient } from './analytics_client';
@@ -113,14 +113,14 @@ describe('AnalyticsClient', () => {
         },
       });
 
-      // eslint-disable-next-line dot-notation
-      const internalQueuePromise = analyticsClient['internalEventQueue$']
-        .pipe(take(3), toArray())
-        .toPromise();
+      const internalQueuePromise = lastValueFrom(
+        // eslint-disable-next-line dot-notation
+        analyticsClient['internalEventQueue$'].pipe(take(3), toArray())
+      );
 
-      const telemetryCounterPromise = analyticsClient.telemetryCounter$
-        .pipe(take(3), toArray())
-        .toPromise();
+      const telemetryCounterPromise = lastValueFrom(
+        analyticsClient.telemetryCounter$.pipe(take(3), toArray())
+      );
 
       analyticsClient.reportEvent('testEvent', { a_field: 'a' });
       analyticsClient.reportEvent('testEvent', { a_field: 'b' });
@@ -259,7 +259,7 @@ describe('AnalyticsClient', () => {
       // Typescript also helps with the config type inference <3
       analyticsClient.registerShipper(MockedShipper, { telemetryCounter$: mockTelemetryCounter$ });
 
-      const counterEventPromise = analyticsClient.telemetryCounter$.pipe(take(1)).toPromise();
+      const counterEventPromise = lastValueFrom(analyticsClient.telemetryCounter$.pipe(take(1)));
 
       const counter: TelemetryCounter = {
         type: TelemetryCounterType.succeeded,
@@ -383,7 +383,7 @@ describe('AnalyticsClient', () => {
         context$,
       });
 
-      const globalContextPromise = globalContext$.pipe(take(2), toArray()).toPromise();
+      const globalContextPromise = lastValueFrom(globalContext$.pipe(take(2), toArray()));
       context$.next({ a_field: true });
       await expect(globalContextPromise).resolves.toEqual([
         {}, // Original empty state
@@ -392,7 +392,7 @@ describe('AnalyticsClient', () => {
     });
 
     test('It does not break if context emits `undefined`', async () => {
-      const context$ = new Subject<{ a_field: boolean }>();
+      const context$ = new Subject<{ a_field: boolean } | undefined | void>();
       analyticsClient.registerContextProvider({
         name: 'contextProviderA',
         schema: {
@@ -406,7 +406,7 @@ describe('AnalyticsClient', () => {
         context$,
       });
 
-      const globalContextPromise = globalContext$.pipe(take(3), toArray()).toPromise();
+      const globalContextPromise = lastValueFrom(globalContext$.pipe(take(3), toArray()));
       context$.next();
       context$.next(undefined);
       await expect(globalContextPromise).resolves.toEqual([
@@ -431,7 +431,7 @@ describe('AnalyticsClient', () => {
         context$,
       });
 
-      const globalContextPromise = globalContext$.pipe(take(1), toArray()).toPromise();
+      const globalContextPromise = lastValueFrom(globalContext$.pipe(take(1), toArray()));
       await expect(globalContextPromise).resolves.toEqual([
         { a_field: true }, // No original empty state
       ]);
@@ -473,7 +473,7 @@ describe('AnalyticsClient', () => {
         context$: contextB$,
       });
 
-      const globalContextPromise = globalContext$.pipe(take(6), toArray()).toPromise();
+      const globalContextPromise = lastValueFrom(globalContext$.pipe(take(6), toArray()));
       contextA$.next({ a_field: true });
       contextB$.next({ b_field: 1 });
       contextB$.next({ a_field: false, b_field: 1 });
@@ -512,7 +512,7 @@ describe('AnalyticsClient', () => {
         context$,
       });
 
-      const globalContextPromise = globalContext$.pipe(take(6), toArray()).toPromise();
+      const globalContextPromise = lastValueFrom(globalContext$.pipe(take(6), toArray()));
       context$.next({ b_field: 1 });
       context$.next({ a_field: false, b_field: 1 });
       context$.next({ a_field: true, b_field: 1 });
@@ -582,7 +582,7 @@ describe('AnalyticsClient', () => {
         context$,
       });
 
-      const globalContextPromise = globalContext$.pipe(take(4), toArray()).toPromise();
+      const globalContextPromise = lastValueFrom(globalContext$.pipe(take(4), toArray()));
       context$.next({ a_field: true });
       // The size of the registry grows on the first emission
       expect(contextProvidersRegistry.size).toBe(1);
@@ -736,11 +736,11 @@ describe('AnalyticsClient', () => {
       // eslint-disable-next-line dot-notation
       const internalEventQueue$ = analyticsClient['internalEventQueue$'];
 
-      const internalQueuePromise = internalEventQueue$.pipe(take(2), toArray()).toPromise();
+      const internalQueuePromise = lastValueFrom(internalEventQueue$.pipe(take(2), toArray()));
 
-      const telemetryCounterPromise = analyticsClient.telemetryCounter$
-        .pipe(take(2), toArray())
-        .toPromise();
+      const telemetryCounterPromise = lastValueFrom(
+        analyticsClient.telemetryCounter$.pipe(take(2), toArray())
+      );
 
       analyticsClient.reportEvent('event-type-a', { a_field: 'a' });
       analyticsClient.reportEvent('event-type-b', { b_field: 100 });
@@ -781,9 +781,9 @@ describe('AnalyticsClient', () => {
     });
 
     test('Sends events from the internal queue when there are shippers and an opt-in response is true', async () => {
-      const telemetryCounterPromise = analyticsClient.telemetryCounter$
-        .pipe(take(3 + 2), toArray()) // Waiting for 3 enqueued + 2 batch-shipped events
-        .toPromise();
+      const telemetryCounterPromise = lastValueFrom(
+        analyticsClient.telemetryCounter$.pipe(take(3 + 2), toArray()) // Waiting for 3 enqueued + 2 batch-shipped events
+      );
 
       // Send multiple events of 1 type to test the grouping logic as well
       analyticsClient.reportEvent('event-type-a', { a_field: 'a' });
@@ -862,9 +862,9 @@ describe('AnalyticsClient', () => {
     });
 
     test('Discards events from the internal queue when there are shippers and an opt-in response is false', async () => {
-      const telemetryCounterPromise = analyticsClient.telemetryCounter$
-        .pipe(take(3), toArray()) // Waiting for 3 enqueued
-        .toPromise();
+      const telemetryCounterPromise = lastValueFrom(
+        analyticsClient.telemetryCounter$.pipe(take(3), toArray()) // Waiting for 3 enqueued
+      );
 
       // Send multiple events of 1 type to test the grouping logic as well
       analyticsClient.reportEvent('event-type-a', { a_field: 'a' });
@@ -904,9 +904,9 @@ describe('AnalyticsClient', () => {
     });
 
     test('Discards only one type of the enqueued events based on event_type config', async () => {
-      const telemetryCounterPromise = analyticsClient.telemetryCounter$
-        .pipe(take(3 + 1), toArray()) // Waiting for 3 enqueued + 1 batch-shipped events
-        .toPromise();
+      const telemetryCounterPromise = lastValueFrom(
+        analyticsClient.telemetryCounter$.pipe(take(3 + 1), toArray()) // Waiting for 3 enqueued + 1 batch-shipped events
+      );
 
       // Send multiple events of 1 type to test the grouping logic as well
       analyticsClient.reportEvent('event-type-a', { a_field: 'a' });
@@ -965,9 +965,9 @@ describe('AnalyticsClient', () => {
     });
 
     test('Discards the event at the shipper level (for a specific event)', async () => {
-      const telemetryCounterPromise = analyticsClient.telemetryCounter$
-        .pipe(take(3 + 2), toArray()) // Waiting for 3 enqueued + 2 batch-shipped events
-        .toPromise();
+      const telemetryCounterPromise = lastValueFrom(
+        analyticsClient.telemetryCounter$.pipe(take(3 + 2), toArray()) // Waiting for 3 enqueued + 2 batch-shipped events
+      );
 
       // Send multiple events of 1 type to test the grouping logic as well
       analyticsClient.reportEvent('event-type-a', { a_field: 'a' });
@@ -1061,9 +1061,9 @@ describe('AnalyticsClient', () => {
     });
 
     test('Discards all the events at the shipper level (globally disabled)', async () => {
-      const telemetryCounterPromise = analyticsClient.telemetryCounter$
-        .pipe(take(3 + 2), toArray()) // Waiting for 3 enqueued + 2 batch-shipped events
-        .toPromise();
+      const telemetryCounterPromise = lastValueFrom(
+        analyticsClient.telemetryCounter$.pipe(take(3 + 2), toArray()) // Waiting for 3 enqueued + 2 batch-shipped events
+      );
 
       // Send multiple events of 1 type to test the grouping logic as well
       analyticsClient.reportEvent('event-type-a', { a_field: 'a' });
@@ -1154,9 +1154,9 @@ describe('AnalyticsClient', () => {
       analyticsClient.registerShipper(MockedShipper1, { reportEventsMock });
       analyticsClient.optIn({ global: { enabled: false } });
 
-      const telemetryCounterPromise = analyticsClient.telemetryCounter$
-        .pipe(take(3), toArray()) // Waiting for 3 enqueued
-        .toPromise();
+      const telemetryCounterPromise = lastValueFrom(
+        analyticsClient.telemetryCounter$.pipe(take(3), toArray()) // Waiting for 3 enqueued
+      );
 
       // Send multiple events of 1 type to test the non-grouping logic at this stage as well
       analyticsClient.reportEvent('event-type-a', { a_field: 'a' });
@@ -1197,9 +1197,9 @@ describe('AnalyticsClient', () => {
       analyticsClient.registerShipper(MockedShipper1, { reportEventsMock });
       analyticsClient.optIn({ global: { enabled: true } });
 
-      const telemetryCounterPromise = analyticsClient.telemetryCounter$
-        .pipe(take(3 * 2), toArray()) // Waiting for 2 events per each reportEvent call: enqueued and sent_to_shipper
-        .toPromise();
+      const telemetryCounterPromise = lastValueFrom(
+        analyticsClient.telemetryCounter$.pipe(take(3 * 2), toArray()) // Waiting for 2 events per each reportEvent call: enqueued and sent_to_shipper
+      );
 
       // Send multiple events of 1 type to test the non-grouping logic at this stage as well
       analyticsClient.reportEvent('event-type-a', { a_field: 'a' });

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-cli-dev-mode/src/optimizer.test.ts

@@ -12,7 +12,6 @@ import * as Rx from 'rxjs';
 import { toArray } from 'rxjs/operators';
 import { OptimizerUpdate } from '@kbn/optimizer';
 import { observeLines, createReplaceSerializer } from '@kbn/dev-utils';
-import { firstValueFrom } from '@kbn/std';
 
 import { Optimizer, Options } from './optimizer';
 
@@ -130,7 +129,7 @@ it('uses options to create valid OptimizerConfig', () => {
 
 it('is ready when optimizer phase is success or issue and logs in familiar format', async () => {
   const writeLogTo = new PassThrough();
-  const linesPromise = firstValueFrom(observeLines(writeLogTo).pipe(toArray()));
+  const linesPromise = Rx.firstValueFrom(observeLines(writeLogTo).pipe(toArray()));
 
   const { update$, optimizer } = setup({
     ...defaultOptions,

packages/kbn-cli-dev-mode/src/watcher.test.ts

@@ -10,7 +10,6 @@ import { EventEmitter } from 'events';
 
 import * as Rx from 'rxjs';
 import { materialize, toArray } from 'rxjs/operators';
-import { firstValueFrom } from '@kbn/std';
 
 import { TestLog } from './log';
 import { Watcher, Options } from './watcher';
@@ -112,7 +111,7 @@ it('closes chokidar watcher when unsubscribed', () => {
 
 it('rethrows chokidar errors', async () => {
   const watcher = new Watcher(defaultOptions);
-  const promise = firstValueFrom(watcher.run$.pipe(materialize(), toArray()));
+  const promise = Rx.firstValueFrom(watcher.run$.pipe(materialize(), toArray()));
 
   isMock(mockChokidar);
   mockChokidar.emit('error', new Error('foo bar'));

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) {