Skip to content

Commit 5b7c9a1

Browse files
yaacovCRyaacovCR
committed
the IncrementalPublisher should handle response building
1 parent c9db779 commit 5b7c9a1

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) {
@@ -302,6 +288,111 @@ export class IncrementalPublisher {
302288
});
303289
}
304290

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

378469
return incrementalResults.length
379-
? { incremental: incrementalResults, hasNext: this.hasNext() }
380-
: encounteredCompletedAsyncIterator && !this.hasNext()
470+
? { incremental: incrementalResults, hasNext: this._hasNext() }
471+
: encounteredCompletedAsyncIterator && !this._hasNext()
381472
? { hasNext: false }
382473
: undefined;
383474
}

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