github-mirrors/kibana
1
0
Fork 0
mirror of https://github.com/elastic/kibana.git synced 2025-04-24 01:38:56 -04:00
Code Issues Projects Releases Packages Wiki Activity

Upgrade RxJS to 7 (#129087)

Browse source 
Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>
This commit is contained in:
Alejandro Fernández Haro 2022-04-12 21:40:55 +02:00 • committed by GitHub
parent 5ee7c3d25a
commit 9d5aca591b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
260 changed files with 11471 additions and 16589 deletions
Download patch file Download diff file Expand all files Collapse all files

7
examples/search_examples/public/search/app.tsx
View file

@ -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>;

9
examples/search_examples/public/search_sessions/app.tsx
View file

@ -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[]) {

4
package.json
View file

@ -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",

82
packages/elastic-analytics/src/analytics_client/analytics_client.test.ts
View file

@ -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' });

10
packages/kbn-cli-dev-mode/src/dev_server.ts
View file

@ -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(

3
packages/kbn-cli-dev-mode/src/optimizer.test.ts
View file

@ -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,

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

@ -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'));

14
packages/kbn-config/src/config_service.ts
View file

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

4
packages/kbn-config/src/raw/raw_config_service.test.ts
View file

@ -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']);

5
packages/kbn-dev-utils/src/tooling_log/tooling_log.test.ts
View file

@ -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');

2
packages/kbn-optimizer/src/cli.ts
View file

@ -9,7 +9,7 @@
import Path from 'path';
import { REPO_ROOT } from '@kbn/utils';
import { lastValueFrom } from '@kbn/std';
import { lastValueFrom } from 'rxjs';
import { run, createFlagError, Flags } from '@kbn/dev-utils';
import { logOptimizerState } from './log_optimizer_state';

82
packages/kbn-optimizer/src/common/rxjs_helpers.test.ts
View file

@ -7,8 +7,7 @@
*/
import * as Rx from 'rxjs';
import { toArray, map } from 'rxjs/operators';
import { lastValueFrom } from '@kbn/std';
import { fakeSchedulers } from 'rxjs-marbles/jest';
import { pipeClosure, debounceTimeBuffer, maybeMap, maybe } from './rxjs_helpers';
@ -21,26 +20,26 @@ describe('pipeClosure()', () => {
const foo$ = Rx.of(1, 2, 3).pipe(
pipeClosure((source$) => {
const multiplier = ++counter;
return source$.pipe(map((i) => i * multiplier));
return source$.pipe(Rx.map((i) => i * multiplier));
}),
toArray()
Rx.toArray()
);
await expect(lastValueFrom(foo$)).resolves.toMatchInlineSnapshot(`
await expect(Rx.lastValueFrom(foo$)).resolves.toMatchInlineSnapshot(`
Array [
1,
2,
3,
]
`);
await expect(lastValueFrom(foo$)).resolves.toMatchInlineSnapshot(`
await expect(Rx.lastValueFrom(foo$)).resolves.toMatchInlineSnapshot(`
Array [
2,
4,
6,
]
`);
await expect(lastValueFrom(foo$)).resolves.toMatchInlineSnapshot(`
await expect(Rx.lastValueFrom(foo$)).resolves.toMatchInlineSnapshot(`
Array [
3,
6,
@ -52,9 +51,9 @@ describe('pipeClosure()', () => {
describe('maybe()', () => {
it('filters out undefined values from the stream', async () => {
const foo$ = Rx.of(1, undefined, 2, undefined, 3).pipe(maybe(), toArray());
const foo$ = Rx.of(1, undefined, 2, undefined, 3).pipe(maybe(), Rx.toArray());
await expect(lastValueFrom(foo$)).resolves.toEqual([1, 2, 3]);
await expect(Rx.lastValueFrom(foo$)).resolves.toEqual([1, 2, 3]);
});
});
@ -62,10 +61,10 @@ describe('maybeMap()', () => {
it('calls map fn and filters out undefined values returned', async () => {
const foo$ = Rx.of(1, 2, 3, 4, 5).pipe(
maybeMap((i) => (i % 2 ? i : undefined)),
toArray()
Rx.toArray()
);
await expect(lastValueFrom(foo$)).resolves.toEqual([1, 3, 5]);
await expect(Rx.lastValueFrom(foo$)).resolves.toEqual([1, 3, 5]);
});
});
@ -78,39 +77,42 @@ describe('debounceTimeBuffer()', () => {
jest.useRealTimers();
});
it('buffers items until there is n milliseconds of silence, then flushes buffer to stream', async () => {
const foo$ = new Rx.Subject<number>();
const dest = new Rx.BehaviorSubject<number | undefined>(undefined);
foo$
.pipe(
debounceTimeBuffer(100),
map((items) => items.reduce((sum, n) => sum + n))
)
.subscribe(dest);
it(
'buffers items until there is n milliseconds of silence, then flushes buffer to stream',
fakeSchedulers(async (advance) => {
const foo$ = new Rx.Subject<number>();
const dest = new Rx.BehaviorSubject<number | undefined>(undefined);
foo$
.pipe(
debounceTimeBuffer(100),
Rx.map((items) => items.reduce((sum, n) => sum + n))
)
.subscribe(dest);
foo$.next(1);
expect(dest.getValue()).toBe(undefined);
foo$.next(1);
expect(dest.getValue()).toBe(undefined);
// only wait 99 milliseconds before sending the next value
jest.advanceTimersByTime(99);
foo$.next(1);
expect(dest.getValue()).toBe(undefined);
// only wait 50 milliseconds before sending the next value
advance(50);
foo$.next(1);
expect(dest.getValue()).toBe(undefined);
// only wait 99 milliseconds before sending the next value
jest.advanceTimersByTime(99);
foo$.next(1);
expect(dest.getValue()).toBe(undefined);
// only wait 99 milliseconds before sending the next value
advance(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);
foo$.next(1);
expect(dest.getValue()).toBe(3);
// send the next value after 100 milliseconds and observe that it was forwarded
advance(500);
foo$.next(1);
expect(dest.getValue()).toBe(3);
foo$.complete();
if (!dest.isStopped) {
throw new Error('Expected destination to stop as soon as the source is completed');
}
});
foo$.complete();
if (!dest.isStopped) {
throw new Error('Expected destination to stop as soon as the source is completed');
}
})
);
it('clears queue as soon as source completes if source completes before time is up', () => {
const foo$ = new Rx.Subject<number>();
@ -118,7 +120,7 @@ describe('debounceTimeBuffer()', () => {
foo$
.pipe(
debounceTimeBuffer(100),
map((items) => items.reduce((sum, n) => sum + n))
Rx.map((items) => items.reduce((sum, n) => sum + n))
)
.subscribe(dest);

14
packages/kbn-optimizer/src/common/rxjs_helpers.ts
View file

@ -7,8 +7,6 @@
*/
import * as Rx from 'rxjs';
import { mergeMap, tap, debounceTime, map, toArray } from 'rxjs/operators';
import { firstValueFrom } from '@kbn/std';
type Operator<T1, T2> = (source: Rx.Observable<T1>) => Rx.Observable<T2>;
type MapFn<T1, T2> = (item: T1, index: number) => T2;
@ -29,7 +27,7 @@ export const pipeClosure = <T1, T2>(fn: Operator<T1, T2>): Operator<T1, T2> => {
* supporting TypeScript
*/
export const maybe = <T1>(): Operator<T1 | undefined, T1> => {
return mergeMap((item) => (item === undefined ? Rx.EMPTY : [item]));
return Rx.mergeMap((item) => (item === undefined ? Rx.EMPTY : [item]));
};
/**
@ -39,7 +37,7 @@ export const maybe = <T1>(): Operator<T1 | undefined, T1> => {
* the filter.
*/
export const maybeMap = <T1, T2>(fn: MapFn<T1, undefined | T2>): Operator<T1, T2> => {
return mergeMap((item, index) => {
return Rx.mergeMap((item, index) => {
const result = fn(item, index);
return result === undefined ? Rx.EMPTY : [result];
});
@ -54,9 +52,9 @@ export const debounceTimeBuffer = <T>(ms: number) =>
pipeClosure((source$: Rx.Observable<T>) => {
const buffer: T[] = [];
return source$.pipe(
tap((item) => buffer.push(item)),
debounceTime(ms),
map(() => {
Rx.tap((item) => buffer.push(item)),
Rx.debounceTime(ms),
Rx.map(() => {
const items = Array.from(buffer);
buffer.length = 0;
return items;
@ -65,4 +63,4 @@ export const debounceTimeBuffer = <T>(ms: number) =>
});
export const allValuesFrom = <T>(observable: Rx.Observable<T>) =>
firstValueFrom(observable.pipe(toArray()));
Rx.firstValueFrom(observable.pipe(Rx.toArray()));

117
packages/kbn-optimizer/src/integration_tests/watch_bundles_for_changes.test.ts
View file

@ -8,8 +8,8 @@
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';
import { Bundle, ascending } from '../common';
import { watchBundlesForChanges$ } from '../optimizer/watch_bundles_for_changes';
@ -65,69 +65,72 @@ afterEach(async () => {
jest.useRealTimers();
});
it('notifies of changes and completes once all bundles have changed', async () => {
expect.assertions(18);
it(
'notifies of changes and completes once all bundles have changed',
fakeSchedulers(async (advance) => {
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;
}
const promise = Rx.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;
}
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');
}
// 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');
}
// 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');
}
})
)
);
// 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];
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);
// 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');
jest.advanceTimersByTime(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');
jest.advanceTimersByTime(1000);
// finish by changing box and car
changeListener(bundleEntryPath(BOX_BUNDLE), 'deleted');
changeListener(bundleEntryPath(CAR_BUNDLE), 'deleted');
advance(1000);
await expect(promise).resolves.toEqual(undefined);
});
await expect(promise).resolves.toEqual(undefined);
})
);

2
packages/kbn-optimizer/src/optimizer/observe_worker.ts
View file

@ -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

2
packages/kbn-optimizer/src/worker/run_compilers.ts
View file

@ -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;
/**

20774
packages/kbn-pm/dist/index.js vendored
View file

File diff suppressed because one or more lines are too long

6
packages/kbn-pm/src/utils/watch.ts
View file

@ -87,8 +87,8 @@ export function waitUntilWatchIsReady(stream: NodeJS.EventEmitter, opts: IWatchO
stream.once('error', onErrorListener);
stream.on('data', onDataListener);
return Rx.race(getWatchHandlers(buildOutput$, opts))
.pipe(
return Rx.firstValueFrom(
Rx.race(getWatchHandlers(buildOutput$, opts)).pipe(
mergeMap((whenReady) => whenReady),
finalize(() => {
stream.removeListener('data', onDataListener);
@ -98,5 +98,5 @@ export function waitUntilWatchIsReady(stream: NodeJS.EventEmitter, opts: IWatchO
buildOutput$.complete();
})
)
.toPromise();
);
}

4
packages/kbn-shared-ux-components/src/exit_full_screen_button/__snapshots__/exit_full_screen_button.test.tsx.snap generated
View file

@ -4,9 +4,7 @@ exports[`<ExitFullScreenButton /> is rendered 1`] = `
<SharedUxServicesProvider
application={
Object {
"currentAppId$": Observable {
"_isScalar": false,
},
"currentAppId$": Observable {},
"navigateToUrl": [Function],
}
}

12
packages/kbn-shared-ux-components/src/page_template/no_data_page/no_data_card/__snapshots__/elastic_agent_card.component.test.tsx.snap generated
View file

@ -4,10 +4,9 @@ exports[`ElasticAgentCardComponent props button 1`] = `
<RedirectAppLinks
currentAppId$={
Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -30,10 +29,9 @@ exports[`ElasticAgentCardComponent props href 1`] = `
<RedirectAppLinks
currentAppId$={
Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -56,10 +54,9 @@ exports[`ElasticAgentCardComponent renders 1`] = `
<RedirectAppLinks
currentAppId$={
Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -81,10 +78,9 @@ exports[`ElasticAgentCardComponent renders with canAccessFleet false 1`] = `
<RedirectAppLinks
currentAppId$={
Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],

16
packages/kbn-shared-ux-components/src/page_template/no_data_page/no_data_card/__snapshots__/elastic_agent_card.test.tsx.snap generated
View file

@ -4,9 +4,7 @@ exports[`ElasticAgentCard renders 1`] = `
<SharedUxServicesProvider
application={
Object {
"currentAppId$": Observable {
"_isScalar": false,
},
"currentAppId$": Observable {},
"navigateToUrl": [Function],
}
}
@ -142,20 +140,12 @@ exports[`ElasticAgentCard renders 1`] = `
<ElasticAgentCard>
<ElasticAgentCardComponent
canAccessFleet={true}
currentAppId$={
Observable {
"_isScalar": false,
}
}
currentAppId$={Observable {}}
href="/app/integrations/browse"
navigateToUrl={[Function]}
>
<RedirectAppLinks
currentAppId$={
Observable {
"_isScalar": false,
}
}
currentAppId$={Observable {}}
navigateToUrl={[Function]}
>
<div>

4
packages/kbn-shared-ux-components/src/toolbar/buttons/icon_button_group/__snapshots__/icon_button_group.test.tsx.snap generated
View file

@ -4,9 +4,7 @@ exports[`<IconButtonGroup /> is rendered 1`] = `
<SharedUxServicesProvider
application={
Object {
"currentAppId$": Observable {
"_isScalar": false,
},
"currentAppId$": Observable {},
"navigateToUrl": [Function],
}
}

4
packages/kbn-shared-ux-components/src/toolbar/buttons/primary/__snapshots__/primary.test.tsx.snap generated
View file

@ -4,9 +4,7 @@ exports[`<ToolbarButton /> is rendered 1`] = `
<SharedUxServicesProvider
application={
Object {
"currentAppId$": Observable {
"_isScalar": false,
},
"currentAppId$": Observable {},
"navigateToUrl": [Function],
}
}

1
packages/kbn-std/src/index.ts
View file

@ -19,7 +19,6 @@ export { isRelativeUrl, modifyUrl, getUrlOrigin } from './url';
export { unset } from './unset';
export { getFlattenedObject } from './get_flattened_object';
export { ensureNoUnsafeProperties } from './ensure_no_unsafe_properties';
export * from './rxjs_7';
export {
map$,
mapWithLimit$,

8
packages/kbn-std/src/iteration/for_each.ts
View file

@ -6,9 +6,7 @@
* Side Public License, v 1.
*/
import { defaultIfEmpty } from 'rxjs/operators';
import { lastValueFrom } from '../rxjs_7';
import { lastValueFrom, defaultIfEmpty } from 'rxjs';
import { mapWithLimit$ } from './observable';
import { IterableInput, AsyncMapFn } from './types';
@ -21,7 +19,7 @@ import { IterableInput, AsyncMapFn } from './types';
* @param fn Function to call for each item
*/
export async function asyncForEach<T>(iterable: IterableInput<T>, fn: AsyncMapFn<T, any>) {
await lastValueFrom(mapWithLimit$(iterable, Infinity, fn).pipe(defaultIfEmpty()));
await lastValueFrom(mapWithLimit$(iterable, Infinity, fn).pipe(defaultIfEmpty(undefined)));
}
/**
@ -40,5 +38,5 @@ export async function asyncForEachWithLimit<T>(
limit: number,
fn: AsyncMapFn<T, any>
) {
await lastValueFrom(mapWithLimit$(iterable, limit, fn).pipe(defaultIfEmpty()));
await lastValueFrom(mapWithLimit$(iterable, limit, fn).pipe(defaultIfEmpty(undefined)));
}

3
packages/kbn-std/src/iteration/map.ts
View file

@ -6,9 +6,8 @@
* Side Public License, v 1.
*/
import { from } from 'rxjs';
import { from, lastValueFrom } from 'rxjs';
import { toArray } from 'rxjs/operators';
import { lastValueFrom } from '../rxjs_7';
import { IterableInput, AsyncMapFn, AsyncMapResult } from './types';
import { mapWithLimit$ } from './observable';

7
packages/kbn-std/src/iteration/observable.test.ts
View file

@ -8,7 +8,6 @@
import * as Rx from 'rxjs';
import { toArray } from 'rxjs/operators';
import { lastValueFrom } from '../rxjs_7';
import { map$, mapWithLimit$ } from './observable';
import { list, sleep, generator } from './test_helpers';
@ -23,7 +22,7 @@ describe('mapWithLimit$', () => {
let active = 0;
const limit = Math.random() > 0.5 ? 20 : 40;
const results = await lastValueFrom(
const results = await Rx.lastValueFrom(
mapWithLimit$(list(100), limit, async (n) => {
active += 1;
if (active > maxConcurrency) {
@ -50,7 +49,7 @@ describe('mapWithLimit$', () => {
['observable', Rx.of(1, 2, 3, 4, 5), [1, 2, 3, 4, 5]] as const,
])('works with %p', async (_, iter, expected) => {
const mock = jest.fn(async (n) => n);
const results = await lastValueFrom(mapWithLimit$(iter, 1, mock).pipe(toArray()));
const results = await Rx.lastValueFrom(mapWithLimit$(iter, 1, mock).pipe(toArray()));
expect(results).toEqual(expected);
});
});
@ -60,7 +59,7 @@ describe('map$', () => {
let maxConcurrency = 0;
let active = 0;
const results = await lastValueFrom(
const results = await Rx.lastValueFrom(
map$(list(100), async (n) => {
active += 1;
if (active > maxConcurrency) {

6
packages/kbn-std/src/iteration/types.ts
View file

@ -6,8 +6,8 @@
* Side Public License, v 1.
*/
import { Subscribable } from 'rxjs';
import { ObservableInput } from 'rxjs';
export type IterableInput<T> = Iterable<T> | Subscribable<T>;
export type AsyncMapResult<T> = Promise<T> | Subscribable<T>;
export type IterableInput<T> = Iterable<T> | ObservableInput<T>;
export type AsyncMapResult<T> = Promise<T> | ObservableInput<T>;
export type AsyncMapFn<T1, T2> = (item: T1, i: number) => AsyncMapResult<T2>;

82
packages/kbn-std/src/rxjs_7.test.ts
View file

@ -1,82 +0,0 @@
/*
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
* or more contributor license agreements. Licensed under the Elastic License
* 2.0 and the Server Side Public License, v 1; you may not use this file except
* in compliance with, at your election, the Elastic License 2.0 or the Server
* Side Public License, v 1.
*/
import * as Rx from 'rxjs';
import { firstValueFrom, lastValueFrom } from './rxjs_7';
// create an empty observable that completes with no notifications
// after a delay to ensure helpers aren't checking for the EMPTY constant
function empty() {
return new Rx.Observable<never>((subscriber) => {
setTimeout(() => {
subscriber.complete();
}, 0);
});
}
describe('firstValueFrom()', () => {
it('resolves to the first value from the observable', async () => {
await expect(firstValueFrom(Rx.of(1, 2, 3))).resolves.toBe(1);
});
it('rejects if the observable is empty', async () => {
await expect(firstValueFrom(empty())).rejects.toThrowErrorMatchingInlineSnapshot(
`"no elements in sequence"`
);
});
it('unsubscribes from a source observable that emits synchronously', async () => {
const values = [1, 2, 3, 4];
let unsubscribed = false;
const source = new Rx.Observable<number>((subscriber) => {
while (!subscriber.closed && values.length) {
subscriber.next(values.shift()!);
}
unsubscribed = subscriber.closed;
subscriber.complete();
});
await expect(firstValueFrom(source)).resolves.toMatchInlineSnapshot(`1`);
if (!unsubscribed) {
throw new Error('expected source to be unsubscribed');
}
expect(values).toEqual([2, 3, 4]);
});
it('unsubscribes from the source observable after first async notification', async () => {
const values = [1, 2, 3, 4];
let unsubscribed = false;
const source = new Rx.Observable<number>((subscriber) => {
setTimeout(() => {
while (!subscriber.closed) {
subscriber.next(values.shift()!);
}
unsubscribed = subscriber.closed;
});
});
await expect(firstValueFrom(source)).resolves.toMatchInlineSnapshot(`1`);
if (!unsubscribed) {
throw new Error('expected source to be unsubscribed');
}
expect(values).toEqual([2, 3, 4]);
});
});
describe('lastValueFrom()', () => {
it('resolves to the last value from the observable', async () => {
await expect(lastValueFrom(Rx.of(1, 2, 3))).resolves.toBe(3);
});
it('rejects if the observable is empty', async () => {
await expect(lastValueFrom(empty())).rejects.toThrowErrorMatchingInlineSnapshot(
`"no elements in sequence"`
);
});
});

19
packages/kbn-std/src/rxjs_7.ts
View file

@ -1,19 +0,0 @@
/*
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
* or more contributor license agreements. Licensed under the Elastic License
* 2.0 and the Server Side Public License, v 1; you may not use this file except
* in compliance with, at your election, the Elastic License 2.0 or the Server
* Side Public License, v 1.
*/
import { Observable } from 'rxjs';
import { first, last } from 'rxjs/operators';
export function firstValueFrom<T>(source: Observable<T>) {
// we can't use SafeSubscriber the same way that RxJS 7 does, so instead we
return source.pipe(first()).toPromise();
}
export function lastValueFrom<T>(source: Observable<T>) {
return source.pipe(last()).toPromise();
}

3
packages/kbn-test/src/kbn_client/kbn_client_saved_objects.ts
View file

@ -10,7 +10,6 @@ import { inspect } from 'util';
import * as Rx from 'rxjs';
import { mergeMap } from 'rxjs/operators';
import { lastValueFrom } from '@kbn/std';
import { ToolingLog, isAxiosResponseError, createFailError } from '@kbn/dev-utils';
import { KbnClientRequester, uriencode } from './kbn_client_requester';
@ -83,7 +82,7 @@ interface DeleteObjectsOptions {
async function concurrently<T>(maxConcurrency: number, arr: T[], fn: (item: T) => Promise<void>) {
if (arr.length) {
await lastValueFrom(
await Rx.lastValueFrom(
Rx.from(arr).pipe(mergeMap(async (item) => await fn(item), maxConcurrency))
);
}

2
src/cli/serve/integration_tests/reload_logging_config.test.ts
View file

@ -117,7 +117,7 @@ describe.skip('Server logging configuration', function () {
)
.toPromise();
const lastMessage = await message$.pipe(take(1)).toPromise();
const lastMessage = await Rx.firstValueFrom(message$);
expect(containsJsonOnly(lastMessage)).toBe(true);
createConfigManager(configFilePath).modify((oldConfig) => {

25
src/core/public/application/__snapshots__/application_service.test.ts.snap generated
View file

@ -4,15 +4,15 @@ exports[`#start() getComponent returns renderable JSX tree 1`] = `
<AppRouter
appStatuses$={
AnonymousSubject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"destination": AnonymousSubject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"destination": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": Array [],
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -21,14 +21,11 @@ exports[`#start() getComponent returns renderable JSX tree 1`] = `
"hasError": false,
"isStopped": false,
"observers": Array [],
"operator": MapOperator {
"project": [Function],
"thisArg": undefined,
},
"operator": [Function],
"source": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": Array [],
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -41,12 +38,12 @@ exports[`#start() getComponent returns renderable JSX tree 1`] = `
"observers": Array [],
"operator": [Function],
"source": AnonymousSubject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"destination": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": Array [],
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -55,14 +52,11 @@ exports[`#start() getComponent returns renderable JSX tree 1`] = `
"hasError": false,
"isStopped": false,
"observers": Array [],
"operator": MapOperator {
"project": [Function],
"thisArg": undefined,
},
"operator": [Function],
"source": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": Array [],
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -85,7 +79,6 @@ exports[`#start() getComponent returns renderable JSX tree 1`] = `
setIsMounting={[Function]}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}

28
src/core/public/application/application_service.test.ts
View file

@ -13,8 +13,8 @@ import {
} from './application_service.test.mocks';
import { createElement } from 'react';
import { BehaviorSubject, Subject } from 'rxjs';
import { bufferCount, take, takeUntil } from 'rxjs/operators';
import { BehaviorSubject, firstValueFrom, Subject } from 'rxjs';
import { bufferCount, takeUntil } from 'rxjs/operators';
import { mount, shallow } from 'enzyme';
import { httpServiceMock } from '../http/http_service.mock';
@ -96,7 +96,7 @@ describe('#setup()', () => {
);
const { applications$ } = await service.start(startDeps);
let applications = await applications$.pipe(take(1)).toPromise();
let applications = await firstValueFrom(applications$);
expect(applications.size).toEqual(2);
expect(applications.get('app1')).toEqual(
expect.objectContaining({
@ -125,7 +125,7 @@ describe('#setup()', () => {
deepLinks: [{ id: 'subapp2', title: 'Subapp 2', path: '/subapp2' }],
}));
applications = await applications$.pipe(take(1)).toPromise();
applications = await firstValueFrom(applications$);
expect(applications.size).toEqual(2);
expect(applications.get('app1')).toEqual(
expect.objectContaining({
@ -205,7 +205,7 @@ describe('#setup()', () => {
})
);
const start = await service.start(startDeps);
const applications = await start.applications$.pipe(take(1)).toPromise();
const applications = await firstValueFrom(start.applications$);
expect(applications.size).toEqual(2);
expect(applications.get('app1')).toEqual(
@ -252,7 +252,7 @@ describe('#setup()', () => {
);
const { applications$ } = await service.start(startDeps);
const applications = await applications$.pipe(take(1)).toPromise();
const applications = await firstValueFrom(applications$);
expect(applications.size).toEqual(2);
expect(applications.get('app1')).toEqual(
@ -295,7 +295,7 @@ describe('#setup()', () => {
);
const start = await service.start(startDeps);
const applications = await start.applications$.pipe(take(1)).toPromise();
const applications = await firstValueFrom(start.applications$);
expect(applications.size).toEqual(1);
expect(applications.get('app1')).toEqual(
@ -410,7 +410,7 @@ describe('#setup()', () => {
updater$.next((app) => ({ defaultPath: '/foo' }));
let appInfos = await applications$.pipe(take(1)).toPromise();
let appInfos = await firstValueFrom(applications$);
expect(appInfos.get('app1')!.deepLinks).toEqual([
{
@ -445,7 +445,7 @@ describe('#setup()', () => {
],
}));
appInfos = await applications$.pipe(take(1)).toPromise();
appInfos = await firstValueFrom(applications$);
expect(appInfos.get('app1')!.deepLinks).toEqual([
{
@ -496,7 +496,7 @@ describe('#start()', () => {
register(Symbol(), createApp({ id: 'app2' }));
const { applications$ } = await service.start(startDeps);
const availableApps = await applications$.pipe(take(1)).toPromise();
const availableApps = await firstValueFrom(applications$);
expect(availableApps.size).toEqual(2);
expect([...availableApps.keys()]).toEqual(['app1', 'app2']);
@ -548,7 +548,7 @@ describe('#start()', () => {
register(Symbol(), createApp({ id: 'app2' }));
const { applications$ } = await service.start(startDeps);
const availableApps = await applications$.pipe(take(1)).toPromise();
const availableApps = await firstValueFrom(applications$);
expect([...availableApps.keys()]).toEqual(['app1']);
});
@ -802,7 +802,7 @@ describe('#start()', () => {
service.setup(setupDeps);
const { currentAppId$, navigateToApp } = await service.start(startDeps);
const stop$ = new Subject();
const stop$ = new Subject<void>();
const promise = currentAppId$.pipe(bufferCount(4), takeUntil(stop$)).toPromise();
await navigateToApp('alpha');
@ -827,7 +827,7 @@ describe('#start()', () => {
service.setup(setupDeps);
const { currentAppId$, navigateToApp } = await service.start(startDeps);
const stop$ = new Subject();
const stop$ = new Subject<void>();
const promise = currentAppId$.pipe(bufferCount(4), takeUntil(stop$)).toPromise();
await navigateToApp('delta', { openInNewTab: true });
@ -871,7 +871,7 @@ describe('#start()', () => {
const { navigateToApp, getComponent } = await service.start(startDeps);
const httpLoadingCount$ = startDeps.http.addLoadingCountSource.mock.calls[0][0];
const stop$ = new Subject();
const stop$ = new Subject<void>();
const currentLoadingCount$ = new BehaviorSubject(0);
httpLoadingCount$.pipe(takeUntil(stop$)).subscribe(currentLoadingCount$);
const loadingPromise = httpLoadingCount$.pipe(bufferCount(5), takeUntil(stop$)).toPromise();

6
src/core/public/application/application_service.tsx
View file

@ -7,7 +7,7 @@
*/
import React from 'react';
import { BehaviorSubject, Observable, Subject, Subscription } from 'rxjs';
import { BehaviorSubject, firstValueFrom, Observable, Subject, Subscription } from 'rxjs';
import { map, shareReplay, takeUntil, distinctUntilChanged, filter, take } from 'rxjs/operators';
import { createBrowserHistory, History } from 'history';
@ -99,7 +99,7 @@ export class ApplicationService {
private currentActionMenu$ = new BehaviorSubject<MountPoint | undefined>(undefined);
private readonly statusUpdaters$ = new BehaviorSubject<Map<symbol, AppUpdaterWrapper>>(new Map());
private readonly subscriptions: Subscription[] = [];
private stop$ = new Subject();
private stop$ = new Subject<void>();
private registrationClosed = false;
private history?: History<any>;
private navigate?: (url: string, state: unknown, replace: boolean) => void;
@ -120,7 +120,7 @@ export class ApplicationService {
createBrowserHistory({
basename,
getUserConfirmation: getUserConfirmationHandler({
overlayPromise: this.overlayStart$.pipe(take(1)).toPromise(),
overlayPromise: firstValueFrom(this.overlayStart$.pipe(take(1))),
}),
});

4
src/core/public/chrome/chrome_service.test.ts
View file

@ -9,7 +9,7 @@
import { shallow } from 'enzyme';
import React from 'react';
import * as Rx from 'rxjs';
import { take, toArray } from 'rxjs/operators';
import { toArray } from 'rxjs/operators';
import { App, PublicAppInfo } from '../application';
import { applicationServiceMock } from '../application/application_service.mock';
import { docLinksServiceMock } from '../doc_links/doc_links_service.mock';
@ -229,7 +229,7 @@ describe('start', () => {
const { chrome, service } = await start({ startDeps });
const promise = chrome.getIsVisible$().pipe(toArray()).toPromise();
const availableApps = await applications$.pipe(take(1)).toPromise();
const availableApps = await Rx.firstValueFrom(applications$);
[...availableApps.keys()].forEach((appId) => navigateToApp(appId));
service.stop();

2
src/core/public/chrome/chrome_service.tsx
View file

@ -55,7 +55,7 @@ export interface StartDeps {
export class ChromeService {
private isVisible$!: Observable<boolean>;
private isForceHidden$!: BehaviorSubject<boolean>;
private readonly stop$ = new ReplaySubject(1);
private readonly stop$ = new ReplaySubject<void>(1);
private readonly navControls = new NavControlsService();
private readonly navLinks = new NavLinksService();
private readonly recentlyAccessed = new RecentlyAccessedService();

2
src/core/public/chrome/nav_controls/nav_controls_service.ts
View file

@ -50,7 +50,7 @@ export interface ChromeNavControls {
/** @internal */
export class NavControlsService {
private readonly stop$ = new ReplaySubject(1);
private readonly stop$ = new ReplaySubject<void>(1);
public start() {
const navControlsLeft$ = new BehaviorSubject<ReadonlySet<ChromeNavControl>>(new Set());

2
src/core/public/chrome/nav_links/nav_links_service.ts
View file

@ -69,7 +69,7 @@ export interface ChromeNavLinks {
}
export class NavLinksService {
private readonly stop$ = new ReplaySubject(1);
private readonly stop$ = new ReplaySubject<void>(1);
public start({ application, http }: StartDeps): ChromeNavLinks {
const navLinks$ = new BehaviorSubject<ReadonlyMap<string, NavLinkWrapper>>(new Map());

2
src/core/public/chrome/recently_accessed/recently_accessed_service.test.ts
View file

@ -90,7 +90,7 @@ describe('RecentlyAccessed#start()', () => {
it('exposes an observable', async () => {
const { recentlyAccessed } = await getStart();
const stop$ = new Subject();
const stop$ = new Subject<void>();
const observedValues$ = recentlyAccessed
.get$()
.pipe(bufferCount(3), takeUntil(stop$))

504
src/core/public/chrome/ui/header/__snapshots__/collapsible_nav.test.tsx.snap generated
View file

@ -4,48 +4,32 @@ exports[`CollapsibleNav renders links grouped by category 1`] = `
<CollapsibleNav
appId$={
BehaviorSubject {
"_isScalar": false,
"_value": "test",
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -65,7 +49,6 @@ exports[`CollapsibleNav renders links grouped by category 1`] = `
closeNav={[Function]}
customNavLink$={
BehaviorSubject {
"_isScalar": false,
"_value": Object {
"baseUrl": "/",
"category": undefined,
@ -77,45 +60,30 @@ exports[`CollapsibleNav renders links grouped by category 1`] = `
"url": "/",
},
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -127,7 +95,6 @@ exports[`CollapsibleNav renders links grouped by category 1`] = `
isNavOpen={true}
navLinks$={
BehaviorSubject {
"_isScalar": false,
"_value": Array [
Object {
"baseUrl": "/",
@ -246,45 +213,30 @@ exports[`CollapsibleNav renders links grouped by category 1`] = `
},
],
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -295,7 +247,6 @@ exports[`CollapsibleNav renders links grouped by category 1`] = `
onIsLockedUpdate={[Function]}
recentlyAccessed$={
BehaviorSubject {
"_isScalar": false,
"_value": Array [
Object {
"id": "recent 1",
@ -309,45 +260,30 @@ exports[`CollapsibleNav renders links grouped by category 1`] = `
},
],
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -2073,48 +2009,32 @@ exports[`CollapsibleNav renders the default nav 1`] = `
<CollapsibleNav
appId$={
BehaviorSubject {
"_isScalar": false,
"_value": "test",
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -2134,48 +2054,32 @@ exports[`CollapsibleNav renders the default nav 1`] = `
closeNav={[Function]}
customNavLink$={
BehaviorSubject {
"_isScalar": false,
"_value": undefined,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -2187,48 +2091,32 @@ exports[`CollapsibleNav renders the default nav 1`] = `
isNavOpen={false}
navLinks$={
BehaviorSubject {
"_isScalar": false,
"_value": Array [],
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -2239,48 +2127,32 @@ exports[`CollapsibleNav renders the default nav 1`] = `
onIsLockedUpdate={[Function]}
recentlyAccessed$={
BehaviorSubject {
"_isScalar": false,
"_value": Array [],
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -2321,48 +2193,32 @@ exports[`CollapsibleNav renders the default nav 2`] = `
<CollapsibleNav
appId$={
BehaviorSubject {
"_isScalar": false,
"_value": "test",
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -2382,48 +2238,32 @@ exports[`CollapsibleNav renders the default nav 2`] = `
closeNav={[Function]}
customNavLink$={
BehaviorSubject {
"_isScalar": false,
"_value": undefined,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -2436,48 +2276,32 @@ exports[`CollapsibleNav renders the default nav 2`] = `
isOpen={true}
navLinks$={
BehaviorSubject {
"_isScalar": false,
"_value": Array [],
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,
@ -2488,48 +2312,32 @@ exports[`CollapsibleNav renders the default nav 2`] = `
onIsLockedUpdate={[Function]}
recentlyAccessed$={
BehaviorSubject {
"_isScalar": false,
"_value": Array [],
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [
Subscriber {
"_parentOrParents": null,
"_subscriptions": Array [
SubjectSubscription {
"_parentOrParents": [Circular],
"_subscriptions": null,
SafeSubscriber {
"_finalizers": Array [
Subscription {
"_finalizers": null,
"_parentage": [Circular],
"closed": false,
"subject": [Circular],
"subscriber": [Circular],
"initialTeardown": [Function],
},
],
"_parentage": null,
"closed": false,
"destination": SafeSubscriber {
"_complete": undefined,
"_context": [Circular],
"_error": undefined,
"_next": [Function],
"_parentOrParents": null,
"_parentSubscriber": [Circular],
"_subscriptions": null,
"closed": false,
"destination": Object {
"closed": true,
"complete": [Function],
"error": [Function],
"destination": ConsumerObserver {
"partialObserver": Object {
"complete": undefined,
"error": undefined,
"next": [Function],
},
"isStopped": false,
"syncErrorThrowable": false,
"syncErrorThrown": false,
"syncErrorValue": null,
},
"initialTeardown": undefined,
"isStopped": false,
"syncErrorThrowable": true,
"syncErrorThrown": false,
"syncErrorValue": null,
},
],
"thrownError": null,

4012
src/core/public/chrome/ui/header/__snapshots__/header.test.tsx.snap generated
View file

File diff suppressed because it is too large Load diff

2
src/core/public/http/loading_count_service.ts
View file

@ -34,7 +34,7 @@ export type LoadingCountStart = LoadingCountSetup;
/** @internal */
export class LoadingCountService implements CoreService<LoadingCountSetup, LoadingCountStart> {
private readonly stop$ = new Subject();
private readonly stop$ = new Subject<void>();
private readonly loadingCount$ = new BehaviorSubject(0);
public setup({ fatalErrors }: { fatalErrors: FatalErrorsSetup }) {

4
src/core/public/notifications/toasts/__snapshots__/toasts_service.test.tsx.snap generated
View file

@ -12,7 +12,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -22,11 +21,10 @@ Array [
dismissToast={[Function]}
toasts$={
Observable {
"_isScalar": false,
"source": BehaviorSubject {
"_isScalar": false,
"_value": Array [],
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],

4
src/core/public/notifications/toasts/toasts_api.test.ts
View file

@ -6,7 +6,7 @@
* Side Public License, v 1.
*/
import { take } from 'rxjs/operators';
import { firstValueFrom } from 'rxjs';
import { ToastsApi } from './toasts_api';
@ -14,7 +14,7 @@ import { uiSettingsServiceMock } from '../../ui_settings/ui_settings_service.moc
import { i18nServiceMock } from '../../i18n/i18n_service.mock';
async function getCurrentToasts(toasts: ToastsApi) {
return await toasts.get$().pipe(take(1)).toPromise();
return await firstValueFrom(toasts.get$());
}
function uiSettingsMock() {

7
src/core/public/overlays/flyout/__snapshots__/flyout_service.test.tsx.snap generated
View file

@ -20,7 +20,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -56,7 +55,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -81,7 +79,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -103,7 +100,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -132,7 +128,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -157,7 +152,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -179,7 +173,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}

49
src/core/public/overlays/modal/__snapshots__/modal_service.test.tsx.snap generated
View file

@ -22,7 +22,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -47,7 +46,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -69,7 +67,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -108,7 +105,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -131,7 +127,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -159,7 +154,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -180,7 +174,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -205,7 +198,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -241,7 +233,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -264,7 +255,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -290,7 +280,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -315,7 +304,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -336,7 +324,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -360,7 +347,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -382,7 +368,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -411,7 +396,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -434,7 +418,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -460,7 +443,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -485,7 +467,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -506,7 +487,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -530,7 +510,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -552,7 +531,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -586,7 +564,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -609,7 +586,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -635,7 +611,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -660,7 +635,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -681,7 +655,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -705,7 +678,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -727,7 +699,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -756,7 +727,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -779,7 +749,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -805,7 +774,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -830,7 +798,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -851,7 +818,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -875,7 +841,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -897,7 +862,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -929,7 +893,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -965,7 +928,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -990,7 +952,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -1012,7 +973,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -1041,7 +1001,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -1066,7 +1025,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -1088,7 +1046,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -1122,7 +1079,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -1147,7 +1103,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -1169,7 +1124,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -1198,7 +1152,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -1223,7 +1176,6 @@ Array [
globalStyles={false}
theme$={
Observable {
"_isScalar": false,
"_subscribe": [Function],
}
}
@ -1245,7 +1197,6 @@ Array [
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}

5
src/core/public/plugins/plugin.ts
View file

@ -6,8 +6,7 @@
* Side Public License, v 1.
*/
import { Subject } from 'rxjs';
import { first } from 'rxjs/operators';
import { firstValueFrom, Subject } from 'rxjs';
import { DiscoveredPlugin, PluginOpaqueId } from '../../server';
import { PluginInitializerContext } from './plugin_context';
import { read } from './plugin_reader';
@ -61,7 +60,7 @@ export class PluginWrapper<
private instance?: Plugin<TSetup, TStart, TPluginsSetup, TPluginsStart>;
private readonly startDependencies$ = new Subject<[CoreStart, TPluginsStart, TStart]>();
public readonly startDependencies = this.startDependencies$.pipe(first()).toPromise();
public readonly startDependencies = firstValueFrom(this.startDependencies$);
constructor(
public readonly discoveredPlugin: DiscoveredPlugin,

2
src/core/public/theme/theme_service.ts
View file

@ -17,7 +17,7 @@ export interface SetupDeps {
export class ThemeService {
private theme$?: Observable<CoreTheme>;
private stop$ = new Subject();
private stop$ = new Subject<void>();
public setup({ injectedMetadata }: SetupDeps): ThemeServiceSetup {
const theme = injectedMetadata.getTheme();

4
src/core/public/ui_settings/ui_settings_api.test.ts
View file

@ -169,7 +169,7 @@ describe('#getLoadingCount$()', () => {
});
const { uiSettingsApi } = setup();
const done$ = new Rx.Subject();
const done$ = new Rx.Subject<void>();
const promise = uiSettingsApi.getLoadingCount$().pipe(takeUntil(done$), toArray()).toPromise();
await uiSettingsApi.batchSet('foo', 'bar');
@ -194,7 +194,7 @@ describe('#getLoadingCount$()', () => {
);
const { uiSettingsApi } = setup();
const done$ = new Rx.Subject();
const done$ = new Rx.Subject<void>();
const promise = uiSettingsApi.getLoadingCount$().pipe(takeUntil(done$), toArray()).toPromise();
await uiSettingsApi.batchSet('foo', 'bar');

5
src/core/public/ui_settings/ui_settings_service.test.ts
View file

@ -34,7 +34,10 @@ describe('#stop', () => {
service.stop();
await expect(
Rx.combineLatest(client.getUpdate$(), client.getUpdateErrors$(), loadingCount$!).toPromise()
Rx.lastValueFrom(
Rx.combineLatest([client.getUpdate$(), client.getUpdateErrors$(), loadingCount$!]),
{ defaultValue: undefined }
)
).resolves.toBe(undefined);
});
});

4
src/core/server/core_usage_data/core_usage_data_service.test.ts
View file

@ -8,9 +8,11 @@
import type { ConfigPath } from '@kbn/config';
import { BehaviorSubject, Observable } from 'rxjs';
import { HotObservable } from 'rxjs/internal/testing/HotObservable';
import { TestScheduler } from 'rxjs/testing';
// eslint-disable-next-line @kbn/imports/no_unresolved_imports
import { HotObservable } from 'rxjs/internal/testing/HotObservable';
import { configServiceMock } from '../config/mocks';
import { mockCoreContext } from '../core_context.mock';

8
src/core/server/core_usage_data/core_usage_data_service.ts
View file

@ -6,8 +6,8 @@
* Side Public License, v 1.
*/
import { Subject, Observable } from 'rxjs';
import { takeUntil, first } from 'rxjs/operators';
import { Subject, Observable, firstValueFrom } from 'rxjs';
import { takeUntil } from 'rxjs/operators';
import { get } from 'lodash';
import { hasConfigPathIntersection, ChangedDeprecatedPaths } from '@kbn/config';
@ -100,7 +100,7 @@ export class CoreUsageDataService
constructor(core: CoreContext) {
this.logger = core.logger.get('core-usage-stats-service');
this.configService = core.configService;
this.stop$ = new Subject();
this.stop$ = new Subject<void>();
}
private async getSavedObjectUsageData(
@ -383,7 +383,7 @@ export class CoreUsageDataService
private async getNonDefaultKibanaConfigs(
exposedConfigsToUsage: ExposedConfigsToUsage
): Promise<ConfigUsageData> {
const config = await this.configService.getConfig$().pipe(first()).toPromise();
const config = await firstValueFrom(this.configService.getConfig$());
const nonDefaultConfigs = config.toRaw();
const usedPaths = await this.configService.getUsedPaths();
const exposedConfigsKeys = [...exposedConfigsToUsage.keys()];

9
src/core/server/deprecations/deprecations_service.ts
View file

@ -6,7 +6,7 @@
* Side Public License, v 1.
*/
import { take } from 'rxjs/operators';
import { firstValueFrom } from 'rxjs';
import { DeprecationsFactory } from './deprecations_factory';
import { DomainDeprecationDetails, RegisterDeprecationsConfig } from './types';
@ -144,10 +144,9 @@ export class DeprecationsService
public async setup({ http }: DeprecationsSetupDeps): Promise<InternalDeprecationsServiceSetup> {
this.logger.debug('Setting up Deprecations service');
const config = await this.configService
.atPath<DeprecationConfigType>(deprecationConfig.path)
.pipe(take(1))
.toPromise();
const config = await firstValueFrom(
this.configService.atPath<DeprecationConfigType>(deprecationConfig.path)
);
this.deprecationsFactory = new DeprecationsFactory({
logger: this.logger,

12
src/core/server/elasticsearch/elasticsearch_service.ts
View file

@ -6,8 +6,8 @@
* Side Public License, v 1.
*/
import { Observable, Subject } from 'rxjs';
import { first, map, shareReplay, takeUntil } from 'rxjs/operators';
import { firstValueFrom, Observable, Subject } from 'rxjs';
import { map, shareReplay, takeUntil } from 'rxjs/operators';
import { CoreService } from '../../types';
import { CoreContext } from '../core_context';
@ -41,7 +41,7 @@ export class ElasticsearchService
{
private readonly log: Logger;
private readonly config$: Observable<ElasticsearchConfig>;
private stop$ = new Subject();
private stop$ = new Subject<void>();
private kibanaVersion: string;
private authHeaders?: IAuthHeadersStorage;
private executionContextClient?: IExecutionContext;
@ -60,7 +60,7 @@ export class ElasticsearchService
public async preboot(): Promise<InternalElasticsearchServicePreboot> {
this.log.debug('Prebooting elasticsearch service');
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
return {
config: {
hosts: config.hosts,
@ -76,7 +76,7 @@ export class ElasticsearchService
public async setup(deps: SetupDeps): Promise<InternalElasticsearchServiceSetup> {
this.log.debug('Setting up elasticsearch service');
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
this.authHeaders = deps.http.authRequestHeaders;
this.executionContextClient = deps.executionContext;
@ -112,7 +112,7 @@ export class ElasticsearchService
throw new Error('ElasticsearchService needs to be setup before calling start');
}
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
// Log every error we may encounter in the connection to Elasticsearch
this.esNodesCompatibility$.subscribe(({ isCompatible, message }) => {

8
src/core/server/environment/environment_service.ts
View file

@ -6,7 +6,7 @@
* Side Public License, v 1.
*/
import { take } from 'rxjs/operators';
import { firstValueFrom } from 'rxjs';
import { PathConfigType, config as pathConfigDef } from '@kbn/utils';
import { CoreContext } from '../core_context';
import { Logger } from '../logging';
@ -49,9 +49,9 @@ export class EnvironmentService {
// IMPORTANT: This code is based on the assumption that none of the configuration values used
// here is supposed to change during preboot phase and it's safe to read them only once.
const [pathConfig, serverConfig, pidConfig] = await Promise.all([
this.configService.atPath<PathConfigType>(pathConfigDef.path).pipe(take(1)).toPromise(),
this.configService.atPath<HttpConfigType>(httpConfigDef.path).pipe(take(1)).toPromise(),
this.configService.atPath<PidConfigType>(pidConfigDef.path).pipe(take(1)).toPromise(),
firstValueFrom(this.configService.atPath<PathConfigType>(pathConfigDef.path)),
firstValueFrom(this.configService.atPath<HttpConfigType>(httpConfigDef.path)),
firstValueFrom(this.configService.atPath<PidConfigType>(pidConfigDef.path)),
]);
// Log unhandled rejections so that we can fix them in preparation for https://github.com/elastic/kibana/issues/77469

4
src/core/server/http/http_server.ts
View file

@ -19,7 +19,7 @@ import {
import agent from 'elastic-apm-node';
import type { Duration } from 'moment';
import { Observable } from 'rxjs';
import { firstValueFrom, Observable } from 'rxjs';
import { take } from 'rxjs/operators';
import apm from 'elastic-apm-node';
import { Logger, LoggerFactory } from '../logging';
@ -223,7 +223,7 @@ export class HttpServer {
if (hasStarted) {
this.log.debug('stopping http server');
const shutdownTimeout = await this.shutdownTimeout$.pipe(take(1)).toPromise();
const shutdownTimeout = await firstValueFrom(this.shutdownTimeout$.pipe(take(1)));
await this.server.stop({ timeout: shutdownTimeout.asMilliseconds() });
this.log.debug(`http server stopped`);

10
src/core/server/http/http_service.ts
View file

@ -6,8 +6,8 @@
* Side Public License, v 1.
*/
import { Observable, Subscription, combineLatest } from 'rxjs';
import { first, map } from 'rxjs/operators';
import { Observable, Subscription, combineLatest, firstValueFrom } from 'rxjs';
import { map } from 'rxjs/operators';
import { pick } from '@kbn/std';
import type { RequestHandlerContext } from 'src/core/server';
@ -84,7 +84,7 @@ export class HttpService
public async preboot(deps: PrebootDeps): Promise<InternalHttpServicePreboot> {
this.log.debug('setting up preboot server');
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
const prebootSetup = await this.prebootServer.setup(config);
prebootSetup.server.route({
@ -147,7 +147,7 @@ export class HttpService
}
});
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
const { registerRouter, ...serverContract } = await this.httpServer.setup(
config,
@ -196,7 +196,7 @@ export class HttpService
}
public async start() {
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
if (this.shouldListen(config)) {
this.log.debug('stopping preboot server');
await this.prebootServer.stop();

9
src/core/server/http/router/request.ts
View file

@ -9,7 +9,7 @@
import { URL } from 'url';
import uuid from 'uuid';
import { Request, RouteOptionsApp, RequestApplicationState, RouteOptions } from '@hapi/hapi';
import { Observable, fromEvent } from 'rxjs';
import { Observable, fromEvent, NEVER } from 'rxjs';
import { shareReplay, first, filter } from 'rxjs/operators';
import { RecursiveReadonly } from '@kbn/utility-types';
import { deepFreeze } from '@kbn/std';
@ -216,6 +216,13 @@ export class KibanaRequest<
}
private getEvents(request: Request): KibanaRequestEvents {
if (!request.raw.res) {
return {
aborted$: NEVER,
completed$: NEVER,
};
}
const completed$ = fromEvent<void>(request.raw.res, 'close').pipe(shareReplay(1), first());
// the response's underlying connection was terminated prematurely
const aborted$ = completed$.pipe(filter(() => !isCompleted(request)));

9
src/core/server/i18n/i18n_service.ts
View file

@ -6,7 +6,7 @@
* Side Public License, v 1.
*/
import { take } from 'rxjs/operators';
import { firstValueFrom } from 'rxjs';
import { Logger } from '../logging';
import { IConfigService } from '../config';
import { CoreContext } from '../core_context';
@ -68,10 +68,9 @@ export class I18nService {
}
private async initTranslations(pluginPaths: string[]) {
const i18nConfig = await this.configService
.atPath<I18nConfigType>(i18nConfigDef.path)
.pipe(take(1))
.toPromise();
const i18nConfig = await firstValueFrom(
this.configService.atPath<I18nConfigType>(i18nConfigDef.path)
);
const locale = i18nConfig.locale;
this.log.debug(`Using locale: ${locale}`);

4
src/core/server/metrics/integration_tests/server_collector.test.ts
View file

@ -76,8 +76,8 @@ describe('ServerMetricsCollector', () => {
it('collect disconnects requests infos', async () => {
const never = new Promise((resolve) => undefined);
const disconnectRequested$ = new Subject(); // Controls the number of requests in the /disconnect endpoint
const disconnectAborted$ = new Subject(); // Controls the abort event in the /disconnect endpoint
const disconnectRequested$ = new Subject<void>(); // Controls the number of requests in the /disconnect endpoint
const disconnectAborted$ = new Subject<void>(); // Controls the abort event in the /disconnect endpoint
router.get({ path: '/', validate: false }, async (ctx, req, res) => {
return res.ok({ body: '' });

10
src/core/server/metrics/metrics_service.ts
View file

@ -6,8 +6,7 @@
* Side Public License, v 1.
*/
import { ReplaySubject } from 'rxjs';
import { first } from 'rxjs/operators';
import { firstValueFrom, ReplaySubject } from 'rxjs';
import { CoreService } from '../../types';
import { CoreContext } from '../core_context';
import { Logger } from '../logging';
@ -38,10 +37,9 @@ export class MetricsService
}
public async setup({ http }: MetricsServiceSetupDeps): Promise<InternalMetricsServiceSetup> {
const config = await this.coreContext.configService
.atPath<OpsConfigType>(opsConfig.path)
.pipe(first())
.toPromise();
const config = await firstValueFrom(
this.coreContext.configService.atPath<OpsConfigType>(opsConfig.path)
);
this.metricsCollector = new OpsMetricsCollector(http.server, {
logger: this.logger,

23
src/core/server/plugins/discovery/plugins_discovery.test.ts
View file

@ -13,8 +13,8 @@ import mockFs from 'mock-fs';
import { loggingSystemMock } from '../../logging/logging_system.mock';
import { getEnvOptions, rawConfigServiceMock } from '../../config/mocks';
import { from } from 'rxjs';
import { first, map, toArray } from 'rxjs/operators';
import { firstValueFrom, from } from 'rxjs';
import { map, toArray } from 'rxjs/operators';
import { resolve } from 'path';
import { ConfigService, Env } from '../../config';
import { PluginsConfig, PluginsConfigType, config } from '../plugins_config';
@ -163,10 +163,7 @@ describe('plugins discovery system', () => {
logger,
};
pluginConfig = await configService
.atPath<PluginsConfigType>('plugins')
.pipe(first())
.toPromise();
pluginConfig = await firstValueFrom(configService.atPath<PluginsConfigType>('plugins'));
// jest relies on the filesystem to get sourcemaps when using console.log
// which breaks with the mocked FS, see https://github.com/tschaub/mock-fs/issues/234
@ -195,7 +192,7 @@ describe('plugins discovery system', () => {
{ createCwd: false }
);
const plugins = await plugin$.pipe(toArray()).toPromise();
const plugins = await firstValueFrom(plugin$.pipe(toArray()));
const pluginNames = plugins.map((plugin) => plugin.name);
expect(pluginNames).toHaveLength(4);
@ -362,7 +359,7 @@ describe('plugins discovery system', () => {
{ createCwd: false }
);
const plugins = await plugin$.pipe(toArray()).toPromise();
const plugins = await firstValueFrom(plugin$.pipe(toArray()));
const pluginNames = plugins.map((plugin) => plugin.name);
expect(pluginNames).toHaveLength(4);
@ -401,7 +398,7 @@ describe('plugins discovery system', () => {
{ createCwd: false }
);
const plugins = await plugin$.pipe(toArray()).toPromise();
const plugins = await firstValueFrom(plugin$.pipe(toArray()));
const pluginNames = plugins.map((plugin) => plugin.name);
expect(pluginNames).toEqual(['pluginA']);
@ -423,7 +420,7 @@ describe('plugins discovery system', () => {
{ createCwd: false }
);
const plugins = await plugin$.pipe(toArray()).toPromise();
const plugins = await firstValueFrom(plugin$.pipe(toArray()));
const pluginNames = plugins.map((plugin) => plugin.name);
expect(pluginNames).toHaveLength(5);
@ -451,7 +448,7 @@ describe('plugins discovery system', () => {
{ createCwd: false }
);
const plugins = await plugin$.pipe(toArray()).toPromise();
const plugins = await firstValueFrom(plugin$.pipe(toArray()));
const pluginNames = plugins.map((plugin) => plugin.name);
expect(pluginNames).toHaveLength(3);
@ -536,7 +533,7 @@ describe('plugins discovery system', () => {
let { plugin$ } = discover(new PluginsConfig(pluginConfig, env), coreContext, instanceInfo);
expect(scanPluginSearchPathsMock).toHaveBeenCalledTimes(1);
let plugins = await plugin$.pipe(toArray()).toPromise();
let plugins = await firstValueFrom(plugin$.pipe(toArray()));
let pluginNames = plugins.map((plugin) => plugin.name);
expect(pluginNames).toHaveLength(3);
@ -555,7 +552,7 @@ describe('plugins discovery system', () => {
plugin$ = discover(new PluginsConfig(pluginConfig, env), coreContext, instanceInfo).plugin$;
expect(scanPluginSearchPathsMock).toHaveBeenCalledTimes(2);
plugins = await plugin$.pipe(toArray()).toPromise();
plugins = await firstValueFrom(plugin$.pipe(toArray()));
pluginNames = plugins.map((plugin) => plugin.name);
expect(pluginNames).toHaveLength(3);

15
src/core/server/plugins/discovery/scan_plugin_search_paths.test.ts
View file

@ -12,6 +12,7 @@ import { toArray } from 'rxjs/operators';
import { resolve } from 'path';
import { scanPluginSearchPaths } from './scan_plugin_search_paths';
import { PluginDiscoveryError } from './plugin_discovery_error';
import { firstValueFrom } from 'rxjs';
const KIBANA_ROOT = process.cwd();
@ -82,7 +83,7 @@ describe('scanPluginSearchPaths', () => {
);
const pluginOrErrors$ = scanPluginSearchPaths(scanPaths, logger);
const { paths, errors } = extract(await pluginOrErrors$.pipe(toArray()).toPromise());
const { paths, errors } = extract(await firstValueFrom(pluginOrErrors$.pipe(toArray())));
expect(paths).toHaveLength(3);
expect(errors).toHaveLength(0);
@ -106,7 +107,7 @@ describe('scanPluginSearchPaths', () => {
);
const pluginOrErrors$ = scanPluginSearchPaths([`${KIBANA_ROOT}/src/plugins`], logger);
const { paths, errors } = extract(await pluginOrErrors$.pipe(toArray()).toPromise());
const { paths, errors } = extract(await firstValueFrom(pluginOrErrors$.pipe(toArray())));
expect(paths).toHaveLength(1);
expect(errors).toHaveLength(0);
@ -128,7 +129,7 @@ describe('scanPluginSearchPaths', () => {
);
const pluginOrErrors$ = scanPluginSearchPaths([`${KIBANA_ROOT}/src/plugins`], logger);
const { paths, errors } = extract(await pluginOrErrors$.pipe(toArray()).toPromise());
const { paths, errors } = extract(await firstValueFrom(pluginOrErrors$.pipe(toArray())));
expect(paths).toHaveLength(0);
expect(errors).toHaveLength(1);
@ -153,7 +154,7 @@ describe('scanPluginSearchPaths', () => {
);
const pluginOrErrors$ = scanPluginSearchPaths([`${KIBANA_ROOT}/src/plugins`], logger);
const { paths, errors } = extract(await pluginOrErrors$.pipe(toArray()).toPromise());
const { paths, errors } = extract(await firstValueFrom(pluginOrErrors$.pipe(toArray())));
expect(errors).toHaveLength(0);
expect(paths).toHaveLength(4);
@ -180,7 +181,7 @@ describe('scanPluginSearchPaths', () => {
);
const pluginOrErrors$ = scanPluginSearchPaths([`${KIBANA_ROOT}/src/plugins`], logger);
const { paths, errors } = extract(await pluginOrErrors$.pipe(toArray()).toPromise());
const { paths, errors } = extract(await firstValueFrom(pluginOrErrors$.pipe(toArray())));
expect(errors).toHaveLength(0);
expect(paths).toHaveLength(1);
@ -203,7 +204,7 @@ describe('scanPluginSearchPaths', () => {
);
const pluginOrErrors$ = scanPluginSearchPaths([`${KIBANA_ROOT}/src/plugins`], logger);
const { paths, errors } = extract(await pluginOrErrors$.pipe(toArray()).toPromise());
const { paths, errors } = extract(await firstValueFrom(pluginOrErrors$.pipe(toArray())));
expect(errors).toHaveLength(0);
expect(paths).toHaveLength(5);
@ -236,7 +237,7 @@ describe('scanPluginSearchPaths', () => {
);
const pluginOrErrors$ = scanPluginSearchPaths([`${KIBANA_ROOT}/plugins`], logger);
const { paths, errors } = extract(await pluginOrErrors$.pipe(toArray()).toPromise());
const { paths, errors } = extract(await firstValueFrom(pluginOrErrors$.pipe(toArray())));
expect(errors).toHaveLength(0);
expect(paths).toHaveLength(2);

4
src/core/server/plugins/discovery/scan_plugin_search_paths.ts
View file

@ -13,8 +13,8 @@ import { catchError, mergeMap } from 'rxjs/operators';
import { Logger } from '../../logging';
import { PluginDiscoveryError } from './plugin_discovery_error';
const fsReadDir$ = bindNodeCallback<string, string[]>(readdir);
const fsStat$ = bindNodeCallback<PathLike, Stats>(stat);
const fsReadDir$ = bindNodeCallback<[string], [string[]]>(readdir);
const fsStat$ = bindNodeCallback<[PathLike], [Stats]>(stat);
const maxScanDepth = 5;

5
src/core/server/plugins/plugin.ts
View file

@ -8,8 +8,7 @@
import { join } from 'path';
import typeDetect from 'type-detect';
import { Subject } from 'rxjs';
import { first } from 'rxjs/operators';
import { firstValueFrom, Subject } from 'rxjs';
import { isPromise } from '@kbn/std';
import { isConfigSchema } from '@kbn/config-schema';
@ -63,7 +62,7 @@ export class PluginWrapper<
| AsyncPlugin<TSetup, TStart, TPluginsSetup, TPluginsStart>;
private readonly startDependencies$ = new Subject<[CoreStart, TPluginsStart, TStart]>();
public readonly startDependencies = this.startDependencies$.pipe(first()).toPromise();
public readonly startDependencies = firstValueFrom(this.startDependencies$);
constructor(
public readonly params: {

20
src/core/server/plugins/plugins_service.ts
View file

@ -7,8 +7,8 @@
*/
import Path from 'path';
import { Observable } from 'rxjs';
import { filter, first, map, tap, toArray } from 'rxjs/operators';
import { firstValueFrom, Observable } from 'rxjs';
import { filter, map, tap, toArray } from 'rxjs/operators';
import { getFlattenedObject } from '@kbn/std';
import { CoreService } from '../../types';
@ -111,7 +111,7 @@ export class PluginsService implements CoreService<PluginsServiceSetup, PluginsS
}
public async discover({ environment }: PluginsServiceDiscoverDeps): Promise<DiscoveredPlugins> {
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
const { error$, plugin$ } = discover(config, this.coreContext, {
uuid: environment.instanceUuid,
@ -151,7 +151,7 @@ export class PluginsService implements CoreService<PluginsServiceSetup, PluginsS
public async preboot(deps: PluginsServicePrebootSetupDeps) {
this.log.debug('Prebooting plugins service');
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
if (config.initialize) {
await this.prebootPluginsSystem.setupPlugins(deps);
this.registerPluginStaticDirs(deps, this.prebootUiPluginInternalInfo);
@ -165,7 +165,7 @@ export class PluginsService implements CoreService<PluginsServiceSetup, PluginsS
public async setup(deps: PluginsServiceSetupDeps) {
this.log.debug('Setting up plugins service');
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
let contracts = new Map<PluginName, unknown>();
if (config.initialize) {
@ -186,7 +186,7 @@ export class PluginsService implements CoreService<PluginsServiceSetup, PluginsS
public async start(deps: PluginsServiceStartDeps) {
this.log.debug('Plugins service starts plugins');
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
if (!config.initialize) {
this.log.info(
'Skipping `start` for `standard` plugins since plugin initialization is disabled.'
@ -246,13 +246,13 @@ export class PluginsService implements CoreService<PluginsServiceSetup, PluginsS
PluginDiscoveryErrorType.InvalidManifest,
];
const errors = await error$
.pipe(
const errors = await firstValueFrom(
error$.pipe(
filter((error) => errorTypesToReport.includes(error.type)),
tap((pluginError) => this.log.error(pluginError)),
toArray()
)
.toPromise();
);
if (errors.length > 0) {
throw new Error(
`Failed to initialize plugins:${errors.map((err) => `\n\t${err.message}`).join('')}`
@ -265,7 +265,7 @@ export class PluginsService implements CoreService<PluginsServiceSetup, PluginsS
PluginName,
{ plugin: PluginWrapper; isEnabled: boolean }
>();
const plugins = await plugin$.pipe(toArray()).toPromise();
const plugins = await firstValueFrom(plugin$.pipe(toArray()));
// Register config descriptors and deprecations
for (const plugin of plugins) {

3
src/core/server/saved_objects/migrations/kibana_migrator.test.ts
View file

@ -16,6 +16,7 @@ import { SavedObjectTypeRegistry } from '../saved_objects_type_registry';
import { SavedObjectsType } from '../types';
import { DocumentMigrator } from './core/document_migrator';
import { ByteSizeValue } from '@kbn/config-schema';
import { lastValueFrom } from 'rxjs';
jest.mock('./core/document_migrator', () => {
return {
@ -134,7 +135,7 @@ describe('KibanaMigrator', () => {
it('emits results on getMigratorResult$()', async () => {
const options = mockV2MigrationOptions();
const migrator = new KibanaMigrator(options);
const migratorStatus = migrator.getStatus$().pipe(take(3)).toPromise();
const migratorStatus = lastValueFrom(migrator.getStatus$().pipe(take(3)));
migrator.prepareMigrations();
await migrator.runMigrations();

20
src/core/server/saved_objects/saved_objects_service.ts
View file

@ -6,8 +6,8 @@
* Side Public License, v 1.
*/
import { Subject, Observable } from 'rxjs';
import { first, filter, take, switchMap } from 'rxjs/operators';
import { Subject, Observable, firstValueFrom } from 'rxjs';
import { filter, take, switchMap } from 'rxjs/operators';
import { CoreService } from '../../types';
import {
SavedObjectsClient,
@ -314,14 +314,12 @@ export class SavedObjectsService
this.setupDeps = setupDeps;
const { http, elasticsearch, coreUsageData, deprecations } = setupDeps;
const savedObjectsConfig = await this.coreContext.configService
.atPath<SavedObjectsConfigType>('savedObjects')
.pipe(first())
.toPromise();
const savedObjectsMigrationConfig = await this.coreContext.configService
.atPath<SavedObjectsMigrationConfigType>('migrations')
.pipe(first())
.toPromise();
const savedObjectsConfig = await firstValueFrom(
this.coreContext.configService.atPath<SavedObjectsConfigType>('savedObjects')
);
const savedObjectsMigrationConfig = await firstValueFrom(
this.coreContext.configService.atPath<SavedObjectsMigrationConfigType>('migrations')
);
this.config = new SavedObjectConfig(savedObjectsConfig, savedObjectsMigrationConfig);
deprecations.getRegistry('savedObjects').registerDeprecations(
@ -340,7 +338,7 @@ export class SavedObjectsService
coreUsageData,
logger: this.logger,
config: this.config,
migratorPromise: this.migrator$.pipe(first()).toPromise(),
migratorPromise: firstValueFrom(this.migrator$),
kibanaIndex,
kibanaVersion: this.kibanaVersion,
});

5
src/core/server/status/routes/integration_tests/status.test.ts
View file

@ -6,8 +6,7 @@
* Side Public License, v 1.
*/
import { BehaviorSubject } from 'rxjs';
import { first } from 'rxjs/operators';
import { BehaviorSubject, firstValueFrom } from 'rxjs';
import supertest from 'supertest';
import { omit } from 'lodash';
@ -144,7 +143,7 @@ describe('GET /api/status', () => {
build_number: 1234,
build_snapshot: true,
});
const metricsMockValue = await metrics.getOpsMetrics$().pipe(first()).toPromise();
const metricsMockValue = await firstValueFrom(metrics.getOpsMetrics$());
expect(result.body.metrics).toEqual({
last_updated: expect.any(String),
collection_interval_in_millis: metrics.collectionInterval,

7
src/core/server/status/routes/status.ts
View file

@ -6,8 +6,7 @@
* Side Public License, v 1.
*/
import { Observable, combineLatest, ReplaySubject } from 'rxjs';
import { first } from 'rxjs/operators';
import { Observable, combineLatest, ReplaySubject, firstValueFrom } from 'rxjs';
import { schema } from '@kbn/config-schema';
import { IRouter } from '../../http';
@ -92,7 +91,7 @@ export const registerStatusRoute = ({
async (context, req, res) => {
const { version, buildSha, buildNum } = config.packageInfo;
const versionWithoutSnapshot = version.replace(SNAPSHOT_POSTFIX, '');
const [overall, coreOverall, core, plugins] = await combinedStatus$.pipe(first()).toPromise();
const [overall, coreOverall, core, plugins] = await firstValueFrom(combinedStatus$);
const { v8format = true, v7format = false } = req.query ?? {};
@ -113,7 +112,7 @@ export const registerStatusRoute = ({
});
}
const lastMetrics = await metrics.getOpsMetrics$().pipe(first()).toPromise();
const lastMetrics = await firstValueFrom(metrics.getOpsMetrics$());
const body: StatusHttpBody = {
name: config.serverName,

6
src/core/server/status/status_service.ts
View file

@ -6,8 +6,8 @@
* Side Public License, v 1.
*/
import { Observable, combineLatest, Subscription, Subject } from 'rxjs';
import { map, distinctUntilChanged, shareReplay, take, debounceTime } from 'rxjs/operators';
import { Observable, combineLatest, Subscription, Subject, firstValueFrom } from 'rxjs';
import { map, distinctUntilChanged, shareReplay, debounceTime } from 'rxjs/operators';
import { isDeepStrictEqual } from 'util';
import { CoreService } from '../../types';
@ -65,7 +65,7 @@ export class StatusService implements CoreService<InternalStatusServiceSetup> {
environment,
coreUsageData,
}: SetupDeps) {
const statusConfig = await this.config$.pipe(take(1)).toPromise();
const statusConfig = await firstValueFrom(this.config$);
const core$ = this.setupCoreStatus({ elasticsearch, savedObjects });
this.pluginsStatus = new PluginsStatusService({ core$, pluginDependencies });

7
src/core/server/ui_settings/ui_settings_service.ts
View file

@ -6,8 +6,7 @@
* Side Public License, v 1.
*/
import { Observable } from 'rxjs';
import { first } from 'rxjs/operators';
import { firstValueFrom, Observable } from 'rxjs';
import { mapToObject } from '@kbn/std';
import { CoreService } from '../../types';
@ -53,7 +52,7 @@ export class UiSettingsService
public async preboot(): Promise<InternalUiSettingsServicePreboot> {
this.log.debug('Prebooting ui settings service');
const { overrides } = await this.config$.pipe(first()).toPromise();
const { overrides } = await firstValueFrom(this.config$);
this.overrides = overrides;
this.register(getCoreSettings({ isDist: this.isDist }));
@ -74,7 +73,7 @@ export class UiSettingsService
savedObjects.registerType(uiSettingsType);
registerRoutes(http.createRouter(''));
const config = await this.config$.pipe(first()).toPromise();
const config = await firstValueFrom(this.config$);
this.overrides = config.overrides;
return {

4
src/dev/build/lib/scan.ts
View file

@ -15,8 +15,8 @@ import { map, mergeAll, mergeMap } from 'rxjs/operators';
// @ts-ignore
import { assertAbsolute } from './fs';
const getStat$ = Rx.bindNodeCallback(Fs.stat) as (path: string) => Rx.Observable<Fs.Stats>;
const getReadDir$ = Rx.bindNodeCallback(Fs.readdir) as (path: string) => Rx.Observable<string[]>;
const getStat$ = Rx.bindNodeCallback<[string], [Fs.Stats]>(Fs.stat);
const getReadDir$ = Rx.bindNodeCallback<[string], [string[]]>(Fs.readdir);
/**
* Return an observable of all files in a directory, starting with the

4
src/dev/build/lib/scan_delete.ts
View file

@ -16,8 +16,8 @@ import { count, map, mergeAll, mergeMap } from 'rxjs/operators';
// @ts-ignore
import { assertAbsolute } from './fs';
const getStat$ = Rx.bindNodeCallback<Fs.PathLike, Fs.Stats>(Fs.stat);
const getReadDir$ = Rx.bindNodeCallback<string, string[]>(Fs.readdir);
const getStat$ = Rx.bindNodeCallback<[Fs.PathLike], [Fs.Stats]>(Fs.stat);
const getReadDir$ = Rx.bindNodeCallback<[string], [string[]]>(Fs.readdir);
interface Options {
directory: string;

2
src/dev/build/tasks/build_kibana_platform_plugins.ts
View file

@ -9,7 +9,7 @@
import Path from 'path';
import { REPO_ROOT } from '@kbn/utils';
import { lastValueFrom } from '@kbn/std';
import { lastValueFrom } from 'rxjs';
import { CiStatsMetric } from '@kbn/dev-utils';
import {
runOptimizer,

3
src/dev/eslint/run_eslint_with_types.ts
View file

@ -16,7 +16,6 @@ import { mergeMap, reduce } from 'rxjs/operators';
import { supportsColor } from 'chalk';
import { run, createFailError } from '@kbn/dev-utils';
import { REPO_ROOT } from '@kbn/utils';
import { lastValueFrom } from '@kbn/std';
import { PROJECTS } from '../typescript/projects';
import { Project } from '../typescript/project';
@ -61,7 +60,7 @@ export function runEslintWithTypes() {
const concurrency = Math.max(1, Math.round((Os.cpus() || []).length / 2) || 1) || 1;
log.info(`Linting ${projects.length} projects, ${concurrency} at a time`);
const failures = await lastValueFrom(
const failures = await Rx.lastValueFrom(
Rx.from(projects).pipe(
mergeMap(async (project) => {
const configFilePath = Path.resolve(project.directory, 'types.eslint.config.js');

3
src/dev/typescript/concurrent_map.ts
View file

@ -8,7 +8,6 @@
import * as Rx from 'rxjs';
import { mergeMap, toArray, map } from 'rxjs/operators';
import { lastValueFrom } from '@kbn/std';
export async function concurrentMap<T, T2>(
concurrency: number,
@ -19,7 +18,7 @@ export async function concurrentMap<T, T2>(
return [];
}
return await lastValueFrom(
return await Rx.lastValueFrom(
Rx.from(arr).pipe(
// execute items in parallel based on concurrency
mergeMap(async (item, index) => ({ index, result: await fn(item, index) }), concurrency),

3
src/dev/typescript/run_type_check_cli.ts
View file

@ -13,7 +13,6 @@ import * as Rx from 'rxjs';
import { mergeMap, reduce } from 'rxjs/operators';
import execa from 'execa';
import { run, createFailError } from '@kbn/dev-utils';
import { lastValueFrom } from '@kbn/std';
import { PROJECTS } from './projects';
import { buildTsRefs } from './build_ts_refs';
@ -65,7 +64,7 @@ export async function runTypeCheckCli() {
: ['--skipLibCheck', 'false']),
];
const failureCount = await lastValueFrom(
const failureCount = await Rx.lastValueFrom(
Rx.from(projects).pipe(
mergeMap(async (p) => {
const relativePath = Path.relative(process.cwd(), p.tsConfigPath);

2
src/plugins/charts/public/services/active_cursor/active_cursor.test.ts
View file

@ -22,8 +22,8 @@ describe('ActiveCursor', () => {
expect(activeCursor.activeCursor$).toMatchInlineSnapshot(`
Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],

5
src/plugins/console/server/services/es_legacy_config_service.ts
View file

@ -6,8 +6,7 @@
* Side Public License, v 1.
*/
import { Observable, Subscription } from 'rxjs';
import { first } from 'rxjs/operators';
import { firstValueFrom, Observable, Subscription } from 'rxjs';
import { ElasticsearchConfig } from 'kibana/server';
export class EsLegacyConfigService {
@ -45,7 +44,7 @@ export class EsLegacyConfigService {
}
if (!this.config) {
return this.config$.pipe(first()).toPromise();
return firstValueFrom(this.config$);
}
return this.config;

4
src/plugins/controls/public/control_group/embeddable/control_group_chaining_system.ts
View file

@ -59,7 +59,7 @@ export const ControlGroupChainingSystems: {
onChildChange: ({ childOutputChangedId, childOrder, recalculateFilters$, getChild }) => {
if (childOutputChangedId === childOrder.lastChildId) {
// the last control's output has updated, recalculate filters
recalculateFilters$.next();
recalculateFilters$.next(null);
return;
}
@ -75,6 +75,6 @@ export const ControlGroupChainingSystems: {
NONE: {
getContainerSettings: () => undefined,
getPrecedingFilters: () => undefined,
onChildChange: ({ recalculateFilters$ }) => recalculateFilters$.next(),
onChildChange: ({ recalculateFilters$ }) => recalculateFilters$.next(null),
},
};

2
src/plugins/controls/public/control_types/time_slider/time_slider_embeddable.tsx
View file

@ -151,7 +151,7 @@ export class TimeSliderControlEmbeddable extends Embeddable<
const min = value ? value[0] : null;
const max = value ? value[1] : null;
const hasRange =
this.componentState.range!.max !== undefined && this.componentState.range!.min !== undefined;
this.componentState.range?.max !== undefined && this.componentState.range?.min !== undefined;
this.getCurrentDataView$().subscribe((dataView) => {
const range: RangeFilterParams = {};

2
src/plugins/dashboard/public/application/hooks/use_dashboard_app_state.ts
View file

@ -310,7 +310,7 @@ export const useDashboardAppState = ({
* the last saved state on save.
*/
setLastSavedState(savedDashboardState);
dashboardBuildContext.$checkForUnsavedChanges.next();
dashboardBuildContext.$checkForUnsavedChanges.next(undefined);
const updateLastSavedState = () => {
setLastSavedState(
savedObjectToDashboardState({

2
src/plugins/dashboard/public/application/lib/sync_dashboard_container_input.ts
View file

@ -60,7 +60,7 @@ export const syncDashboardContainerInput = (
.getInput$()
.subscribe(() => applyContainerChangesToState(syncDashboardContainerProps))
);
subscriptions.add($onDashboardStateChange.subscribe(() => $triggerDashboardRefresh.next()));
subscriptions.add($onDashboardStateChange.subscribe(() => $triggerDashboardRefresh.next({})));
subscriptions.add(
getSessionURLObservable(history).subscribe(() => {
$triggerDashboardRefresh.next({ force: true });

7
src/plugins/dashboard/public/application/lib/sync_dashboard_filter_state.ts
View file

@ -103,13 +103,14 @@ export const syncDashboardFilterState = ({
.subscribe(() => applyFilters(queryString.getQuery(), filterManager.getFilters()));
const timeRefreshSubscription = merge(
...[timefilterService.getRefreshIntervalUpdate$(), timefilterService.getTimeUpdate$()]
timefilterService.getRefreshIntervalUpdate$(),
timefilterService.getTimeUpdate$()
).subscribe(() => {
$triggerDashboardRefresh.next();
$triggerDashboardRefresh.next({});
// manually check for unsaved changes here because the time range is not stored on the dashboardState,
// but it could trigger the unsaved changes badge.
$checkForUnsavedChanges.next();
$checkForUnsavedChanges.next(undefined);
});
const forceRefreshSubscription = timefilterService

42
src/plugins/dashboard/public/application/listing/__snapshots__/dashboard_listing.test.tsx.snap generated
View file

@ -25,9 +25,9 @@ exports[`after fetch When given a title that matches multiple dashboards, filter
application={
Object {
"applications$": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -39,10 +39,9 @@ exports[`after fetch When given a title that matches multiple dashboards, filter
"navLinks": Object {},
},
"currentAppId$": Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -129,7 +128,6 @@ exports[`after fetch When given a title that matches multiple dashboards, filter
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -175,9 +173,9 @@ exports[`after fetch initialFilter 1`] = `
application={
Object {
"applications$": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -189,10 +187,9 @@ exports[`after fetch initialFilter 1`] = `
"navLinks": Object {},
},
"currentAppId$": Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -279,7 +276,6 @@ exports[`after fetch initialFilter 1`] = `
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -324,9 +320,9 @@ exports[`after fetch renders all table rows 1`] = `
application={
Object {
"applications$": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -338,10 +334,9 @@ exports[`after fetch renders all table rows 1`] = `
"navLinks": Object {},
},
"currentAppId$": Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -428,7 +423,6 @@ exports[`after fetch renders all table rows 1`] = `
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -473,9 +467,9 @@ exports[`after fetch renders call to action when no dashboards exist 1`] = `
application={
Object {
"applications$": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -487,10 +481,9 @@ exports[`after fetch renders call to action when no dashboards exist 1`] = `
"navLinks": Object {},
},
"currentAppId$": Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -577,7 +570,6 @@ exports[`after fetch renders call to action when no dashboards exist 1`] = `
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -622,9 +614,9 @@ exports[`after fetch renders call to action with continue when no dashboards exi
application={
Object {
"applications$": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -636,10 +628,9 @@ exports[`after fetch renders call to action with continue when no dashboards exi
"navLinks": Object {},
},
"currentAppId$": Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -737,7 +728,6 @@ exports[`after fetch renders call to action with continue when no dashboards exi
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -782,9 +772,9 @@ exports[`after fetch renders warning when listingLimit is exceeded 1`] = `
application={
Object {
"applications$": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -796,10 +786,9 @@ exports[`after fetch renders warning when listingLimit is exceeded 1`] = `
"navLinks": Object {},
},
"currentAppId$": Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -886,7 +875,6 @@ exports[`after fetch renders warning when listingLimit is exceeded 1`] = `
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}
@ -931,9 +919,9 @@ exports[`after fetch showWriteControls 1`] = `
application={
Object {
"applications$": BehaviorSubject {
"_isScalar": false,
"_value": Map {},
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -945,10 +933,9 @@ exports[`after fetch showWriteControls 1`] = `
"navLinks": Object {},
},
"currentAppId$": Observable {
"_isScalar": false,
"source": Subject {
"_isScalar": false,
"closed": false,
"currentObservers": null,
"hasError": false,
"isStopped": false,
"observers": Array [],
@ -1005,7 +992,6 @@ exports[`after fetch showWriteControls 1`] = `
theme={
Object {
"theme$": Observable {
"_isScalar": false,
"_subscribe": [Function],
},
}

7
src/plugins/data/common/search/aggs/buckets/_terms_other_bucket_helper.ts
View file

@ -15,6 +15,7 @@ import {
buildQueryFromFilters,
Filter,
} from '@kbn/es-query';
import { lastValueFrom } from 'rxjs';
import { AggGroupNames } from '../agg_groups';
import { IAggConfigs } from '../agg_configs';
import { IAggType } from '../agg_type';
@ -342,8 +343,8 @@ export const createOtherBucketPostFlightRequest = (
nestedSearchSource.setField('aggs', filterAgg);
const { rawResponse: response } = await nestedSearchSource
.fetch$({
const { rawResponse: response } = await lastValueFrom(
nestedSearchSource.fetch$({
abortSignal,
sessionId: searchSessionId,
inspector: {
@ -358,7 +359,7 @@ export const createOtherBucketPostFlightRequest = (
}),
},
})
.toPromise();
);
resp = mergeOtherBucketAggResponse(
aggConfigs,

19
src/plugins/data/common/search/expressions/eql.ts
View file

@ -11,6 +11,7 @@ import { buildEsQuery } from '@kbn/es-query';
import { ExpressionFunctionDefinition } from 'src/plugins/expressions/common';
import { EqlSearchRequest } from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
import { lastValueFrom } from 'rxjs';
import { RequestStatistics, RequestAdapter } from '../../../../inspector/common';
import {
ISearchGeneric,
@ -143,15 +144,17 @@ export const getEqlFn = ({
});
try {
const response = await search<EqlSearchStrategyRequest, EqlSearchStrategyResponse>(
{
params: {
index: args.index,
body: dsl,
const response = await lastValueFrom(
search<EqlSearchStrategyRequest, EqlSearchStrategyResponse>(
{
params: {
index: args.index,
body: dsl,
},
},
},
{ abortSignal, strategy: EQL_SEARCH_STRATEGY }
).toPromise();
{ abortSignal, strategy: EQL_SEARCH_STRATEGY }
)
);
const stats: RequestStatistics = {};

21
src/plugins/data/common/search/expressions/esdsl.ts
View file

@ -10,6 +10,7 @@ import { i18n } from '@kbn/i18n';
import { buildEsQuery } from '@kbn/es-query';
import { ExpressionFunctionDefinition } from 'src/plugins/expressions/common';
import { lastValueFrom } from 'rxjs';
import { EsRawResponse } from './es_raw_response';
import { RequestStatistics, RequestAdapter } from '../../../../inspector/common';
import { ISearchGeneric, KibanaContext } from '..';
@ -125,16 +126,18 @@ export const getEsdslFn = ({
});
try {
const { rawResponse } = await search(
{
params: {
index: args.index,
size: args.size,
body: dsl,
const { rawResponse } = await lastValueFrom(
search(
{
params: {
index: args.index,
size: args.size,
body: dsl,
},
},
},
{ abortSignal }
).toPromise();
{ abortSignal }
)
);
const stats: RequestStatistics = {};

4
src/plugins/data/common/search/search_source/search_source.test.ts
View file

@ -6,7 +6,7 @@
* Side Public License, v 1.
*/
import { of, throwError } from 'rxjs';
import { lastValueFrom, of, throwError } from 'rxjs';
import { IndexPattern } from '../..';
import { SearchSource, SearchSourceDependencies, SortDirection } from './';
import { AggConfigs, AggTypesRegistryStart } from '../../';
@ -1240,7 +1240,7 @@ describe('SearchSource', () => {
const fetch$ = searchSource.fetch$({});
fetch$.subscribe(fetchSub);
const resp = await fetch$.toPromise();
const resp = await lastValueFrom(fetch$);
expect(fetchSub.next).toHaveBeenCalledTimes(3);
expect(fetchSub.complete).toHaveBeenCalledTimes(1);

10
src/plugins/data/common/search/search_source/search_source.ts
View file

@ -70,7 +70,7 @@ import {
switchMap,
tap,
} from 'rxjs/operators';
import { defer, EMPTY, from, Observable } from 'rxjs';
import { defer, EMPTY, from, lastValueFrom, Observable } from 'rxjs';
import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
import { buildEsQuery, Filter } from '@kbn/es-query';
import { normalizeSortRequest } from './normalize_sort_request';
@ -325,11 +325,9 @@ export class SearchSource {
* @removeBy 8.1
*/
fetch(options: ISearchOptions = {}) {
return this.fetch$(options)
.toPromise()
.then((r) => {
return r.rawResponse as estypes.SearchResponse<any>;
});
return lastValueFrom(this.fetch$(options)).then((r) => {
return r.rawResponse as estypes.SearchResponse<any>;
});
}
/**

6
src/plugins/data/public/query/timefilter/timefilter.ts
View file

@ -32,10 +32,10 @@ export class Timefilter {
// Fired when isTimeRangeSelectorEnabled \ isAutoRefreshSelectorEnabled are toggled
private enabledUpdated$ = new BehaviorSubject(false);
// Fired when a user changes the timerange
private timeUpdate$ = new Subject();
private timeUpdate$ = new Subject<void>();
// Fired when a user changes the the autorefresh settings
private refreshIntervalUpdate$ = new Subject();
private fetch$ = new Subject();
private refreshIntervalUpdate$ = new Subject<void>();
private fetch$ = new Subject<void>();
private _time: TimeRange;
// Denotes whether setTime has been called, can be used to determine if the constructor defaults are being used.

2
src/plugins/data/public/query/timefilter/timefilter_service.mock.ts
View file

@ -23,7 +23,7 @@ const createSetupContractMock = () => {
getTimeUpdate$: jest.fn(),
getRefreshIntervalUpdate$: jest.fn(),
getAutoRefreshFetch$: jest.fn(() => new Observable<() => void>()),
getFetch$: jest.fn(() => new Observable<() => void>()),
getFetch$: jest.fn().mockImplementation(() => new Observable<() => void>()),
getTime: jest.fn(),
setTime: jest.fn(),
setRefreshInterval: jest.fn(),

42
src/plugins/data/public/search/session/session_helpers.test.ts
View file

@ -9,6 +9,7 @@
import { waitUntilNextSessionCompletes$ } from './session_helpers';
import { ISessionService, SessionService } from './session_service';
import { BehaviorSubject } from 'rxjs';
import { fakeSchedulers } from 'rxjs-marbles/jest';
import { SearchSessionState } from './search_session_state';
import { NowProviderInternalContract } from '../../now_provider';
import { coreMock } from '../../../../../core/public/mocks';
@ -62,27 +63,30 @@ describe('waitUntilNextSessionCompletes$', () => {
afterEach(() => {
jest.useRealTimers();
});
test('emits when next session starts', () => {
sessionService.start();
let untrackSearch = sessionService.trackSearch({ abort: () => {} });
untrackSearch();
test(
'emits when next session starts',
fakeSchedulers((advance) => {
sessionService.start();
let untrackSearch = sessionService.trackSearch({ abort: () => {} });
untrackSearch();
const next = jest.fn();
const complete = jest.fn();
waitUntilNextSessionCompletes$(sessionService).subscribe({ next, complete });
expect(next).not.toBeCalled();
const next = jest.fn();
const complete = jest.fn();
waitUntilNextSessionCompletes$(sessionService).subscribe({ next, complete });
expect(next).not.toBeCalled();
sessionService.start();
expect(next).not.toBeCalled();
sessionService.start();
expect(next).not.toBeCalled();
untrackSearch = sessionService.trackSearch({ abort: () => {} });
untrackSearch();
untrackSearch = sessionService.trackSearch({ abort: () => {} });
untrackSearch();
expect(next).not.toBeCalled();
jest.advanceTimersByTime(500);
expect(next).not.toBeCalled();
jest.advanceTimersByTime(1000);
expect(next).toBeCalledTimes(1);
expect(complete).toBeCalled();
});
expect(next).not.toBeCalled();
advance(500);
expect(next).not.toBeCalled();
advance(1000);
expect(next).toBeCalledTimes(1);
expect(complete).toBeCalled();
})
);
});

5
src/plugins/data/server/autocomplete/value_suggestions_route.ts
View file

@ -8,8 +8,7 @@
import { schema } from '@kbn/config-schema';
import { IRouter } from 'kibana/server';
import { Observable } from 'rxjs';
import { first } from 'rxjs/operators';
import { firstValueFrom, Observable } from 'rxjs';
import { getRequestAbortedSignal } from '../lib';
import { getKbnServerError, reportServerError } from '../../../kibana_utils/server';
import type { ConfigSchema } from '../../config';
@ -42,7 +41,7 @@ export function registerValueSuggestionsRoute(router: IRouter, config$: Observab
},
},
async (context, request, response) => {
const config = await config$.pipe(first()).toPromise();
const config = await firstValueFrom(config$);
const { field: fieldName, query, filters, fieldMeta, method } = request.body;
const { index } = request.params;
const abortSignal = getRequestAbortedSignal(request.events.aborted$);

11
src/plugins/data/server/search/routes/bsearch.ts
View file

@ -6,7 +6,8 @@
* Side Public License, v 1.
*/
import { catchError, first } from 'rxjs/operators';
import { firstValueFrom } from 'rxjs';
import { catchError } from 'rxjs/operators';
import { BfetchServerSetup } from 'src/plugins/bfetch/server';
import type { ExecutionContextSetup } from 'src/core/server';
import apm from 'elastic-apm-node';
@ -37,10 +38,8 @@ export function registerBsearchRoute(
return executionContextService.withContext(executionContext, () => {
apm.addLabels(executionContextService.getAsLabels());
return search
.search(requestData, restOptions)
.pipe(
first(),
return firstValueFrom(
search.search(requestData, restOptions).pipe(
catchError((err) => {
// Re-throw as object, to get attributes passed to the client
// eslint-disable-next-line no-throw-literal
@ -51,7 +50,7 @@ export function registerBsearchRoute(
};
})
)
.toPromise();
);
});
},
};

4
src/plugins/data/server/search/search_service.test.ts
View file

@ -16,7 +16,7 @@ import { createIndexPatternsStartMock } from '../data_views/mocks';
import { SearchService, SearchServiceSetupDependencies } from './search_service';
import { bfetchPluginMock } from '../../../bfetch/server/mocks';
import { of } from 'rxjs';
import { lastValueFrom, of } from 'rxjs';
import type {
IEsSearchRequest,
IEsSearchResponse,
@ -182,7 +182,7 @@ describe('Search service', () => {
throw new NoSearchIdInSessionError();
});
const res = await mockScopedClient.search(searchRequest, options).toPromise();
const res = await lastValueFrom(mockScopedClient.search(searchRequest, options));
const [request, callOptions] = mockStrategy.search.mock.calls[0];
expect(callOptions).toBe(options);

20
src/plugins/data/server/search/search_service.ts
View file

@ -6,7 +6,7 @@
* Side Public License, v 1.
*/
import { from, Observable, throwError } from 'rxjs';
import { firstValueFrom, from, Observable, throwError } from 'rxjs';
import { pick } from 'lodash';
import moment from 'moment';
import {
@ -19,7 +19,7 @@ import {
SharedGlobalConfig,
StartServicesAccessor,
} from 'src/core/server';
import { first, map, switchMap, tap, withLatestFrom } from 'rxjs/operators';
import { map, switchMap, tap, withLatestFrom } from 'rxjs/operators';
import { BfetchServerSetup } from 'src/plugins/bfetch/server';
import { ExpressionsServerSetup } from 'src/plugins/expressions/server';
import type {
@ -221,16 +221,12 @@ export class SearchService implements Plugin<ISearchSetup, ISearchStart> {
const aggs = this.aggsService.setup({ registerFunction: expressions.registerFunction });
this.initializerContext.config
.create<ConfigSchema>()
.pipe(first())
.toPromise()
.then((value) => {
if (value.search.aggs.shardDelay.enabled) {
aggs.types.registerBucket(SHARD_DELAY_AGG_NAME, getShardDelayBucketAgg);
expressions.registerFunction(aggShardDelay);
}
});
firstValueFrom(this.initializerContext.config.create<ConfigSchema>()).then((value) => {
if (value.search.aggs.shardDelay.enabled) {
aggs.types.registerBucket(SHARD_DELAY_AGG_NAME, getShardDelayBucketAgg);
expressions.registerFunction(aggShardDelay);
}
});
return {
__enhance: (enhancements: SearchEnhancements) => {

6
src/plugins/data/server/search/strategies/es_search/es_search_strategy.ts
View file

@ -6,8 +6,8 @@
* Side Public License, v 1.
*/
import { from, Observable } from 'rxjs';
import { first, tap } from 'rxjs/operators';
import { firstValueFrom, from, Observable } from 'rxjs';
import { tap } from 'rxjs/operators';
import type { Logger, SharedGlobalConfig } from 'kibana/server';
import type { ISearchStrategy } from '../../types';
import type { SearchUsage } from '../../collectors';
@ -37,7 +37,7 @@ export const esSearchStrategyProvider = (
const search = async () => {
try {
const config = await config$.pipe(first()).toPromise();
const config = await firstValueFrom(config$);
// @ts-expect-error params fall back to any, but should be valid SearchRequest params
const { terminateAfter, ...requestParams } = request.params ?? {};
const params = {

Some files were not shown because too many files have changed in this diff Show more