maxmustermann/kibana
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Refactor execution service to use observables underneath (#96065)

Browse source 
* Refactor execution service to use observables underneath
* Fix canvas plugin to initialize workspace after assets
* Update expression functions implementations to resolve observables instead of promises
This commit is contained in:
Michael Dokolin 2021-04-22 20:36:25 +02:00 committed by GitHub
parent 6bb289368b
commit 49a18483d3
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
45 changed files with 916 additions and 578 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.execution.interpret.md
View file

@ -7,7 +7,7 @@
<b>Signature:</b>
```typescript
interpret<T>(ast: ExpressionAstNode, input: T): Promise<unknown>;
interpret<T>(ast: ExpressionAstNode, input: T): Observable<unknown>;
```
## Parameters
@ -19,5 +19,5 @@ interpret<T>(ast: ExpressionAstNode, input: T): Promise<unknown>;
<b>Returns:</b>
`Promise<unknown>`
`Observable<unknown>`

4
docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.execution.invokechain.md
View file

@ -7,7 +7,7 @@
<b>Signature:</b>
```typescript
invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Promise<any>;
invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Observable<any>;
```
## Parameters
@ -19,5 +19,5 @@ invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Promise<any>;
<b>Returns:</b>
`Promise<any>`
`Observable<any>`

4
docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.execution.invokefunction.md
View file

@ -7,7 +7,7 @@
<b>Signature:</b>
```typescript
invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unknown>): Promise<any>;
invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unknown>): Observable<any>;
```
## Parameters
@ -20,5 +20,5 @@ invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unkn
<b>Returns:</b>
`Promise<any>`
`Observable<any>`

2
docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.execution.md
View file

@ -26,7 +26,7 @@ export declare class Execution<Input = unknown, Output = unknown, InspectorAdapt
| [expression](./kibana-plugin-plugins-expressions-public.execution.expression.md) | | <code>string</code> | |
| [input](./kibana-plugin-plugins-expressions-public.execution.input.md) | | <code>Input</code> | Initial input of the execution.<!-- -->N.B. It is initialized to <code>null</code> rather than <code>undefined</code> for legacy reasons, because in legacy interpreter it was set to <code>null</code> by default. |
| [inspectorAdapters](./kibana-plugin-plugins-expressions-public.execution.inspectoradapters.md) | | <code>InspectorAdapters</code> | |
| [result](./kibana-plugin-plugins-expressions-public.execution.result.md) | | <code>Promise&lt;Output &#124; ExpressionValueError&gt;</code> | |
| [result](./kibana-plugin-plugins-expressions-public.execution.result.md) | | <code>Observable&lt;Output &#124; ExpressionValueError&gt;</code> | Future that tracks result or error of this execution. |
| [state](./kibana-plugin-plugins-expressions-public.execution.state.md) | | <code>ExecutionContainer&lt;Output &#124; ExpressionValueError&gt;</code> | Dynamic state of the execution. |
## Methods

4
docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.execution.resolveargs.md
View file

@ -7,7 +7,7 @@
<b>Signature:</b>
```typescript
resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Promise<any>;
resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Observable<any>;
```
## Parameters
@ -20,5 +20,5 @@ resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Promise<an
<b>Returns:</b>
`Promise<any>`
`Observable<any>`

4
docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.execution.result.md
View file

@ -4,8 +4,10 @@
## Execution.result property
Future that tracks result or error of this execution.
<b>Signature:</b>
```typescript
get result(): Promise<Output | ExpressionValueError>;
readonly result: Observable<Output | ExpressionValueError>;
```

4
docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.execution.start.md
View file

@ -11,7 +11,7 @@ N.B. `input` is initialized to `null` rather than `undefined` for legacy reasons
<b>Signature:</b>
```typescript
start(input?: Input): void;
start(input?: Input): Observable<Output | ExpressionValueError>;
```
## Parameters
@ -22,5 +22,5 @@ start(input?: Input): void;
<b>Returns:</b>
`void`
`Observable<Output | ExpressionValueError>`

4
docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.executor.run.md
View file

@ -9,7 +9,7 @@ Execute expression and return result.
<b>Signature:</b>
```typescript
run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?: ExpressionExecutionParams): Promise<Output>;
run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?: ExpressionExecutionParams): Observable<Output | ExpressionValueError>;
```
## Parameters
@ -22,5 +22,5 @@ run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?:
<b>Returns:</b>
`Promise<Output>`
`Observable<Output | ExpressionValueError>`

2
docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.typestring.md
View file

@ -11,5 +11,5 @@ If the type extends a Promise, we still need to return the string representation
<b>Signature:</b>
```typescript
export declare type TypeString<T> = KnownTypeToString<UnwrapPromiseOrReturn<T>>;
export declare type TypeString<T> = KnownTypeToString<T extends ObservableLike<any> ? UnwrapObservable<T> : UnwrapPromiseOrReturn<T>>;
```

4
docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.execution.interpret.md
View file

@ -7,7 +7,7 @@
<b>Signature:</b>
```typescript
interpret<T>(ast: ExpressionAstNode, input: T): Promise<unknown>;
interpret<T>(ast: ExpressionAstNode, input: T): Observable<unknown>;
```
## Parameters
@ -19,5 +19,5 @@ interpret<T>(ast: ExpressionAstNode, input: T): Promise<unknown>;
<b>Returns:</b>
`Promise<unknown>`
`Observable<unknown>`

4
docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.execution.invokechain.md
View file

@ -7,7 +7,7 @@
<b>Signature:</b>
```typescript
invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Promise<any>;
invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Observable<any>;
```
## Parameters
@ -19,5 +19,5 @@ invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Promise<any>;
<b>Returns:</b>
`Promise<any>`
`Observable<any>`

4
docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.execution.invokefunction.md
View file

@ -7,7 +7,7 @@
<b>Signature:</b>
```typescript
invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unknown>): Promise<any>;
invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unknown>): Observable<any>;
```
## Parameters
@ -20,5 +20,5 @@ invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unkn
<b>Returns:</b>
`Promise<any>`
`Observable<any>`

2
docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.execution.md
View file

@ -26,7 +26,7 @@ export declare class Execution<Input = unknown, Output = unknown, InspectorAdapt
| [expression](./kibana-plugin-plugins-expressions-server.execution.expression.md) | | <code>string</code> | |
| [input](./kibana-plugin-plugins-expressions-server.execution.input.md) | | <code>Input</code> | Initial input of the execution.<!-- -->N.B. It is initialized to <code>null</code> rather than <code>undefined</code> for legacy reasons, because in legacy interpreter it was set to <code>null</code> by default. |
| [inspectorAdapters](./kibana-plugin-plugins-expressions-server.execution.inspectoradapters.md) | | <code>InspectorAdapters</code> | |
| [result](./kibana-plugin-plugins-expressions-server.execution.result.md) | | <code>Promise&lt;Output &#124; ExpressionValueError&gt;</code> | |
| [result](./kibana-plugin-plugins-expressions-server.execution.result.md) | | <code>Observable&lt;Output &#124; ExpressionValueError&gt;</code> | Future that tracks result or error of this execution. |
| [state](./kibana-plugin-plugins-expressions-server.execution.state.md) | | <code>ExecutionContainer&lt;Output &#124; ExpressionValueError&gt;</code> | Dynamic state of the execution. |
## Methods

4
docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.execution.resolveargs.md
View file

@ -7,7 +7,7 @@
<b>Signature:</b>
```typescript
resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Promise<any>;
resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Observable<any>;
```
## Parameters
@ -20,5 +20,5 @@ resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Promise<an
<b>Returns:</b>
`Promise<any>`
`Observable<any>`

4
docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.execution.result.md
View file

@ -4,8 +4,10 @@
## Execution.result property
Future that tracks result or error of this execution.
<b>Signature:</b>
```typescript
get result(): Promise<Output | ExpressionValueError>;
readonly result: Observable<Output | ExpressionValueError>;
```

4
docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.execution.start.md
View file

@ -11,7 +11,7 @@ N.B. `input` is initialized to `null` rather than `undefined` for legacy reasons
<b>Signature:</b>
```typescript
start(input?: Input): void;
start(input?: Input): Observable<Output | ExpressionValueError>;
```
## Parameters
@ -22,5 +22,5 @@ start(input?: Input): void;
<b>Returns:</b>
`void`
`Observable<Output | ExpressionValueError>`

4
docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.executor.run.md
View file

@ -9,7 +9,7 @@ Execute expression and return result.
<b>Signature:</b>
```typescript
run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?: ExpressionExecutionParams): Promise<Output>;
run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?: ExpressionExecutionParams): Observable<Output | ExpressionValueError>;
```
## Parameters
@ -22,5 +22,5 @@ run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?:
<b>Returns:</b>
`Promise<Output>`
`Observable<Output | ExpressionValueError>`

2
docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.typestring.md
View file

@ -11,5 +11,5 @@ If the type extends a Promise, we still need to return the string representation
<b>Signature:</b>
```typescript
export declare type TypeString<T> = KnownTypeToString<UnwrapPromiseOrReturn<T>>;
export declare type TypeString<T> = KnownTypeToString<T extends ObservableLike<any> ? UnwrapObservable<T> : UnwrapPromiseOrReturn<T>>;
```

11
src/plugins/expressions/common/execution/execution.abortion.test.ts
View file

@ -6,6 +6,7 @@
* Side Public License, v 1.
*/
import { first } from 'rxjs/operators';
import { waitFor } from '@testing-library/react';
import { Execution } from './execution';
import { parseExpression } from '../ast';
@ -39,7 +40,7 @@ describe('Execution abortion tests', () => {
execution.start();
execution.cancel();
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toMatchObject({
type: 'error',
@ -57,7 +58,7 @@ describe('Execution abortion tests', () => {
jest.advanceTimersByTime(100);
execution.cancel();
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toMatchObject({
type: 'error',
@ -75,7 +76,7 @@ describe('Execution abortion tests', () => {
execution.start();
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
execution.cancel();
@ -130,12 +131,12 @@ describe('Execution abortion tests', () => {
params: {},
});
execution.start();
execution.start().toPromise();
await waitFor(() => expect(started).toHaveBeenCalledTimes(1));
execution.cancel();
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toMatchObject({
type: 'error',
error: {

314
src/plugins/expressions/common/execution/execution.test.ts
View file

@ -6,6 +6,9 @@
* Side Public License, v 1.
*/
import { of } from 'rxjs';
import { first, scan } from 'rxjs/operators';
import { TestScheduler } from 'rxjs/testing';
import { Execution } from './execution';
import { parseExpression, ExpressionAstExpression } from '../ast';
import { createUnitTestExecutor } from '../test_helpers';
@ -42,10 +45,18 @@ const run = async (
) => {
const execution = createExecution(expression, context);
execution.start(input);
return await execution.result;
return await execution.result.pipe(first()).toPromise();
};
let testScheduler: TestScheduler;
describe('Execution', () => {
beforeEach(() => {
testScheduler = new TestScheduler((actual, expected) => {
return expect(actual).toStrictEqual(expected);
});
});
test('can instantiate', () => {
const execution = createExecution('foo bar=123');
expect(execution.state.get().ast.chain[0].arguments.bar).toEqual([123]);
@ -73,7 +84,7 @@ describe('Execution', () => {
/* eslint-enable no-console */
execution.start(123);
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toBe(123);
expect(spy).toHaveBeenCalledTimes(1);
@ -91,7 +102,7 @@ describe('Execution', () => {
value: -1,
});
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toEqual({
type: 'num',
@ -106,7 +117,7 @@ describe('Execution', () => {
value: 0,
});
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toEqual({
type: 'num',
@ -114,16 +125,102 @@ describe('Execution', () => {
});
});
test('casts input to correct type', async () => {
const execution = createExecution('add val=1');
describe('.input', () => {
test('casts input to correct type', async () => {
const execution = createExecution('add val=1');
// Below 1 is cast to { type: 'num', value: 1 }.
execution.start(1);
const result = await execution.result;
// Below 1 is cast to { type: 'num', value: 1 }.
execution.start(1);
const result = await execution.result.pipe(first()).toPromise();
expect(result).toEqual({
type: 'num',
value: 2,
expect(result).toEqual({
type: 'num',
value: 2,
});
});
test('supports promises on input', async () => {
const execution = createExecution('add val=1');
execution.start(Promise.resolve(1));
const result = await execution.result.pipe(first()).toPromise();
expect(result).toEqual({
type: 'num',
value: 2,
});
});
test('supports observables on input', async () => {
const execution = createExecution('add val=1');
execution.start(of(1));
const result = await execution.result.pipe(first()).toPromise();
expect(result).toEqual({
type: 'num',
value: 2,
});
});
test('handles observables on input', () => {
const execution = createExecution('add val=1');
testScheduler.run(({ cold, expectObservable }) => {
const input = cold(' -a--b-c-', { a: 1, b: 2, c: 3 });
const subscription = ' ---^---!';
const expected = ' ---ab-c-';
expectObservable(execution.start(input), subscription).toBe(expected, {
a: { type: 'num', value: 2 },
b: { type: 'num', value: 3 },
c: { type: 'num', value: 4 },
});
});
});
test('stops when input errors', () => {
const execution = createExecution('add val=1');
testScheduler.run(({ cold, expectObservable }) => {
const input = cold('-a-#-b-', { a: 1, b: 2 });
const expected = ' -a-#';
expectObservable(execution.start(input)).toBe(expected, {
a: { type: 'num', value: 2 },
});
});
});
test('does not complete when input completes', () => {
const execution = createExecution('add val=1');
testScheduler.run(({ cold, expectObservable }) => {
const input = cold('-a-b|', { a: 1, b: 2 });
const expected = ' -a-b-';
expectObservable(execution.start(input)).toBe(expected, {
a: { type: 'num', value: 2 },
b: { type: 'num', value: 3 },
});
});
});
test('handles partial results', () => {
const execution = createExecution('sum');
testScheduler.run(({ cold, expectObservable }) => {
const items = cold(' -a--b-c-', { a: 1, b: 2, c: 3 });
const subscription = ' ---^---!';
const expected = ' ---ab-c-';
const input = items.pipe(scan((result, value) => [...result, value], new Array<number>()));
expectObservable(execution.start(input), subscription).toBe(expected, {
a: { type: 'num', value: 1 },
b: { type: 'num', value: 3 },
c: { type: 'num', value: 6 },
});
});
});
});
@ -251,7 +348,7 @@ describe('Execution', () => {
value: 0,
});
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toEqual({
type: 'num',
@ -267,13 +364,32 @@ describe('Execution', () => {
test('result is undefined until execution completes', async () => {
const execution = createExecution('sleep 10');
expect(execution.state.get().result).toBe(undefined);
execution.start(null);
execution.start(null).subscribe(jest.fn());
expect(execution.state.get().result).toBe(undefined);
await new Promise((r) => setTimeout(r, 1));
expect(execution.state.get().result).toBe(undefined);
await new Promise((r) => setTimeout(r, 11));
expect(execution.state.get().result).toBe(null);
});
test('handles functions returning observables', () => {
testScheduler.run(({ cold, expectObservable }) => {
const arg = cold(' -a-b-c|', { a: 1, b: 2, c: 3 });
const expected = ' -a-b-c-';
const observable: ExpressionFunctionDefinition<'observable', any, {}, any> = {
name: 'observable',
args: {},
help: '',
fn: () => arg,
};
const executor = createUnitTestExecutor();
executor.registerFunction(observable);
const result = executor.run('observable', null, {});
expectObservable(result).toBe(expected, { a: 1, b: 2, c: 3 });
});
});
});
describe('when function throws', () => {
@ -309,7 +425,7 @@ describe('Execution', () => {
const execution = await createExecution('error "foo"');
execution.start(null);
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toMatchObject({
type: 'error',
@ -330,7 +446,7 @@ describe('Execution', () => {
const executor = createUnitTestExecutor();
executor.registerFunction(spy);
await executor.run('error "..." | spy', null);
await executor.run('error "..." | spy', null).pipe(first()).toPromise();
expect(spy.fn).toHaveBeenCalledTimes(0);
});
@ -360,14 +476,14 @@ describe('Execution', () => {
test('execution state is "result" when execution successfully completes', async () => {
const execution = createExecution('sleep 1');
execution.start(null);
await execution.result;
await execution.result.pipe(first()).toPromise();
expect(execution.state.get().state).toBe('result');
});
test('execution state is "result" when execution successfully completes - 2', async () => {
const execution = createExecution('var foo');
execution.start(null);
await execution.result;
await execution.result.pipe(first()).toPromise();
expect(execution.state.get().state).toBe('result');
});
});
@ -413,10 +529,142 @@ describe('Execution', () => {
expect(result).toBe(66);
});
test('supports observables in arguments', () => {
const observable = {
name: 'observable',
args: {},
help: '',
fn: () => of(1),
};
const executor = createUnitTestExecutor();
executor.registerFunction(observable);
expect(
executor.run('add val={observable}', 1, {}).pipe(first()).toPromise()
).resolves.toEqual({
type: 'num',
value: 2,
});
});
test('supports observables in arguments emitting multiple values', () => {
testScheduler.run(({ cold, expectObservable }) => {
const arg = cold('-a-b-c-', { a: 1, b: 2, c: 3 });
const expected = '-a-b-c-';
const observable = {
name: 'observable',
args: {},
help: '',
fn: () => arg,
};
const executor = createUnitTestExecutor();
executor.registerFunction(observable);
const result = executor.run('add val={observable}', 1, {});
expectObservable(result).toBe(expected, {
a: { type: 'num', value: 2 },
b: { type: 'num', value: 3 },
c: { type: 'num', value: 4 },
});
});
});
test('combines multiple observables in arguments', () => {
testScheduler.run(({ cold, expectObservable }) => {
const arg1 = cold('--ab-c-', { a: 0, b: 2, c: 4 });
const arg2 = cold('-a--bc-', { a: 1, b: 3, c: 5 });
const expected = ' --abc(de)-';
const observable1 = {
name: 'observable1',
args: {},
help: '',
fn: () => arg1,
};
const observable2 = {
name: 'observable2',
args: {},
help: '',
fn: () => arg2,
};
const max: ExpressionFunctionDefinition<'max', any, { val1: number; val2: number }, any> = {
name: 'max',
args: {
val1: { help: '', types: ['number'] },
val2: { help: '', types: ['number'] },
},
help: '',
fn: (input, { val1, val2 }) => ({ type: 'num', value: Math.max(val1, val2) }),
};
const executor = createUnitTestExecutor();
executor.registerFunction(observable1);
executor.registerFunction(observable2);
executor.registerFunction(max);
const result = executor.run('max val1={observable1} val2={observable2}', {});
expectObservable(result).toBe(expected, {
a: { type: 'num', value: 1 },
b: { type: 'num', value: 2 },
c: { type: 'num', value: 3 },
d: { type: 'num', value: 4 },
e: { type: 'num', value: 5 },
});
});
});
test('does not complete when an argument completes', () => {
testScheduler.run(({ cold, expectObservable }) => {
const arg = cold('-a|', { a: 1 });
const expected = '-a-';
const observable = {
name: 'observable',
args: {},
help: '',
fn: () => arg,
};
const executor = createUnitTestExecutor();
executor.registerFunction(observable);
const result = executor.run('add val={observable}', 1, {});
expectObservable(result).toBe(expected, {
a: { type: 'num', value: 2 },
});
});
});
test('handles error in observable arguments', () => {
testScheduler.run(({ cold, expectObservable }) => {
const arg = cold('-a-#', { a: 1 }, new Error('some error'));
const expected = '-a-b';
const observable = {
name: 'observable',
args: {},
help: '',
fn: () => arg,
};
const executor = createUnitTestExecutor();
executor.registerFunction(observable);
const result = executor.run('add val={observable}', 1, {});
expectObservable(result).toBe(expected, {
a: { type: 'num', value: 2 },
b: {
error: expect.objectContaining({
message: '[add] > [observable] > some error',
}),
type: 'error',
},
});
});
});
});
describe('when arguments are missing', () => {
test('when required argument is missing and has not alias, returns error', async () => {
it('when required argument is missing and has not alias, returns error', async () => {
const requiredArg: ExpressionFunctionDefinition<'requiredArg', any, { arg: any }, any> = {
name: 'requiredArg',
args: {
@ -430,7 +678,7 @@ describe('Execution', () => {
};
const executor = createUnitTestExecutor();
executor.registerFunction(requiredArg);
const result = await executor.run('requiredArg', null, {});
const result = await executor.run('requiredArg', null, {}).pipe(first()).toPromise();
expect(result).toMatchObject({
type: 'error',
@ -456,7 +704,7 @@ describe('Execution', () => {
test('can execute expression in debug mode', async () => {
const execution = createExecution('add val=1 | add val=2 | add val=3', {}, true);
execution.start(-1);
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toEqual({
type: 'num',
@ -471,7 +719,7 @@ describe('Execution', () => {
true
);
execution.start(0);
const result = await execution.result;
const result = await execution.result.pipe(first()).toPromise();
expect(result).toEqual({
type: 'num',
@ -483,7 +731,7 @@ describe('Execution', () => {
test('sets "success" flag on all functions to true', async () => {
const execution = createExecution('add val=1 | add val=2 | add val=3', {}, true);
execution.start(-1);
await execution.result;
await execution.result.pipe(first()).toPromise();
for (const node of execution.state.get().ast.chain) {
expect(node.debug?.success).toBe(true);
@ -493,7 +741,7 @@ describe('Execution', () => {
test('stores "fn" reference to the function', async () => {
const execution = createExecution('add val=1 | add val=2 | add val=3', {}, true);
execution.start(-1);
await execution.result;
await execution.result.pipe(first()).toPromise();
for (const node of execution.state.get().ast.chain) {
expect(node.debug?.fn).toBe('add');
@ -503,7 +751,7 @@ describe('Execution', () => {
test('saves duration it took to execute each function', async () => {
const execution = createExecution('add val=1 | add val=2 | add val=3', {}, true);
execution.start(-1);
await execution.result;
await execution.result.pipe(first()).toPromise();
for (const node of execution.state.get().ast.chain) {
expect(typeof node.debug?.duration).toBe('number');
@ -515,7 +763,7 @@ describe('Execution', () => {
test('adds .debug field in expression AST on each executed function', async () => {
const execution = createExecution('add val=1 | add val=2 | add val=3', {}, true);
execution.start(-1);
await execution.result;
await execution.result.pipe(first()).toPromise();
for (const node of execution.state.get().ast.chain) {
expect(typeof node.debug).toBe('object');
@ -526,7 +774,7 @@ describe('Execution', () => {
test('stores input of each function', async () => {
const execution = createExecution('add val=1 | add val=2 | add val=3', {}, true);
execution.start(-1);
await execution.result;
await execution.result.pipe(first()).toPromise();
const { chain } = execution.state.get().ast;
@ -544,7 +792,7 @@ describe('Execution', () => {
test('stores output of each function', async () => {
const execution = createExecution('add val=1 | add val=2 | add val=3', {}, true);
execution.start(-1);
await execution.result;
await execution.result.pipe(first()).toPromise();
const { chain } = execution.state.get().ast;
@ -569,7 +817,7 @@ describe('Execution', () => {
true
);
execution.start(-1);
await execution.result;
await execution.result.pipe(first()).toPromise();
const { chain } = execution.state.get().ast;
@ -592,7 +840,7 @@ describe('Execution', () => {
true
);
execution.start(0);
await execution.result;
await execution.result.pipe(first()).toPromise();
const { chain } = execution.state.get().ast.chain[0].arguments
.val[0] as ExpressionAstExpression;
@ -627,7 +875,7 @@ describe('Execution', () => {
params: { debug: true },
});
execution.start(0);
await execution.result;
await execution.result.pipe(first()).toPromise();
const node1 = execution.state.get().ast.chain[0];
const node2 = execution.state.get().ast.chain[1];
@ -645,7 +893,7 @@ describe('Execution', () => {
params: { debug: true },
});
execution.start(0);
await execution.result;
await execution.result.pipe(first()).toPromise();
const node2 = execution.state.get().ast.chain[1];
@ -666,7 +914,7 @@ describe('Execution', () => {
params: { debug: true },
});
execution.start(0);
await execution.result;
await execution.result.pipe(first()).toPromise();
const node2 = execution.state.get().ast.chain[1];

473
src/plugins/expressions/common/execution/execution.ts
View file

@ -7,14 +7,28 @@
*/
import { i18n } from '@kbn/i18n';
import { isPromise } from '@kbn/std';
import { keys, last, mapValues, reduce, zipObject } from 'lodash';
import {
combineLatest,
defer,
from,
isObservable,
of,
race,
throwError,
Observable,
ReplaySubject,
} from 'rxjs';
import { catchError, finalize, map, shareReplay, switchMap, tap } from 'rxjs/operators';
import { Executor } from '../executor';
import { createExecutionContainer, ExecutionContainer } from './container';
import { createError } from '../util';
import { abortSignalToPromise, Defer, now } from '../../../kibana_utils/common';
import { abortSignalToPromise, now } from '../../../kibana_utils/common';
import { RequestAdapter, Adapters } from '../../../inspector/common';
import { isExpressionValueError, ExpressionValueError } from '../expression_types/specs/error';
import {
ExpressionAstArgument,
ExpressionAstExpression,
ExpressionAstFunction,
parse,
@ -23,8 +37,8 @@ import {
ExpressionAstNode,
} from '../ast';
import { ExecutionContext, DefaultInspectorAdapters } from './types';
import { getType, ExpressionValue, Datatable } from '../expression_types';
import { ArgumentType, ExpressionFunction } from '../expression_functions';
import { getType, Datatable } from '../expression_types';
import { ExpressionFunction } from '../expression_functions';
import { getByAlias } from '../util/get_by_alias';
import { ExecutionContract } from './execution_contract';
import { ExpressionExecutionParams } from '../service';
@ -87,6 +101,11 @@ export class Execution<
*/
public input: Input = null as any;
/**
* Input of the started execution.
*/
private input$ = new ReplaySubject<Input>(1);
/**
* Execution context - object that allows to do side-effects. Context is passed
* to every function.
@ -104,10 +123,10 @@ export class Execution<
private readonly abortRejection = abortSignalToPromise(this.abortController.signal);
/**
* Races a given promise against the "abort" event of `abortController`.
* Races a given observable against the "abort" event of `abortController`.
*/
private race<T>(promise: Promise<T>): Promise<T> {
return Promise.race<T>([this.abortRejection.promise, promise]);
private race<T>(observable: Observable<T>): Observable<T> {
return race(from(this.abortRejection.promise), observable);
}
/**
@ -118,7 +137,7 @@ export class Execution<
/**
* Future that tracks result or error of this execution.
*/
private readonly firstResultFuture = new Defer<Output | ExpressionValueError>();
public readonly result: Observable<Output | ExpressionValueError>;
/**
* Keeping track of any child executions
@ -139,10 +158,6 @@ export class Execution<
public readonly expression: string;
public get result(): Promise<Output | ExpressionValueError> {
return this.firstResultFuture.promise;
}
public get inspectorAdapters(): InspectorAdapters {
return this.context.inspectorAdapters;
}
@ -184,6 +199,28 @@ export class Execution<
isSyncColorsEnabled: () => execution.params.syncColors,
...(execution.params as any).extraContext,
};
this.result = this.input$.pipe(
switchMap((input) => this.race(this.invokeChain(this.state.get().ast.chain, input))),
catchError((error) => {
if (this.abortController.signal.aborted) {
this.childExecutions.forEach((childExecution) => childExecution.cancel());
return of(createAbortErrorValue());
}
return throwError(error);
}),
tap({
next: (result) => {
this.context.inspectorAdapters.expression?.logAST(this.state.get().ast);
this.state.transitions.setResult(result);
},
error: (error) => this.state.transitions.setError(error),
}),
finalize(() => this.abortRejection.cleanup()),
shareReplay(1)
);
}
/**
@ -199,150 +236,139 @@ export class Execution<
* N.B. `input` is initialized to `null` rather than `undefined` for legacy reasons,
* because in legacy interpreter it was set to `null` by default.
*/
public start(input: Input = null as any) {
public start(input: Input = null as any): Observable<Output | ExpressionValueError> {
if (this.hasStarted) throw new Error('Execution already started.');
this.hasStarted = true;
this.input = input;
this.state.transitions.start();
const { resolve, reject } = this.firstResultFuture;
const chainPromise = this.invokeChain(this.state.get().ast.chain, input);
this.race(chainPromise).then(resolve, (error) => {
if (this.abortController.signal.aborted) {
this.childExecutions.forEach((ex) => ex.cancel());
resolve(createAbortErrorValue());
} else reject(error);
});
this.firstResultFuture.promise
.then(
(result) => {
if (this.context.inspectorAdapters.expression) {
this.context.inspectorAdapters.expression.logAST(this.state.get().ast);
}
this.state.transitions.setResult(result);
},
(error) => {
this.state.transitions.setError(error);
}
)
.finally(() => {
this.abortRejection.cleanup();
});
}
async invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Promise<any> {
if (!chainArr.length) return input;
for (const link of chainArr) {
const { function: fnName, arguments: fnArgs } = link;
const fn = getByAlias(this.state.get().functions, fnName);
if (!fn) {
return createError({
name: 'fn not found',
message: i18n.translate('expressions.execution.functionNotFound', {
defaultMessage: `Function {fnName} could not be found.`,
values: {
fnName,
},
}),
});
}
if (fn.disabled) {
return createError({
name: 'fn is disabled',
message: i18n.translate('expressions.execution.functionDisabled', {
defaultMessage: `Function {fnName} is disabled.`,
values: {
fnName,
},
}),
});
}
let args: Record<string, ExpressionValue> = {};
let timeStart: number | undefined;
try {
// `resolveArgs` returns an object because the arguments themselves might
// actually have a `then` function which would be treated as a `Promise`.
const { resolvedArgs } = await this.race(this.resolveArgs(fn, input, fnArgs));
args = resolvedArgs;
timeStart = this.execution.params.debug ? now() : 0;
const output = await this.race(this.invokeFunction(fn, input, resolvedArgs));
if (this.execution.params.debug) {
const timeEnd: number = now();
(link as ExpressionAstFunction).debug = {
success: true,
fn: fn.name,
input,
args: resolvedArgs,
output,
duration: timeEnd - timeStart,
};
}
if (getType(output) === 'error') return output;
input = output;
} catch (rawError) {
const timeEnd: number = this.execution.params.debug ? now() : 0;
const error = createError(rawError) as ExpressionValueError;
error.error.message = `[${fnName}] > ${error.error.message}`;
if (this.execution.params.debug) {
(link as ExpressionAstFunction).debug = {
success: false,
fn: fn.name,
input,
args,
error,
rawError,
duration: timeStart ? timeEnd - timeStart : undefined,
};
}
return error;
}
if (isObservable<Input>(input)) {
// `input$` should never complete
input.subscribe(
(value) => this.input$.next(value),
(error) => this.input$.error(error)
);
} else if (isPromise(input)) {
input.then(
(value) => this.input$.next(value),
(error) => this.input$.error(error)
);
} else {
this.input$.next(input);
}
return input;
return this.result;
}
async invokeFunction(
invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Observable<any> {
return of(input).pipe(
...(chainArr.map((link) =>
switchMap((currentInput) => {
const { function: fnName, arguments: fnArgs } = link;
const fn = getByAlias(this.state.get().functions, fnName);
if (!fn) {
throw createError({
name: 'fn not found',
message: i18n.translate('expressions.execution.functionNotFound', {
defaultMessage: `Function {fnName} could not be found.`,
values: {
fnName,
},
}),
});
}
if (fn.disabled) {
throw createError({
name: 'fn is disabled',
message: i18n.translate('expressions.execution.functionDisabled', {
defaultMessage: `Function {fnName} is disabled.`,
values: {
fnName,
},
}),
});
}
if (this.execution.params.debug) {
link.debug = {
args: {},
duration: 0,
fn: fn.name,
input: currentInput,
success: true,
};
}
const timeStart = this.execution.params.debug ? now() : 0;
// `resolveArgs` returns an object because the arguments themselves might
// actually have `then` or `subscribe` methods which would be treated as a `Promise`
// or an `Observable` accordingly.
return this.race(this.resolveArgs(fn, currentInput, fnArgs)).pipe(
tap((args) => this.execution.params.debug && Object.assign(link.debug, { args })),
switchMap((args) => this.race(this.invokeFunction(fn, currentInput, args))),
switchMap((output) => (getType(output) === 'error' ? throwError(output) : of(output))),
tap((output) => this.execution.params.debug && Object.assign(link.debug, { output })),
catchError((rawError) => {
const error = createError(rawError);
error.error.message = `[${fnName}] > ${error.error.message}`;
if (this.execution.params.debug) {
Object.assign(link.debug, { error, rawError, success: false });
}
return throwError(error);
}),
finalize(() => {
if (this.execution.params.debug) {
Object.assign(link.debug, { duration: now() - timeStart });
}
})
);
})
) as Parameters<Observable<unknown>['pipe']>),
catchError((error) => of(error))
);
}
invokeFunction(
fn: ExpressionFunction,
input: unknown,
args: Record<string, unknown>
): Promise<any> {
const normalizedInput = this.cast(input, fn.inputTypes);
const output = await this.race(fn.fn(normalizedInput, args, this.context));
): Observable<any> {
return of(input).pipe(
map((currentInput) => this.cast(currentInput, fn.inputTypes)),
switchMap((normalizedInput) => this.race(of(fn.fn(normalizedInput, args, this.context)))),
switchMap((fnResult: any) =>
isObservable(fnResult) ? fnResult : from(isPromise(fnResult) ? fnResult : [fnResult])
),
map((output) => {
// Validate that the function returned the type it said it would.
// This isn't required, but it keeps function developers honest.
const returnType = getType(output);
const expectedType = fn.type;
if (expectedType && returnType !== expectedType) {
throw new Error(
`Function '${fn.name}' should return '${expectedType}',` +
` actually returned '${returnType}'`
);
}
// Validate that the function returned the type it said it would.
// This isn't required, but it keeps function developers honest.
const returnType = getType(output);
const expectedType = fn.type;
if (expectedType && returnType !== expectedType) {
throw new Error(
`Function '${fn.name}' should return '${expectedType}',` +
` actually returned '${returnType}'`
);
}
// Validate the function output against the type definition's validate function.
const type = this.context.types[fn.type];
if (type && type.validate) {
try {
type.validate(output);
} catch (e) {
throw new Error(`Output of '${fn.name}' is not a valid type '${fn.type}': ${e}`);
}
}
// Validate the function output against the type definition's validate function.
const type = this.context.types[fn.type];
if (type && type.validate) {
try {
type.validate(output);
} catch (e) {
throw new Error(`Output of '${fn.name}' is not a valid type '${fn.type}': ${e}`);
}
}
return output;
return output;
})
);
}
public cast(value: any, toTypeNames?: string[]) {
@ -371,98 +397,96 @@ export class Execution<
}
// Processes the multi-valued AST argument values into arguments that can be passed to the function
async resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Promise<any> {
const argDefs = fnDef.args;
resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Observable<any> {
return defer(() => {
const { args: argDefs } = fnDef;
// Use the non-alias name from the argument definition
const dealiasedArgAsts = reduce(
argAsts,
(acc, argAst, argName) => {
const argDef = getByAlias(argDefs, argName);
if (!argDef) {
throw new Error(`Unknown argument '${argName}' passed to function '${fnDef.name}'`);
// Use the non-alias name from the argument definition
const dealiasedArgAsts = reduce(
argAsts as Record<string, ExpressionAstArgument>,
(acc, argAst, argName) => {
const argDef = getByAlias(argDefs, argName);
if (!argDef) {
throw new Error(`Unknown argument '${argName}' passed to function '${fnDef.name}'`);
}
acc[argDef.name] = (acc[argDef.name] || []).concat(argAst);
return acc;
},
{} as Record<string, ExpressionAstArgument[]>
);
// Check for missing required arguments.
for (const { aliases, default: argDefault, name, required } of Object.values(argDefs)) {
if (!(name in dealiasedArgAsts) && typeof argDefault !== 'undefined') {
dealiasedArgAsts[name] = [parse(argDefault, 'argument')];
}
acc[argDef.name] = (acc[argDef.name] || []).concat(argAst);
return acc;
},
{} as any
);
// Check for missing required arguments.
for (const argDef of Object.values(argDefs)) {
const {
aliases,
default: argDefault,
name: argName,
required,
} = argDef as ArgumentType<any> & { name: string };
if (
typeof argDefault !== 'undefined' ||
!required ||
typeof dealiasedArgAsts[argName] !== 'undefined'
)
continue;
if (!required || name in dealiasedArgAsts) {
continue;
}
if (!aliases || aliases.length === 0) {
throw new Error(`${fnDef.name} requires an argument`);
if (!aliases?.length) {
throw new Error(`${fnDef.name} requires an argument`);
}
// use an alias if _ is the missing arg
const errorArg = name === '_' ? aliases[0] : name;
throw new Error(`${fnDef.name} requires an "${errorArg}" argument`);
}
// use an alias if _ is the missing arg
const errorArg = argName === '_' ? aliases[0] : argName;
throw new Error(`${fnDef.name} requires an "${errorArg}" argument`);
}
// Create the functions to resolve the argument ASTs into values
// These are what are passed to the actual functions if you opt out of resolving
const resolveArgFns = mapValues(dealiasedArgAsts, (asts, argName) =>
asts.map((item) => (subInput = input) =>
this.interpret(item, subInput).pipe(
map((output) => {
if (isExpressionValueError(output)) {
throw output.error;
}
// Fill in default values from argument definition
const argAstsWithDefaults = reduce(
argDefs,
(acc: any, argDef: any, argName: any) => {
if (typeof acc[argName] === 'undefined' && typeof argDef.default !== 'undefined') {
acc[argName] = [parse(argDef.default, 'argument')];
}
return this.cast(output, argDefs[argName].types);
})
)
)
);
return acc;
},
dealiasedArgAsts
);
const argNames = keys(resolveArgFns);
// Create the functions to resolve the argument ASTs into values
// These are what are passed to the actual functions if you opt out of resolving
const resolveArgFns = mapValues(argAstsWithDefaults, (asts, argName) => {
return asts.map((item: ExpressionAstExpression) => {
return async (subInput = input) => {
const output = await this.interpret(item, subInput);
if (isExpressionValueError(output)) throw output.error;
const casted = this.cast(output, argDefs[argName as any].types);
return casted;
};
});
if (!argNames.length) {
return from([[]]);
}
const resolvedArgValuesObservable = combineLatest(
argNames.map((argName) => {
const interpretFns = resolveArgFns[argName];
// `combineLatest` does not emit a value on an empty collection
// @see https://github.com/ReactiveX/RxSwift/issues/1879
if (!interpretFns.length) {
return of([]);
}
return argDefs[argName].resolve
? combineLatest(interpretFns.map((fn) => fn()))
: of(interpretFns);
})
);
return resolvedArgValuesObservable.pipe(
map((resolvedArgValues) =>
mapValues(
// Return an object here because the arguments themselves might actually have a 'then'
// function which would be treated as a promise
zipObject(argNames, resolvedArgValues),
// Just return the last unless the argument definition allows multiple
(argValues, argName) => (argDefs[argName].multi ? argValues : last(argValues))
)
)
);
});
const argNames = keys(resolveArgFns);
// Actually resolve unless the argument definition says not to
const resolvedArgValues = await Promise.all(
argNames.map((argName) => {
const interpretFns = resolveArgFns[argName];
if (!argDefs[argName].resolve) return interpretFns;
return Promise.all(interpretFns.map((fn: any) => fn()));
})
);
const resolvedMultiArgs = zipObject(argNames, resolvedArgValues);
// Just return the last unless the argument definition allows multiple
const resolvedArgs = mapValues(resolvedMultiArgs, (argValues, argName) => {
if (argDefs[argName as any].multi) return argValues;
return last(argValues as any);
});
// Return an object here because the arguments themselves might actually have a 'then'
// function which would be treated as a promise
return { resolvedArgs };
}
public async interpret<T>(ast: ExpressionAstNode, input: T): Promise<unknown> {
public interpret<T>(ast: ExpressionAstNode, input: T): Observable<unknown> {
switch (getType(ast)) {
case 'expression':
const execution = this.execution.executor.createExecution(
@ -470,15 +494,14 @@ export class Execution<
this.execution.params
);
this.childExecutions.push(execution);
execution.start(input);
return await execution.result;
return execution.start(input);
case 'string':
case 'number':
case 'null':
case 'boolean':
return ast;
return of(ast);
default:
throw new Error(`Unknown AST object: ${JSON.stringify(ast)}`);
return throwError(new Error(`Unknown AST object: ${JSON.stringify(ast)}`));
}
}
}

5
src/plugins/expressions/common/execution/execution_contract.test.ts
View file

@ -6,6 +6,7 @@
* Side Public License, v 1.
*/
import { first } from 'rxjs/operators';
import { Execution } from './execution';
import { parseExpression } from '../ast';
import { createUnitTestExecutor } from '../test_helpers';
@ -108,7 +109,7 @@ describe('ExecutionContract', () => {
const contract = new ExecutionContract(execution);
execution.start();
await execution.result;
await execution.result.pipe(first()).toPromise();
expect(contract.isPending).toBe(false);
expect(execution.state.get().state).toBe('result');
@ -119,7 +120,7 @@ describe('ExecutionContract', () => {
const contract = new ExecutionContract(execution);
execution.start();
await execution.result;
await execution.result.pipe(first()).toPromise();
execution.state.get().state = 'error';
expect(contract.isPending).toBe(false);

29
src/plugins/expressions/common/execution/execution_contract.ts
View file

@ -6,6 +6,8 @@
* Side Public License, v 1.
*/
import { of } from 'rxjs';
import { catchError, take } from 'rxjs/operators';
import { Execution } from './execution';
import { ExpressionValueError } from '../expression_types/specs';
import { ExpressionAstExpression } from '../ast';
@ -38,18 +40,21 @@ export class ExecutionContract<Input = unknown, Output = unknown, InspectorAdapt
* This function never throws.
*/
getData = async (): Promise<Output | ExpressionValueError> => {
try {
return await this.execution.result;
} catch (e) {
return {
type: 'error',
error: {
name: e.name,
message: e.message,
stack: e.stack,
},
};
}
return this.execution.result
.pipe(
take(1),
catchError(({ name, message, stack }) =>
of({
type: 'error',
error: {
name,
message,
stack,
},
} as ExpressionValueError)
)
)
.toPromise();
};
/**

11
src/plugins/expressions/common/executor/executor.ts
View file

@ -9,6 +9,7 @@
/* eslint-disable max-classes-per-file */
import { cloneDeep, mapValues } from 'lodash';
import { Observable } from 'rxjs';
import { ExecutorState, ExecutorContainer } from './container';
import { createExecutorContainer } from './container';
import { AnyExpressionFunctionDefinition, ExpressionFunction } from '../expression_functions';
@ -17,7 +18,7 @@ import { IRegistry } from '../types';
import { ExpressionType } from '../expression_types/expression_type';
import { AnyExpressionTypeDefinition } from '../expression_types/types';
import { ExpressionAstExpression, ExpressionAstFunction } from '../ast';
import { typeSpecs } from '../expression_types/specs';
import { ExpressionValueError, typeSpecs } from '../expression_types/specs';
import { functionSpecs } from '../expression_functions/specs';
import { getByAlias } from '../util';
import { SavedObjectReference } from '../../../../core/types';
@ -156,14 +157,12 @@ export class Executor<Context extends Record<string, unknown> = Record<string, u
* @param context Extra global context object that will be merged into the
* expression global context object that is provided to each function to allow side-effects.
*/
public async run<Input, Output>(
public run<Input, Output>(
ast: string | ExpressionAstExpression,
input: Input,
params: ExpressionExecutionParams = {}
) {
const execution = this.createExecution(ast, params);
execution.start(input);
return (await execution.result) as Output;
): Observable<Output | ExpressionValueError> {
return this.createExecution<Input, Output>(ast, params).start(input);
}
public createExecution<Input = unknown, Output = unknown>(

4
src/plugins/expressions/common/expression_functions/expression_function.ts
View file

@ -10,7 +10,6 @@ import { identity } from 'lodash';
import { AnyExpressionFunctionDefinition } from './types';
import { ExpressionFunctionParameter } from './expression_function_parameter';
import { ExpressionValue } from '../expression_types/types';
import { ExecutionContext } from '../execution';
import { ExpressionAstFunction } from '../ast';
import { SavedObjectReference } from '../../../../core/types';
import { PersistableState, SerializableState } from '../../../kibana_utils/common';
@ -89,8 +88,7 @@ export class ExpressionFunction implements PersistableState<ExpressionAstFunctio
this.name = name;
this.type = type;
this.aliases = aliases || [];
this.fn = (input, params, handlers) =>
Promise.resolve(fn(input, params, handlers as ExecutionContext));
this.fn = fn as ExpressionFunction['fn'];
this.help = help || '';
this.inputTypes = inputTypes || context?.types;
this.disabled = disabled || false;

10
src/plugins/expressions/common/expression_functions/specs/map_column.ts
View file

@ -6,6 +6,8 @@
* Side Public License, v 1.
*/
import { Observable } from 'rxjs';
import { take } from 'rxjs/operators';
import { i18n } from '@kbn/i18n';
import { ExpressionFunctionDefinition } from '../types';
import { Datatable, getType } from '../../expression_types';
@ -13,7 +15,7 @@ import { Datatable, getType } from '../../expression_types';
export interface MapColumnArguments {
id?: string | null;
name: string;
expression?: (datatable: Datatable) => Promise<boolean | number | string | null>;
expression?(datatable: Datatable): Observable<boolean | number | string | null>;
copyMetaFrom?: string | null;
}
@ -79,7 +81,11 @@ export const mapColumn: ExpressionFunctionDefinition<
},
},
fn: (input, args) => {
const expression = args.expression || (() => Promise.resolve(null));
const expression = (...params: Parameters<Required<MapColumnArguments>['expression']>) =>
args
.expression?.(...params)
.pipe(take(1))
.toPromise() ?? Promise.resolve(null);
const columnId = args.id != null ? args.id : args.name;
const columns = [...input.columns];

190
src/plugins/expressions/common/expression_functions/specs/tests/map_column.test.ts
View file

@ -6,85 +6,80 @@
* Side Public License, v 1.
*/
import { of } from 'rxjs';
import { Datatable } from '../../../expression_types';
import { mapColumn, MapColumnArguments } from '../map_column';
import { emptyTable, functionWrapper, testTable } from './utils';
const pricePlusTwo = (datatable: Datatable) => Promise.resolve(datatable.rows[0].price + 2);
const pricePlusTwo = (datatable: Datatable) => of(datatable.rows[0].price + 2);
describe('mapColumn', () => {
const fn = functionWrapper(mapColumn);
const runFn = (input: Datatable, args: MapColumnArguments) =>
fn(input, args) as Promise<Datatable>;
it('returns a datatable with a new column with the values from mapping a function over each row in a datatable', () => {
return runFn(testTable, {
it('returns a datatable with a new column with the values from mapping a function over each row in a datatable', async () => {
const arbitraryRowIndex = 2;
const result = await runFn(testTable, {
id: 'pricePlusTwo',
name: 'pricePlusTwo',
expression: pricePlusTwo,
}).then((result) => {
const arbitraryRowIndex = 2;
});
expect(result.type).toBe('datatable');
expect(result.columns).toEqual([
...testTable.columns,
{ id: 'pricePlusTwo', name: 'pricePlusTwo', meta: { type: 'number' } },
]);
expect(result.columns[result.columns.length - 1]).toHaveProperty('name', 'pricePlusTwo');
expect(result.rows[arbitraryRowIndex]).toHaveProperty('pricePlusTwo');
expect(result.type).toBe('datatable');
expect(result.columns).toEqual([
...testTable.columns,
{ id: 'pricePlusTwo', name: 'pricePlusTwo', meta: { type: 'number' } },
]);
expect(result.columns[result.columns.length - 1]).toHaveProperty('name', 'pricePlusTwo');
expect(result.rows[arbitraryRowIndex]).toHaveProperty('pricePlusTwo');
});
it('overwrites existing column with the new column if an existing column name is provided', async () => {
const result = await runFn(testTable, { name: 'name', expression: pricePlusTwo });
const nameColumnIndex = result.columns.findIndex(({ name }) => name === 'name');
const arbitraryRowIndex = 4;
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(testTable.columns.length);
expect(result.columns[nameColumnIndex]).toHaveProperty('name', 'name');
expect(result.columns[nameColumnIndex].meta).toHaveProperty('type', 'number');
expect(result.rows[arbitraryRowIndex]).toHaveProperty('name', 202);
});
it('adds a column to empty tables', async () => {
const result = await runFn(emptyTable, { name: 'name', expression: pricePlusTwo });
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(1);
expect(result.columns[0]).toHaveProperty('name', 'name');
expect(result.columns[0].meta).toHaveProperty('type', 'null');
});
it('should assign specific id, different from name, when id arg is passed for new columns', async () => {
const result = await runFn(emptyTable, { name: 'name', id: 'myid', expression: pricePlusTwo });
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(1);
expect(result.columns[0]).toHaveProperty('name', 'name');
expect(result.columns[0]).toHaveProperty('id', 'myid');
expect(result.columns[0].meta).toHaveProperty('type', 'null');
});
it('should assign specific id, different from name, when id arg is passed for copied column', async () => {
const result = await runFn(testTable, { name: 'name', id: 'myid', expression: pricePlusTwo });
const nameColumnIndex = result.columns.findIndex(({ name }) => name === 'name');
expect(result.type).toBe('datatable');
expect(result.columns[nameColumnIndex]).toEqual({
id: 'myid',
name: 'name',
meta: { type: 'number' },
});
});
it('overwrites existing column with the new column if an existing column name is provided', () => {
return runFn(testTable, { name: 'name', expression: pricePlusTwo }).then((result) => {
const nameColumnIndex = result.columns.findIndex(({ name }) => name === 'name');
const arbitraryRowIndex = 4;
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(testTable.columns.length);
expect(result.columns[nameColumnIndex]).toHaveProperty('name', 'name');
expect(result.columns[nameColumnIndex].meta).toHaveProperty('type', 'number');
expect(result.rows[arbitraryRowIndex]).toHaveProperty('name', 202);
});
});
it('adds a column to empty tables', () => {
return runFn(emptyTable, { name: 'name', expression: pricePlusTwo }).then((result) => {
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(1);
expect(result.columns[0]).toHaveProperty('name', 'name');
expect(result.columns[0].meta).toHaveProperty('type', 'null');
});
});
it('should assign specific id, different from name, when id arg is passed for new columns', () => {
return runFn(emptyTable, { name: 'name', id: 'myid', expression: pricePlusTwo }).then(
(result) => {
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(1);
expect(result.columns[0]).toHaveProperty('name', 'name');
expect(result.columns[0]).toHaveProperty('id', 'myid');
expect(result.columns[0].meta).toHaveProperty('type', 'null');
}
);
});
it('should assign specific id, different from name, when id arg is passed for copied column', () => {
return runFn(testTable, { name: 'name', id: 'myid', expression: pricePlusTwo }).then(
(result) => {
const nameColumnIndex = result.columns.findIndex(({ name }) => name === 'name');
expect(result.type).toBe('datatable');
expect(result.columns[nameColumnIndex]).toEqual({
id: 'myid',
name: 'name',
meta: { type: 'number' },
});
}
);
});
it('should copy over the meta information from the specified column', () => {
return runFn(
it('should copy over the meta information from the specified column', async () => {
const result = await runFn(
{
...testTable,
columns: [
@ -99,52 +94,53 @@ describe('mapColumn', () => {
rows: testTable.rows.map((row) => ({ ...row, myId: Date.now() })),
},
{ name: 'name', copyMetaFrom: 'myId', expression: pricePlusTwo }
).then((result) => {
const nameColumnIndex = result.columns.findIndex(({ name }) => name === 'name');
expect(result.type).toBe('datatable');
expect(result.columns[nameColumnIndex]).toEqual({
id: 'name',
name: 'name',
meta: { type: 'date', params: { id: 'number', params: { digits: 2 } } },
});
);
const nameColumnIndex = result.columns.findIndex(({ name }) => name === 'name');
expect(result.type).toBe('datatable');
expect(result.columns[nameColumnIndex]).toEqual({
id: 'name',
name: 'name',
meta: { type: 'date', params: { id: 'number', params: { digits: 2 } } },
});
});
it('should be resilient if the references column for meta information does not exists', () => {
return runFn(emptyTable, { name: 'name', copyMetaFrom: 'time', expression: pricePlusTwo }).then(
(result) => {
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(1);
expect(result.columns[0]).toHaveProperty('name', 'name');
expect(result.columns[0]).toHaveProperty('id', 'name');
expect(result.columns[0].meta).toHaveProperty('type', 'null');
}
);
it('should be resilient if the references column for meta information does not exists', async () => {
const result = await runFn(emptyTable, {
name: 'name',
copyMetaFrom: 'time',
expression: pricePlusTwo,
});
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(1);
expect(result.columns[0]).toHaveProperty('name', 'name');
expect(result.columns[0]).toHaveProperty('id', 'name');
expect(result.columns[0].meta).toHaveProperty('type', 'null');
});
it('should correctly infer the type fromt he first row if the references column for meta information does not exists', () => {
return runFn(
it('should correctly infer the type fromt he first row if the references column for meta information does not exists', async () => {
const result = await runFn(
{ ...emptyTable, rows: [...emptyTable.rows, { value: 5 }] },
{ name: 'value', copyMetaFrom: 'time', expression: pricePlusTwo }
).then((result) => {
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(1);
expect(result.columns[0]).toHaveProperty('name', 'value');
expect(result.columns[0]).toHaveProperty('id', 'value');
expect(result.columns[0].meta).toHaveProperty('type', 'number');
});
);
expect(result.type).toBe('datatable');
expect(result.columns).toHaveLength(1);
expect(result.columns[0]).toHaveProperty('name', 'value');
expect(result.columns[0]).toHaveProperty('id', 'value');
expect(result.columns[0].meta).toHaveProperty('type', 'number');
});
describe('expression', () => {
it('maps null values to the new column', () => {
return runFn(testTable, { name: 'empty' }).then((result) => {
const emptyColumnIndex = result.columns.findIndex(({ name }) => name === 'empty');
const arbitraryRowIndex = 8;
it('maps null values to the new column', async () => {
const result = await runFn(testTable, { name: 'empty' });
const emptyColumnIndex = result.columns.findIndex(({ name }) => name === 'empty');
const arbitraryRowIndex = 8;
expect(result.columns[emptyColumnIndex]).toHaveProperty('name', 'empty');
expect(result.columns[emptyColumnIndex].meta).toHaveProperty('type', 'null');
expect(result.rows[arbitraryRowIndex]).toHaveProperty('empty', null);
});
expect(result.columns[emptyColumnIndex]).toHaveProperty('name', 'empty');
expect(result.columns[emptyColumnIndex].meta).toHaveProperty('type', 'null');
expect(result.rows[arbitraryRowIndex]).toHaveProperty('empty', null);
});
});
});

3
src/plugins/expressions/common/service/expressions_services.ts
View file

@ -6,6 +6,7 @@
* Side Public License, v 1.
*/
import { take } from 'rxjs/operators';
// eslint-disable-next-line @kbn/eslint/no-restricted-paths
import type { KibanaRequest } from 'src/core/server';
@ -228,7 +229,7 @@ export class ExpressionsService implements PersistableStateService<ExpressionAst
): void => this.renderers.register(definition);
public readonly run: ExpressionsServiceStart['run'] = (ast, input, params) =>
this.executor.run(ast, input, params);
this.executor.run(ast, input, params).pipe(take(1)).toPromise<any>();
public readonly getFunction: ExpressionsServiceStart['getFunction'] = (name) =>
this.executor.getFunction(name);

2
src/plugins/expressions/common/test_helpers/expression_functions/index.ts
View file

@ -12,6 +12,7 @@ import { error } from './error';
import { introspectContext } from './introspect_context';
import { mult } from './mult';
import { sleep } from './sleep';
import { sum } from './sum';
import { AnyExpressionFunctionDefinition } from '../../expression_functions';
export const functionTestSpecs: AnyExpressionFunctionDefinition[] = [
@ -21,4 +22,5 @@ export const functionTestSpecs: AnyExpressionFunctionDefinition[] = [
introspectContext,
mult,
sleep,
sum,
];

23
src/plugins/expressions/common/test_helpers/expression_functions/sum.ts Normal file
View file

@ -0,0 +1,23 @@
/*
* 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 { ExpressionFunctionDefinition } from '../../expression_functions';
import { ExpressionValueNum } from '../../expression_types';
export const sum: ExpressionFunctionDefinition<'sum', unknown[], {}, ExpressionValueNum> = {
name: 'sum',
help: 'This function summarizes the input',
inputTypes: [],
args: {},
fn: (values) => {
return {
type: 'num',
value: Array.isArray(values) ? values.reduce((a, b) => a + b) : values,
};
},
};

12
src/plugins/expressions/common/types/common.ts
View file

@ -6,7 +6,7 @@
* Side Public License, v 1.
*/
import { UnwrapPromiseOrReturn } from '@kbn/utility-types';
import { ObservableLike, UnwrapObservable, UnwrapPromiseOrReturn } from '@kbn/utility-types';
/**
* This can convert a type into a known Expression string representation of
@ -23,9 +23,9 @@ export type TypeToString<T> = KnownTypeToString<T> | UnmappedTypeStrings;
* the `type` key as a string literal type for it.
*/
// prettier-ignore
export type KnownTypeToString<T> =
T extends string ? 'string' :
T extends boolean ? 'boolean' :
export type KnownTypeToString<T> =
T extends string ? 'string' :
T extends boolean ? 'boolean' :
T extends number ? 'number' :
T extends null ? 'null' :
T extends { type: string } ? T['type'] :
@ -36,7 +36,9 @@ export type KnownTypeToString<T> =
*
* `someArgument: Promise<boolean | string>` results in `types: ['boolean', 'string']`
*/
export type TypeString<T> = KnownTypeToString<UnwrapPromiseOrReturn<T>>;
export type TypeString<T> = KnownTypeToString<
T extends ObservableLike<any> ? UnwrapObservable<T> : UnwrapPromiseOrReturn<T>
>;
/**
* Types used in Expressions that don't map to a primitive cleanly:

7
src/plugins/expressions/public/loader.test.ts
View file

@ -43,6 +43,11 @@ jest.mock('./services', () => {
};
service.registerFunction(testFn);
// eslint-disable-next-line @typescript-eslint/no-var-requires
for (const func of require('../common/test_helpers/expression_functions').functionTestSpecs) {
service.registerFunction(func);
}
const moduleMock = {
__execution: undefined,
__getLastExecution: () => moduleMock.__execution,
@ -144,7 +149,7 @@ describe('ExpressionLoader', () => {
});
it('cancels the previous request when the expression is updated', () => {
const expressionLoader = new ExpressionLoader(element, 'var foo', {});
const expressionLoader = new ExpressionLoader(element, 'sleep 10', {});
const execution = __getLastExecution();
jest.spyOn(execution, 'cancel');

19
src/plugins/expressions/public/public.api.md
View file

@ -11,10 +11,12 @@ import { EnvironmentMode } from '@kbn/config';
import { EventEmitter } from 'events';
import { KibanaRequest } from 'src/core/server';
import { Observable } from 'rxjs';
import { ObservableLike } from '@kbn/utility-types';
import { PackageInfo } from '@kbn/config';
import { Plugin as Plugin_2 } from 'src/core/public';
import { PluginInitializerContext as PluginInitializerContext_2 } from 'src/core/public';
import React from 'react';
import { UnwrapObservable } from '@kbn/utility-types';
import { UnwrapPromiseOrReturn } from '@kbn/utility-types';
// Warning: (ae-missing-release-tag) "AnyExpressionFunctionDefinition" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
@ -110,16 +112,15 @@ export class Execution<Input = unknown, Output = unknown, InspectorAdapters exte
// (undocumented)
get inspectorAdapters(): InspectorAdapters;
// (undocumented)
interpret<T>(ast: ExpressionAstNode, input: T): Promise<unknown>;
interpret<T>(ast: ExpressionAstNode, input: T): Observable<unknown>;
// (undocumented)
invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Promise<any>;
invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Observable<any>;
// (undocumented)
invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unknown>): Promise<any>;
invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unknown>): Observable<any>;
// (undocumented)
resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Promise<any>;
// (undocumented)
get result(): Promise<Output | ExpressionValueError>;
start(input?: Input): void;
resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Observable<any>;
readonly result: Observable<Output | ExpressionValueError>;
start(input?: Input): Observable<Output | ExpressionValueError>;
readonly state: ExecutionContainer<Output | ExpressionValueError>;
}
@ -229,7 +230,7 @@ export class Executor<Context extends Record<string, unknown> = Record<string, u
registerFunction(functionDefinition: AnyExpressionFunctionDefinition | (() => AnyExpressionFunctionDefinition)): void;
// (undocumented)
registerType(typeDefinition: AnyExpressionTypeDefinition | (() => AnyExpressionTypeDefinition)): void;
run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?: ExpressionExecutionParams): Promise<Output>;
run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?: ExpressionExecutionParams): Observable<Output | ExpressionValueError>;
// (undocumented)
readonly state: ExecutorContainer<Context>;
// (undocumented)
@ -1160,7 +1161,7 @@ export class TypesRegistry implements IRegistry<ExpressionType> {
// Warning: (ae-missing-release-tag) "TypeString" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
//
// @public
export type TypeString<T> = KnownTypeToString<UnwrapPromiseOrReturn<T>>;
export type TypeString<T> = KnownTypeToString<T extends ObservableLike<any> ? UnwrapObservable<T> : UnwrapPromiseOrReturn<T>>;
// Warning: (ae-missing-release-tag) "TypeToString" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
//

19
src/plugins/expressions/server/server.api.md
View file

@ -10,8 +10,10 @@ import { Ensure } from '@kbn/utility-types';
import { EventEmitter } from 'events';
import { KibanaRequest } from 'src/core/server';
import { Observable } from 'rxjs';
import { ObservableLike } from '@kbn/utility-types';
import { Plugin as Plugin_2 } from 'src/core/server';
import { PluginInitializerContext } from 'src/core/server';
import { UnwrapObservable } from '@kbn/utility-types';
import { UnwrapPromiseOrReturn } from '@kbn/utility-types';
// Warning: (ae-missing-release-tag) "AnyExpressionFunctionDefinition" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
@ -108,16 +110,15 @@ export class Execution<Input = unknown, Output = unknown, InspectorAdapters exte
// (undocumented)
get inspectorAdapters(): InspectorAdapters;
// (undocumented)
interpret<T>(ast: ExpressionAstNode, input: T): Promise<unknown>;
interpret<T>(ast: ExpressionAstNode, input: T): Observable<unknown>;
// (undocumented)
invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Promise<any>;
invokeChain(chainArr: ExpressionAstFunction[], input: unknown): Observable<any>;
// (undocumented)
invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unknown>): Promise<any>;
invokeFunction(fn: ExpressionFunction, input: unknown, args: Record<string, unknown>): Observable<any>;
// (undocumented)
resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Promise<any>;
// (undocumented)
get result(): Promise<Output | ExpressionValueError>;
start(input?: Input): void;
resolveArgs(fnDef: ExpressionFunction, input: unknown, argAsts: any): Observable<any>;
readonly result: Observable<Output | ExpressionValueError>;
start(input?: Input): Observable<Output | ExpressionValueError>;
readonly state: ExecutionContainer<Output | ExpressionValueError>;
}
@ -211,7 +212,7 @@ export class Executor<Context extends Record<string, unknown> = Record<string, u
registerFunction(functionDefinition: AnyExpressionFunctionDefinition | (() => AnyExpressionFunctionDefinition)): void;
// (undocumented)
registerType(typeDefinition: AnyExpressionTypeDefinition | (() => AnyExpressionTypeDefinition)): void;
run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?: ExpressionExecutionParams): Promise<Output>;
run<Input, Output>(ast: string | ExpressionAstExpression, input: Input, params?: ExpressionExecutionParams): Observable<Output | ExpressionValueError>;
// (undocumented)
readonly state: ExecutorContainer<Context>;
// (undocumented)
@ -919,7 +920,7 @@ export class TypesRegistry implements IRegistry<ExpressionType> {
// Warning: (ae-missing-release-tag) "TypeString" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
//
// @public
export type TypeString<T> = KnownTypeToString<UnwrapPromiseOrReturn<T>>;
export type TypeString<T> = KnownTypeToString<T extends ObservableLike<any> ? UnwrapObservable<T> : UnwrapPromiseOrReturn<T>>;
// Warning: (ae-missing-release-tag) "TypeToString" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
//

33
x-pack/plugins/canvas/canvas_plugin_src/functions/common/case.test.js
View file

@ -5,6 +5,7 @@
* 2.0.
*/
import { of } from 'rxjs';
import { functionWrapper } from '../../../test_helpers/function_wrapper';
import { caseFn } from './case';
@ -19,10 +20,10 @@ describe('case', () => {
describe('function', () => {
describe('no args', () => {
it('should return a case object that matches with the result as the context', async () => {
it('should return a case object that matches with the result as the context', () => {
const context = null;
const args = {};
expect(await fn(context, args)).toEqual({
expect(fn(context, args)).resolves.toEqual({
type: 'case',
matches: true,
result: context,
@ -31,24 +32,24 @@ describe('case', () => {
});
describe('no if or value', () => {
it('should return the result if provided', async () => {
it('should return the result if provided', () => {
const context = null;
const args = {
then: () => 'foo',
then: () => of('foo'),
};
expect(await fn(context, args)).toEqual({
expect(fn(context, args)).resolves.toEqual({
type: 'case',
matches: true,
result: args.then(),
result: 'foo',
});
});
});
describe('with if', () => {
it('should return as the matches prop', async () => {
it('should return as the matches prop', () => {
const context = null;
const args = { if: false };
expect(await fn(context, args)).toEqual({
expect(fn(context, args)).resolves.toEqual({
type: 'case',
matches: args.if,
result: context,
@ -57,17 +58,17 @@ describe('case', () => {
});
describe('with value', () => {
it('should return whether it matches the context as the matches prop', async () => {
it('should return whether it matches the context as the matches prop', () => {
const args = {
when: () => 'foo',
then: () => 'bar',
when: () => of('foo'),
then: () => of('bar'),
};
expect(await fn('foo', args)).toEqual({
expect(fn('foo', args)).resolves.toEqual({
type: 'case',
matches: true,
result: args.then(),
result: 'bar',
});
expect(await fn('bar', args)).toEqual({
expect(fn('bar', args)).resolves.toEqual({
type: 'case',
matches: false,
result: null,
@ -76,13 +77,13 @@ describe('case', () => {
});
describe('with if and value', () => {
it('should return the if as the matches prop', async () => {
it('should return the if as the matches prop', () => {
const context = null;
const args = {
when: () => 'foo',
if: true,
};
expect(await fn(context, args)).toEqual({
expect(fn(context, args)).resolves.toEqual({
type: 'case',
matches: args.if,
result: context,

21
x-pack/plugins/canvas/canvas_plugin_src/functions/common/case.ts
View file

@ -5,13 +5,15 @@
* 2.0.
*/
import { Observable } from 'rxjs';
import { take } from 'rxjs/operators';
import { ExpressionFunctionDefinition } from 'src/plugins/expressions/common';
import { getFunctionHelp } from '../../../i18n';
interface Arguments {
when: () => any;
if: boolean;
then: () => any;
when?(): Observable<any>;
if?: boolean;
then?(): Observable<any>;
}
interface Case {
@ -31,16 +33,16 @@ export function caseFn(): ExpressionFunctionDefinition<'case', any, Arguments, P
when: {
aliases: ['_'],
resolve: false,
help: argHelp.when,
help: argHelp.when!,
},
if: {
types: ['boolean'],
help: argHelp.if,
help: argHelp.if!,
},
then: {
resolve: false,
required: true,
help: argHelp.then,
help: argHelp.then!,
},
},
fn: async (input, args) => {
@ -56,14 +58,11 @@ async function doesMatch(context: any, args: Arguments) {
return args.if;
}
if (typeof args.when !== 'undefined') {
return (await args.when()) === context;
return (await args.when().pipe(take(1)).toPromise()) === context;
}
return true;
}
async function getResult(context: any, args: Arguments) {
if (typeof args.then !== 'undefined') {
return await args.then();
}
return context;
return args.then?.().pipe(take(1)).toPromise() ?? context;
}

27
x-pack/plugins/canvas/canvas_plugin_src/functions/common/filterrows.test.js
View file

@ -5,35 +5,38 @@
* 2.0.
*/
import { of } from 'rxjs';
import { functionWrapper } from '../../../test_helpers/function_wrapper';
import { testTable } from './__fixtures__/test_tables';
import { filterrows } from './filterrows';
const inStock = (datatable) => datatable.rows[0].in_stock;
const returnFalse = () => false;
const inStock = (datatable) => of(datatable.rows[0].in_stock);
const returnFalse = () => of(false);
describe('filterrows', () => {
const fn = functionWrapper(filterrows);
it('returns a datable', () => {
return fn(testTable, { fn: inStock }).then((result) => {
expect(result).toHaveProperty('type', 'datatable');
});
expect(fn(testTable, { fn: inStock })).resolves.toHaveProperty('type', 'datatable');
});
it('keeps rows that evaluate to true and removes rows that evaluate to false', () => {
const inStockRows = testTable.rows.filter((row) => row.in_stock);
return fn(testTable, { fn: inStock }).then((result) => {
expect(result.columns).toEqual(testTable.columns);
expect(result.rows).toEqual(inStockRows);
});
expect(fn(testTable, { fn: inStock })).resolves.toEqual(
expect.objectContaining({
columns: testTable.columns,
rows: inStockRows,
})
);
});
it('returns datatable with no rows when no rows meet function condition', () => {
return fn(testTable, { fn: returnFalse }).then((result) => {
expect(result.rows).toEqual([]);
});
expect(fn(testTable, { fn: returnFalse })).resolves.toEqual(
expect.objectContaining({
rows: [],
})
);
});
it('throws when no function is provided', () => {

6
x-pack/plugins/canvas/canvas_plugin_src/functions/common/filterrows.ts
View file

@ -5,11 +5,13 @@
* 2.0.
*/
import { Observable } from 'rxjs';
import { take } from 'rxjs/operators';
import { Datatable, ExpressionFunctionDefinition } from '../../../types';
import { getFunctionHelp } from '../../../i18n';
interface Arguments {
fn: (datatable: Datatable) => Promise<boolean>;
fn: (datatable: Datatable) => Observable<boolean>;
}
export function filterrows(): ExpressionFunctionDefinition<
@ -41,6 +43,8 @@ export function filterrows(): ExpressionFunctionDefinition<
...input,
rows: [row],
})
.pipe(take(1))
.toPromise()
);
return Promise.all(checks)

67
x-pack/plugins/canvas/canvas_plugin_src/functions/common/if.test.js
View file

@ -5,6 +5,7 @@
* 2.0.
*/
import { of } from 'rxjs';
import { functionWrapper } from '../../../test_helpers/function_wrapper';
import { ifFn } from './if';
@ -19,52 +20,68 @@ describe('if', () => {
describe('function', () => {
describe('condition passed', () => {
it('with then', async () => {
expect(await fn(null, { condition: true, then: () => 'foo' })).toBe('foo');
expect(await fn(null, { condition: true, then: () => 'foo', else: () => 'bar' })).toBe(
'foo'
);
it('with then', () => {
expect(fn(null, { condition: true, then: () => of('foo') })).resolves.toBe('foo');
expect(
fn(null, { condition: true, then: () => of('foo'), else: () => of('bar') })
).resolves.toBe('foo');
});
it('without then', async () => {
expect(await fn(null, { condition: true })).toBe(null);
expect(await fn('some context', { condition: true })).toBe('some context');
it('without then', () => {
expect(fn(null, { condition: true })).resolves.toBe(null);
expect(fn('some context', { condition: true })).resolves.toBe('some context');
});
});
describe('condition failed', () => {
it('with else', async () =>
it('with else', () =>
expect(
await fn('some context', { condition: false, then: () => 'foo', else: () => 'bar' })
).toBe('bar'));
fn('some context', {
condition: false,
then: () => of('foo'),
else: () => of('bar'),
})
).resolves.toBe('bar'));
it('without else', async () =>
expect(await fn('some context', { condition: false, then: () => 'foo' })).toBe(
it('without else', () =>
expect(fn('some context', { condition: false, then: () => of('foo') })).resolves.toBe(
'some context'
));
});
describe('falsy values', () => {
describe('for then', () => {
it('with null', async () =>
expect(await fn('some context', { condition: true, then: () => null })).toBe(null));
it('with null', () => {
expect(fn('some context', { condition: true, then: () => of(null) })).resolves.toBe(null);
});
it('with false', async () =>
expect(await fn('some context', { condition: true, then: () => false })).toBe(false));
it('with false', () => {
expect(fn('some context', { condition: true, then: () => of(false) })).resolves.toBe(
false
);
});
it('with 0', async () =>
expect(await fn('some context', { condition: true, then: () => 0 })).toBe(0));
it('with 0', () => {
expect(fn('some context', { condition: true, then: () => of(0) })).resolves.toBe(0);
});
});
describe('for else', () => {
it('with null', async () =>
expect(await fn('some context', { condition: false, else: () => null })).toBe(null));
it('with null', () => {
expect(fn('some context', { condition: false, else: () => of(null) })).resolves.toBe(
null
);
});
it('with false', async () =>
expect(await fn('some context', { condition: false, else: () => false })).toBe(false));
it('with false', () => {
expect(fn('some context', { condition: false, else: () => of(false) })).resolves.toBe(
false
);
});
it('with 0', async () =>
expect(await fn('some context', { condition: false, else: () => 0 })).toBe(0));
it('with 0', () => {
expect(fn('some context', { condition: false, else: () => of(0) })).resolves.toBe(0);
});
});
});
});

20
x-pack/plugins/canvas/canvas_plugin_src/functions/common/if.ts
View file

@ -5,13 +5,15 @@
* 2.0.
*/
import { Observable } from 'rxjs';
import { take } from 'rxjs/operators';
import { ExpressionFunctionDefinition } from 'src/plugins/expressions/common';
import { getFunctionHelp } from '../../../i18n';
interface Arguments {
condition: boolean | null;
then: () => Promise<any>;
else: () => Promise<any>;
then?(): Observable<any>;
else?(): Observable<any>;
}
export function ifFn(): ExpressionFunctionDefinition<'if', unknown, Arguments, unknown> {
@ -29,24 +31,18 @@ export function ifFn(): ExpressionFunctionDefinition<'if', unknown, Arguments, u
},
then: {
resolve: false,
help: argHelp.then,
help: argHelp.then!,
},
else: {
resolve: false,
help: argHelp.else,
help: argHelp.else!,
},
},
fn: async (input, args) => {
if (args.condition) {
if (typeof args.then === 'undefined') {
return input;
}
return await args.then();
return args.then?.().pipe(take(1)).toPromise() ?? input;
} else {
if (typeof args.else === 'undefined') {
return input;
}
return await args.else();
return args.else?.().pipe(take(1)).toPromise() ?? input;
}
},
};

75
x-pack/plugins/canvas/canvas_plugin_src/functions/common/ply.test.js
View file

@ -5,6 +5,7 @@
* 2.0.
*/
import { of } from 'rxjs';
import { functionWrapper } from '../../../test_helpers/function_wrapper';
import { getFunctionErrors } from '../../../i18n';
import { testTable } from './__fixtures__/test_tables';
@ -15,7 +16,7 @@ const errors = getFunctionErrors().ply;
const averagePrice = (datatable) => {
const average = datatable.rows.reduce((sum, row) => sum + row.price, 0) / datatable.rows.length;
return Promise.resolve({
return of({
type: 'datatable',
columns: [{ id: 'average_price', name: 'average_price', meta: { type: 'number' } }],
rows: [{ average_price: average }],
@ -23,17 +24,17 @@ const averagePrice = (datatable) => {
};
const doublePrice = (datatable) => {
const newRows = datatable.rows.map((row) => ({ double_price: row.price * 2 }));
const newRows = datatable.rows.map((row) => of({ double_price: row.price * 2 }));
return Promise.resolve({
return of({
type: 'datatable',
columns: [{ id: 'double_price', name: 'double_price', meta: { type: 'number' } }],
rows: newRows,
});
};
const rowCount = (datatable) => {
return Promise.resolve({
const rowCount = (datatable) =>
of({
type: 'datatable',
columns: [{ id: 'row_count', name: 'row_count', meta: { type: 'number' } }],
rows: [
@ -42,43 +43,40 @@ const rowCount = (datatable) => {
},
],
});
};
describe('ply', () => {
const fn = functionWrapper(ply);
it('maps a function over sub datatables grouped by specified columns and merges results into one datatable', () => {
it('maps a function over sub datatables grouped by specified columns and merges results into one datatable', async () => {
const arbitaryRowIndex = 0;
const result = await fn(testTable, {
by: ['name', 'in_stock'],
expression: [averagePrice, rowCount],
});
return fn(testTable, { by: ['name', 'in_stock'], expression: [averagePrice, rowCount] }).then(
(result) => {
expect(result.type).toBe('datatable');
expect(result.columns).toEqual([
{ id: 'name', name: 'name', meta: { type: 'string' } },
{ id: 'in_stock', name: 'in_stock', meta: { type: 'boolean' } },
{ id: 'average_price', name: 'average_price', meta: { type: 'number' } },
{ id: 'row_count', name: 'row_count', meta: { type: 'number' } },
]);
expect(result.rows[arbitaryRowIndex]).toHaveProperty('average_price');
expect(result.rows[arbitaryRowIndex]).toHaveProperty('row_count');
}
);
expect(result.type).toBe('datatable');
expect(result.columns).toEqual([
{ id: 'name', name: 'name', meta: { type: 'string' } },
{ id: 'in_stock', name: 'in_stock', meta: { type: 'boolean' } },
{ id: 'average_price', name: 'average_price', meta: { type: 'number' } },
{ id: 'row_count', name: 'row_count', meta: { type: 'number' } },
]);
expect(result.rows[arbitaryRowIndex]).toHaveProperty('average_price');
expect(result.rows[arbitaryRowIndex]).toHaveProperty('row_count');
});
describe('missing args', () => {
it('returns the original datatable if both args are missing', () => {
return fn(testTable).then((result) => expect(result).toEqual(testTable));
expect(fn(testTable)).resolves.toEqual(testTable);
});
describe('by', () => {
it('passes the entire context into the expression when no columns are provided', () => {
return fn(testTable, { expression: [rowCount] }).then((result) =>
expect(result).toEqual({
type: 'datatable',
rows: [{ row_count: testTable.rows.length }],
columns: [{ id: 'row_count', name: 'row_count', meta: { type: 'number' } }],
})
);
expect(fn(testTable, { expression: [rowCount] })).resolves.toEqual({
type: 'datatable',
rows: [{ row_count: testTable.rows.length }],
columns: [{ id: 'row_count', name: 'row_count', meta: { type: 'number' } }],
});
});
it('throws when by is an invalid column', () => {
@ -93,20 +91,23 @@ describe('ply', () => {
});
describe('expression', () => {
it('returns the original datatable grouped by the specified columns', () => {
it('returns the original datatable grouped by the specified columns', async () => {
const arbitaryRowIndex = 6;
const result = await fn(testTable, { by: ['price', 'quantity'] });
return fn(testTable, { by: ['price', 'quantity'] }).then((result) => {
expect(result.columns[0]).toHaveProperty('name', 'price');
expect(result.columns[1]).toHaveProperty('name', 'quantity');
expect(result.rows[arbitaryRowIndex]).toHaveProperty('price');
expect(result.rows[arbitaryRowIndex]).toHaveProperty('quantity');
});
expect(result.columns[0]).toHaveProperty('name', 'price');
expect(result.columns[1]).toHaveProperty('name', 'quantity');
expect(result.rows[arbitaryRowIndex]).toHaveProperty('price');
expect(result.rows[arbitaryRowIndex]).toHaveProperty('quantity');
});
it('throws when row counts do not match across resulting datatables', () => {
return fn(testTable, { by: ['name'], expression: [doublePrice, rowCount] }).catch((e) =>
expect(e.message).toBe(errors.rowCountMismatch().message)
expect(
fn(testTable, { by: ['name'], expression: [doublePrice, rowCount] })
).rejects.toEqual(
expect.objectContaining({
message: errors.rowCountMismatch().message,
})
);
});
});

6
x-pack/plugins/canvas/canvas_plugin_src/functions/common/ply.ts
View file

@ -5,13 +5,15 @@
* 2.0.
*/
import { Observable } from 'rxjs';
import { take } from 'rxjs/operators';
import { groupBy, flatten, pick, map } from 'lodash';
import { Datatable, DatatableColumn, ExpressionFunctionDefinition } from '../../../types';
import { getFunctionHelp, getFunctionErrors } from '../../../i18n';
interface Arguments {
by: string[];
expression: Array<(datatable: Datatable) => Promise<Datatable>>;
expression: Array<(datatable: Datatable) => Observable<Datatable>>;
}
type Output = Datatable | Promise<Datatable>;
@ -73,7 +75,7 @@ export function ply(): ExpressionFunctionDefinition<'ply', Datatable, Arguments,
if (args.expression) {
expressionResultPromises = args.expression.map((expression) =>
expression(originalDatatable)
expression(originalDatatable).pipe(take(1)).toPromise()
);
} else {
expressionResultPromises.push(Promise.resolve(originalDatatable));

25
x-pack/plugins/canvas/canvas_plugin_src/functions/common/switch.test.js
View file

@ -5,12 +5,13 @@
* 2.0.
*/
import { of } from 'rxjs';
import { functionWrapper } from '../../../test_helpers/function_wrapper';
import { switchFn } from './switch';
describe('switch', () => {
const fn = functionWrapper(switchFn);
const getter = (value) => () => value;
const getter = (value) => () => of(value);
const mockCases = [
{
type: 'case',
@ -48,32 +49,32 @@ describe('switch', () => {
describe('function', () => {
describe('with no cases', () => {
it('should return the context if no default is provided', async () => {
it('should return the context if no default is provided', () => {
const context = 'foo';
expect(await fn(context, {})).toBe(context);
expect(fn(context, {})).resolves.toBe(context);
});
it('should return the default if provided', async () => {
it('should return the default if provided', () => {
const context = 'foo';
const args = { default: () => 'bar' };
expect(await fn(context, args)).toBe(args.default());
const args = { default: () => of('bar') };
expect(fn(context, args)).resolves.toBe('bar');
});
});
describe('with no matching cases', () => {
it('should return the context if no default is provided', async () => {
it('should return the context if no default is provided', () => {
const context = 'foo';
const args = { case: nonMatchingCases.map(getter) };
expect(await fn(context, args)).toBe(context);
expect(fn(context, args)).resolves.toBe(context);
});
it('should return the default if provided', async () => {
it('should return the default if provided', () => {
const context = 'foo';
const args = {
case: nonMatchingCases.map(getter),
default: () => 'bar',
default: () => of('bar'),
};
expect(await fn(context, args)).toBe(args.default());
expect(fn(context, args)).resolves.toBe('bar');
});
});
@ -82,7 +83,7 @@ describe('switch', () => {
const context = 'foo';
const args = { case: mockCases.map(getter) };
const firstMatch = mockCases.find((c) => c.matches);
expect(await fn(context, args)).toBe(firstMatch.result);
expect(fn(context, args)).resolves.toBe(firstMatch.result);
});
});
});

16
x-pack/plugins/canvas/canvas_plugin_src/functions/common/switch.ts
View file

@ -5,13 +5,15 @@
* 2.0.
*/
import { Observable } from 'rxjs';
import { take } from 'rxjs/operators';
import { ExpressionFunctionDefinition } from 'src/plugins/expressions/common';
import { Case } from '../../../types';
import { getFunctionHelp } from '../../../i18n';
interface Arguments {
case: Array<() => Promise<Case>>;
default: () => any;
case: Array<() => Observable<Case>>;
default?(): Observable<any>;
}
export function switchFn(): ExpressionFunctionDefinition<'switch', unknown, Arguments, unknown> {
@ -32,25 +34,21 @@ export function switchFn(): ExpressionFunctionDefinition<'switch', unknown, Argu
default: {
aliases: ['finally'],
resolve: false,
help: argHelp.default,
help: argHelp.default!,
},
},
fn: async (input, args) => {
const cases = args.case || [];
for (let i = 0; i < cases.length; i++) {
const { matches, result } = await cases[i]();
const { matches, result } = await cases[i]().pipe(take(1)).toPromise();
if (matches) {
return result;
}
}
if (typeof args.default !== 'undefined') {
return await args.default();
}
return input;
return args.default?.().pipe(take(1)).toPromise() ?? input;
},
};
}

2
x-pack/plugins/canvas/public/apps/workpad/routes.ts
View file

@ -70,8 +70,8 @@ export const routes = [
const fetchedWorkpad = await workpadService.get(params.id);
const { assets, ...workpad } = fetchedWorkpad;
dispatch(setWorkpad(workpad));
dispatch(setAssets(assets));
dispatch(setWorkpad(workpad));
// reset transient properties when changing workpads
dispatch(setZoomScale(1));