demonstrate new publisher

yaacovCR · yaacovCR · commit e8f1f9e14258 · 2022-11-25T00:50:44.000+02:00
depends on #3784 The proposed new publisher does not use the event loop to manage AsyncRecord dependencies => and so if multiple items within a stream are released from publishing because their parent has just been published, they are all released at once. Another difference is that different sets are used to store the AsyncRecords that are pending vs ready for publishing, etc. This provides a performance benefit in that on a call to next, the set of all AsyncRecords is not inspected. As a side-effect of this change, the incremental array is ordered by which items are ready for delivery first, and not by the initial document. The subscribe algorithm used does not use Promise.race -- this may also be beneficial as the implementation of Promise.race within V8 has a known memory leak for long-running promises. (see https://bugs.chromium.org/p/v8/issues/detail?id=9858)
diff --git a/src/execution/__tests__/defer-test.ts b/src/execution/__tests__/defer-test.ts
@@ -605,11 +605,6 @@ describe('Execute: defer directive', () => {
             data: { slowField: 'slow', friends: [{}, {}, {}] },
             path: ['hero'],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           { data: { name: 'Han' }, path: ['hero', 'friends', 0] },
           { data: { name: 'Leia' }, path: ['hero', 'friends', 1] },
           { data: { name: 'C-3PO' }, path: ['hero', 'friends', 2] },
@@ -653,11 +648,6 @@ describe('Execute: defer directive', () => {
             },
             path: ['hero'],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           { data: { name: 'Han' }, path: ['hero', 'friends', 0] },
           { data: { name: 'Leia' }, path: ['hero', 'friends', 1] },
           { data: { name: 'C-3PO' }, path: ['hero', 'friends', 2] },
diff --git a/src/execution/__tests__/stream-test.ts b/src/execution/__tests__/stream-test.ts
@@ -151,11 +151,10 @@ describe('Execute: stream directive', () => {
         hasNext: true,
       },
       {
-        incremental: [{ items: ['banana'], path: ['scalarList', 1] }],
-        hasNext: true,
-      },
-      {
-        incremental: [{ items: ['coconut'], path: ['scalarList', 2] }],
+        incremental: [
+          { items: ['banana'], path: ['scalarList', 1] },
+          { items: ['coconut'], path: ['scalarList', 2] },
+        ],
         hasNext: false,
       },
     ]);
@@ -173,15 +172,11 @@ describe('Execute: stream directive', () => {
         hasNext: true,
       },
       {
-        incremental: [{ items: ['apple'], path: ['scalarList', 0] }],
-        hasNext: true,
-      },
-      {
-        incremental: [{ items: ['banana'], path: ['scalarList', 1] }],
-        hasNext: true,
-      },
-      {
-        incremental: [{ items: ['coconut'], path: ['scalarList', 2] }],
+        incremental: [
+          { items: ['apple'], path: ['scalarList', 0] },
+          { items: ['banana'], path: ['scalarList', 1] },
+          { items: ['coconut'], path: ['scalarList', 2] },
+        ],
         hasNext: false,
       },
     ]);
@@ -230,11 +225,6 @@ describe('Execute: stream directive', () => {
             path: ['scalarList', 1],
             label: 'scalar-stream',
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: ['coconut'],
             path: ['scalarList', 2],
@@ -296,11 +286,6 @@ describe('Execute: stream directive', () => {
             items: [['banana', 'banana', 'banana']],
             path: ['scalarListList', 1],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [['coconut', 'coconut', 'coconut']],
             path: ['scalarListList', 2],
@@ -379,20 +364,10 @@ describe('Execute: stream directive', () => {
             items: [{ name: 'Luke', id: '1' }],
             path: ['friendList', 0],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ name: 'Han', id: '2' }],
             path: ['friendList', 1],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ name: 'Leia', id: '3' }],
             path: ['friendList', 2],
@@ -483,11 +458,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ name: 'Leia', id: '3' }],
             path: ['friendList', 2],
@@ -585,9 +555,6 @@ describe('Execute: stream directive', () => {
             path: ['friendList', 2],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -627,7 +594,7 @@ describe('Execute: stream directive', () => {
         }
       }
     `);
-    const result = await completeAsync(document, 3, {
+    const result = await completeAsync(document, 2, {
       async *friendList() {
         yield await Promise.resolve(friends[0]);
         yield await Promise.resolve(friends[1]);
@@ -656,10 +623,9 @@ describe('Execute: stream directive', () => {
               path: ['friendList', 2],
             },
           ],
-          hasNext: true,
+          hasNext: false,
         },
       },
-      { done: false, value: { hasNext: false } },
       { done: true, value: undefined },
     ]);
   });
@@ -887,11 +853,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ nonNullName: 'Han' }],
             path: ['friendList', 2],
@@ -980,19 +941,11 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ nonNullName: 'Han' }],
             path: ['friendList', 2],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -1140,6 +1093,10 @@ describe('Execute: stream directive', () => {
       },
       {
         incremental: [
+          {
+            items: [{ name: 'Luke' }],
+            path: ['nestedObject', 'nestedFriendList', 0],
+          },
           {
             data: { scalarField: null },
             path: ['otherNestedObject'],
@@ -1151,10 +1108,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-          {
-            items: [{ name: 'Luke' }],
-            path: ['nestedObject', 'nestedFriendList', 0],
-          },
         ],
         hasNext: false,
       },
@@ -1258,9 +1211,6 @@ describe('Execute: stream directive', () => {
             ],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -1407,9 +1357,6 @@ describe('Execute: stream directive', () => {
             path: ['friendList', 2],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -1463,28 +1410,10 @@ describe('Execute: stream directive', () => {
             data: { scalarField: 'slow', nestedFriendList: [] },
             path: ['nestedObject'],
           },
-        ],
-        hasNext: true,
-      },
-      done: false,
-    });
-    const result3 = await iterator.next();
-    expectJSON(result3).toDeepEqual({
-      value: {
-        incremental: [
           {
             items: [{ name: 'Luke' }],
             path: ['nestedObject', 'nestedFriendList', 0],
           },
-        ],
-        hasNext: true,
-      },
-      done: false,
-    });
-    const result4 = await iterator.next();
-    expectJSON(result4).toDeepEqual({
-      value: {
-        incremental: [
           {
             items: [{ name: 'Han' }],
             path: ['nestedObject', 'nestedFriendList', 1],
diff --git a/src/execution/execute.ts b/src/execution/execute.ts
@@ -2150,12 +2150,13 @@ class DeferredFragmentRecord {
   errors: Array<GraphQLError>;
   label: string | undefined;
   path: Array<string | number>;
-  promise: Promise<void>;
   data: ObjMap<unknown> | null;
   parentContext: AsyncPayloadRecord | undefined;
-  isCompleted: boolean;
-  _exeContext: ExecutionContext;
-  _resolve?: (arg: PromiseOrValue<ObjMap<unknown> | null>) => void;
+  _publisher: Publisher<
+    AsyncPayloadRecord,
+    IncrementalResult,
+    SubsequentIncrementalExecutionResult
+  >;
   constructor(opts: {
     label: string | undefined;
     path: Path | undefined;
@@ -2167,27 +2168,21 @@ class DeferredFragmentRecord {
     this.path = pathToArray(opts.path);
     this.parentContext = opts.parentContext;
     this.errors = [];
-    this._exeContext = opts.exeContext;
-    this._exeContext.publisher.add(this);
-    this.isCompleted = false;
+    this._publisher = opts.exeContext.publisher;
+    this._publisher.add(this);
     this.data = null;
-    this.promise = new Promise<ObjMap<unknown> | null>((resolve) => {
-      this._resolve = (promiseOrValue) => {
-        resolve(promiseOrValue);
-      };
-    }).then((data) => {
-      this.data = data;
-      this.isCompleted = true;
-    });
   }
 
   addData(data: PromiseOrValue<ObjMap<unknown> | null>) {
-    const parentData = this.parentContext?.promise;
-    if (parentData) {
-      this._resolve?.(parentData.then(() => data));
+    if (isPromise(data)) {
+      data.then((resolved) => {
+        this.data = resolved;
+        this._publisher.complete(this);
+      });
       return;
     }
-    this._resolve?.(data);
+    this.data = data;
+    this._publisher.complete(this);
   }
 }
 
@@ -2197,13 +2192,14 @@ class StreamRecord {
   label: string | undefined;
   path: Array<string | number>;
   items: Array<unknown> | null;
-  promise: Promise<void>;
   parentContext: AsyncPayloadRecord | undefined;
   iterator: AsyncIterator<unknown> | undefined;
   isCompletedIterator?: boolean;
-  isCompleted: boolean;
-  _exeContext: ExecutionContext;
-  _resolve?: (arg: PromiseOrValue<Array<unknown> | null>) => void;
+  _publisher: Publisher<
+    AsyncPayloadRecord,
+    IncrementalResult,
+    SubsequentIncrementalExecutionResult
+  >;
   constructor(opts: {
     label: string | undefined;
     path: Path | undefined;
@@ -2218,27 +2214,21 @@ class StreamRecord {
     this.parentContext = opts.parentContext;
     this.iterator = opts.iterator;
     this.errors = [];
-    this._exeContext = opts.exeContext;
-    this._exeContext.publisher.add(this);
-    this.isCompleted = false;
+    this._publisher = opts.exeContext.publisher;
+    this._publisher.add(this);
     this.items = null;
-    this.promise = new Promise<Array<unknown> | null>((resolve) => {
-      this._resolve = (promiseOrValue) => {
-        resolve(promiseOrValue);
-      };
-    }).then((items) => {
-      this.items = items;
-      this.isCompleted = true;
-    });
   }
 
   addItems(items: PromiseOrValue<Array<unknown> | null>) {
-    const parentData = this.parentContext?.promise;
-    if (parentData) {
-      this._resolve?.(parentData.then(() => items));
+    if (isPromise(items)) {
+      items.then((resolved) => {
+        this.items = resolved;
+        this._publisher.complete(this);
+      });
       return;
     }
-    this._resolve?.(items);
+    this.items = items;
+    this._publisher.complete(this);
   }
 
   setIsCompletedIterator() {
diff --git a/src/execution/publisher.ts b/src/execution/publisher.ts
