Skip to content

Commit 290f1d9

Browse files
yaacovCRyaacovCR
committed
the IncrementalPublisher should handle response building
1 parent cfbe5a4 commit 290f1d9

File tree

10 files changed

+212
-221
lines changed

10 files changed

+212
-221
lines changed

‎src/execution/IncrementalPublisher.ts

Lines changed: 183 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,72 @@
1+
import { isPromise } from '../jsutils/isPromise.js';
12
import type { ObjMap } from '../jsutils/ObjMap.js';
23
import type { Path } from '../jsutils/Path.js';
34
import { pathToArray } from '../jsutils/Path.js';
5+
import type { PromiseOrValue } from '../jsutils/PromiseOrValue.js';
46
import { promiseWithResolvers } from '../jsutils/promiseWithResolvers.js';
57

68
import type {
79
GraphQLError,
810
GraphQLFormattedError,
911
} from '../error/GraphQLError.js';
1012

13+
/**
14+
* The result of GraphQL execution.
15+
*
16+
* - `errors` is included when any errors occurred as a non-empty array.
17+
* - `data` is the result of a successful execution of the query.
18+
* - `hasNext` is true if a future payload is expected.
19+
* - `extensions` is reserved for adding non-standard properties.
20+
* - `incremental` is a list of the results from defer/stream directives.
21+
*/
22+
export interface ExecutionResult<
23+
TData = ObjMap<unknown>,
24+
TExtensions = ObjMap<unknown>,
25+
> {
26+
errors?: ReadonlyArray<GraphQLError>;
27+
data?: TData | null;
28+
extensions?: TExtensions;
29+
}
30+
31+
export interface FormattedExecutionResult<
32+
TData = ObjMap<unknown>,
33+
TExtensions = ObjMap<unknown>,
34+
> {
35+
errors?: ReadonlyArray<GraphQLFormattedError>;
36+
data?: TData | null;
37+
extensions?: TExtensions;
38+
}
39+
40+
export interface ExperimentalIncrementalExecutionResults<
41+
TData = ObjMap<unknown>,
42+
TExtensions = ObjMap<unknown>,
43+
> {
44+
initialResult: InitialIncrementalExecutionResult<TData, TExtensions>;
45+
subsequentResults: AsyncGenerator<
46+
SubsequentIncrementalExecutionResult<TData, TExtensions>,
47+
void,
48+
void
49+
>;
50+
}
51+
52+
export interface InitialIncrementalExecutionResult<
53+
TData = ObjMap<unknown>,
54+
TExtensions = ObjMap<unknown>,
55+
> extends ExecutionResult<TData, TExtensions> {
56+
hasNext: boolean;
57+
incremental?: ReadonlyArray<IncrementalResult<TData, TExtensions>>;
58+
extensions?: TExtensions;
59+
}
60+
61+
export interface FormattedInitialIncrementalExecutionResult<
62+
TData = ObjMap<unknown>,
63+
TExtensions = ObjMap<unknown>,
64+
> extends FormattedExecutionResult<TData, TExtensions> {
65+
hasNext: boolean;
66+
incremental?: ReadonlyArray<FormattedIncrementalResult<TData, TExtensions>>;
67+
extensions?: TExtensions;
68+
}
69+
1170
export interface SubsequentIncrementalExecutionResult<
1271
TData = ObjMap<unknown>,
1372
TExtensions = ObjMap<unknown>,
@@ -113,86 +172,6 @@ export class IncrementalPublisher {
113172
this._reset();
114173
}
115174

116-
hasNext(): boolean {
117-
return this._pending.size > 0;
118-
}
119-
120-
subscribe(): AsyncGenerator<
121-
SubsequentIncrementalExecutionResult,
122-
void,
123-
void
124-
> {
125-
let isDone = false;
126-
127-
const _next = async (): Promise<
128-
IteratorResult<SubsequentIncrementalExecutionResult, void>
129-
> => {
130-
// eslint-disable-next-line no-constant-condition
131-
while (true) {
132-
if (isDone) {
133-
return { value: undefined, done: true };
134-
}
135-
136-
for (const item of this._released) {
137-
this._pending.delete(item);
138-
}
139-
const released = this._released;
140-
this._released = new Set();
141-
142-
const result = this._getIncrementalResult(released);
143-
144-
if (!this.hasNext()) {
145-
isDone = true;
146-
}
147-
148-
if (result !== undefined) {
149-
return { value: result, done: false };
150-
}
151-
152-
// eslint-disable-next-line no-await-in-loop
153-
await this._signalled;
154-
}
155-
};
156-
157-
const returnStreamIterators = async (): Promise<void> => {
158-
const promises: Array<Promise<IteratorResult<unknown>>> = [];
159-
this._pending.forEach((incrementalDataRecord) => {
160-
if (
161-
isStreamItemsRecord(incrementalDataRecord) &&
162-
incrementalDataRecord.asyncIterator?.return
163-
) {
164-
promises.push(incrementalDataRecord.asyncIterator.return());
165-
}
166-
});
167-
await Promise.all(promises);
168-
};
169-
170-
const _return = async (): Promise<
171-
IteratorResult<SubsequentIncrementalExecutionResult, void>
172-
> => {
173-
isDone = true;
174-
await returnStreamIterators();
175-
return { value: undefined, done: true };
176-
};
177-
178-
const _throw = async (
179-
error?: unknown,
180-
): Promise<IteratorResult<SubsequentIncrementalExecutionResult, void>> => {
181-
isDone = true;
182-
await returnStreamIterators();
183-
return Promise.reject(error);
184-
};
185-
186-
return {
187-
[Symbol.asyncIterator]() {
188-
return this;
189-
},
190-
next: _next,
191-
return: _return,
192-
throw: _throw,
193-
};
194-
}
195-
196175
prepareInitialResultRecord(): InitialResultRecord {
197176
return {
198177
errors: [],
@@ -256,19 +235,26 @@ export class IncrementalPublisher {
256235
incrementalDataRecord.errors.push(error);
257236
}
258237

259-
publishInitial(initialResult: InitialResultRecord) {
260-
for (const child of initialResult.children) {
261-
if (child.filtered) {
262-
continue;
263-
}
264-
this._publish(child);
238+
handleInitialResultData(
239+
initialResultRecord: InitialResultRecord,
240+
data: PromiseOrValue<ObjMap<unknown>>,
241+
): PromiseOrValue<ExecutionResult | ExperimentalIncrementalExecutionResults> {
242+
if (isPromise(data)) {
243+
return data.then(
244+
(resolved) => this._buildInitialResponse(initialResultRecord, resolved),
245+
(error) => this.handleInitialResultError(initialResultRecord, error),
246+
);
265247
}
248+
return this._buildInitialResponse(initialResultRecord, data);
266249
}
267250

268-
getInitialErrors(
269-
initialResult: InitialResultRecord,
270-
): ReadonlyArray<GraphQLError> {
271-
return initialResult.errors;
251+
handleInitialResultError(
252+
initialResultRecord: InitialResultRecord,
253+
error: GraphQLError,
254+
): ExecutionResult {
255+
const errors = initialResultRecord.errors;
256+
errors.push(error);
257+
return { data: null, errors };
272258
}
273259

274260
filter(nullPath: Path, erroringIncrementalDataRecord: IncrementalDataRecord) {
@@ -301,6 +287,111 @@ export class IncrementalPublisher {
301287
});
302288
}
303289

290+
private _buildInitialResponse(
291+
initialResultRecord: InitialResultRecord,
292+
data: ObjMap<unknown> | null,
293+
): ExecutionResult | ExperimentalIncrementalExecutionResults {
294+
for (const child of initialResultRecord.children) {
295+
if (child.filtered) {
296+
continue;
297+
}
298+
this._publish(child);
299+
}
300+
301+
const errors = initialResultRecord.errors;
302+
const initialResult = errors.length === 0 ? { data } : { errors, data };
303+
if (this._hasNext()) {
304+
return {
305+
initialResult: {
306+
...initialResult,
307+
hasNext: true,
308+
},
309+
subsequentResults: this._subscribe(),
310+
};
311+
}
312+
return initialResult;
313+
}
314+
315+
private _hasNext(): boolean {
316+
return this._pending.size > 0;
317+
}
318+
319+
private _subscribe(): AsyncGenerator<
320+
SubsequentIncrementalExecutionResult,
321+
void,
322+
void
323+
> {
324+
let isDone = false;
325+
326+
const _next = async (): Promise<
327+
IteratorResult<SubsequentIncrementalExecutionResult, void>
328+
> => {
329+
// eslint-disable-next-line no-constant-condition
330+
while (true) {
331+
if (isDone) {
332+
return { value: undefined, done: true };
333+
}
334+
335+
for (const item of this._released) {
336+
this._pending.delete(item);
337+
}
338+
const released = this._released;
339+
this._released = new Set();
340+
341+
const result = this._getIncrementalResult(released);
342+
343+
if (!this._hasNext()) {
344+
isDone = true;
345+
}
346+
347+
if (result !== undefined) {
348+
return { value: result, done: false };
349+
}
350+
351+
// eslint-disable-next-line no-await-in-loop
352+
await this._signalled;
353+
}
354+
};
355+
356+
const returnStreamIterators = async (): Promise<void> => {
357+
const promises: Array<Promise<IteratorResult<unknown>>> = [];
358+
this._pending.forEach((incrementalDataRecord) => {
359+
if (
360+
isStreamItemsRecord(incrementalDataRecord) &&
361+
incrementalDataRecord.asyncIterator?.return
362+
) {
363+
promises.push(incrementalDataRecord.asyncIterator.return());
364+
}
365+
});
366+
await Promise.all(promises);
367+
};
368+
369+
const _return = async (): Promise<
370+
IteratorResult<SubsequentIncrementalExecutionResult, void>
371+
> => {
372+
isDone = true;
373+
await returnStreamIterators();
374+
return { value: undefined, done: true };
375+
};
376+
377+
const _throw = async (
378+
error?: unknown,
379+
): Promise<IteratorResult<SubsequentIncrementalExecutionResult, void>> => {
380+
isDone = true;
381+
await returnStreamIterators();
382+
return Promise.reject(error);
383+
};
384+
385+
return {
386+
[Symbol.asyncIterator]() {
387+
return this;
388+
},
389+
next: _next,
390+
return: _return,
391+
throw: _throw,
392+
};
393+
}
394+
304395
private _trigger() {
305396
this._resolve();
306397
this._reset();
@@ -369,8 +460,8 @@ export class IncrementalPublisher {
369460
}
370461

371462
return incrementalResults.length
372-
? { incremental: incrementalResults, hasNext: this.hasNext() }
373-
: encounteredCompletedAsyncIterator && !this.hasNext()
463+
? { incremental: incrementalResults, hasNext: this._hasNext() }
464+
: encounteredCompletedAsyncIterator && !this._hasNext()
374465
? { hasNext: false }
375466
: undefined;
376467
}

‎src/execution/__tests__/defer-test.ts

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,11 @@ import {
1616
import { GraphQLID, GraphQLString } from '../../type/scalars.js';
1717
import { GraphQLSchema } from '../../type/schema.js';
1818

19-
import type { InitialIncrementalExecutionResult } from '../execute.js';
2019
import { execute, experimentalExecuteIncrementally } from '../execute.js';
21-
import type { SubsequentIncrementalExecutionResult } from '../IncrementalPublisher.js';
20+
import type {
21+
InitialIncrementalExecutionResult,
22+
SubsequentIncrementalExecutionResult,
23+
} from '../IncrementalPublisher.js';
2224

2325
const friendType = new GraphQLObjectType({
2426
fields: {

‎src/execution/__tests__/lists-test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,8 @@ import { GraphQLSchema } from '../../type/schema.js';
1818

1919
import { buildSchema } from '../../utilities/buildASTSchema.js';
2020

21-
import type { ExecutionResult } from '../execute.js';
2221
import { execute, executeSync } from '../execute.js';
22+
import type { ExecutionResult } from '../IncrementalPublisher.js';
2323

2424
describe('Execute: Accepts any iterable as list value', () => {
2525
function complete(rootValue: unknown) {

‎src/execution/__tests__/nonnull-test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,8 @@ import { GraphQLSchema } from '../../type/schema.js';
1313

1414
import { buildSchema } from '../../utilities/buildASTSchema.js';
1515

16-
import type { ExecutionResult } from '../execute.js';
1716
import { execute, executeSync } from '../execute.js';
17+
import type { ExecutionResult } from '../IncrementalPublisher.js';
1818

1919
const syncError = new Error('sync');
2020
const syncNonNullError = new Error('syncNonNull');

‎src/execution/__tests__/oneof-test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,8 @@ import { parse } from '../../language/parser.js';
66

77
import { buildSchema } from '../../utilities/buildASTSchema.js';
88

9-
import type { ExecutionResult } from '../execute.js';
109
import { execute } from '../execute.js';
10+
import type { ExecutionResult } from '../IncrementalPublisher.js';
1111

1212
const schema = buildSchema(`
1313
type Query {

‎src/execution/__tests__/stream-test.ts

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,11 @@ import {
1717
import { GraphQLID, GraphQLString } from '../../type/scalars.js';
1818
import { GraphQLSchema } from '../../type/schema.js';
1919

20-
import type { InitialIncrementalExecutionResult } from '../execute.js';
2120
import { experimentalExecuteIncrementally } from '../execute.js';
22-
import type { SubsequentIncrementalExecutionResult } from '../IncrementalPublisher.js';
21+
import type {
22+
InitialIncrementalExecutionResult,
23+
SubsequentIncrementalExecutionResult,
24+
} from '../IncrementalPublisher.js';
2325

2426
const friendType = new GraphQLObjectType({
2527
fields: {

‎src/execution/__tests__/subscribe-test.ts

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,9 @@ import {
2020
} from '../../type/scalars.js';
2121
import { GraphQLSchema } from '../../type/schema.js';
2222

23-
import type { ExecutionArgs, ExecutionResult } from '../execute.js';
23+
import type { ExecutionArgs } from '../execute.js';
2424
import { createSourceEventStream, subscribe } from '../execute.js';
25+
import type { ExecutionResult } from '../IncrementalPublisher.js';
2526

2627
import { SimplePubSub } from './simplePubSub.js';
2728

0 commit comments

Comments
 (0)