demonstrate new publisher

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)

Author: yaacovCR

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] },

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,11 +941,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ nonNullName: 'Han' }],
             path: ['friendList', 2],
@@ -1140,6 +1096,10 @@ describe('Execute: stream directive', () => {
       },
       {
         incremental: [
+          {
+            items: [{ name: 'Luke' }],
+            path: ['nestedObject', 'nestedFriendList', 0],
+          },
           {
             data: { scalarField: null },
             path: ['otherNestedObject'],
@@ -1151,10 +1111,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-          {
-            items: [{ name: 'Luke' }],
-            path: ['nestedObject', 'nestedFriendList', 0],
-          },
         ],
         hasNext: false,
       },
@@ -1258,9 +1214,6 @@ describe('Execute: stream directive', () => {
             ],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -1407,9 +1360,6 @@ describe('Execute: stream directive', () => {
             path: ['friendList', 2],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -1463,15 +1413,6 @@ 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],
@@ -1481,8 +1422,8 @@ describe('Execute: stream directive', () => {
       },
       done: false,
     });
-    const result4 = await iterator.next();
-    expectJSON(result4).toDeepEqual({
+    const result3 = await iterator.next();
+    expectJSON(result3).toDeepEqual({
       value: {
         incremental: [
           {
@@ -1494,13 +1435,13 @@ describe('Execute: stream directive', () => {
       },
       done: false,
     });
-    const result5 = await iterator.next();
-    expectJSON(result5).toDeepEqual({
+    const result4 = await iterator.next();
+    expectJSON(result4).toDeepEqual({
       value: { hasNext: false },
       done: false,
     });
-    const result6 = await iterator.next();
-    expectJSON(result6).toDeepEqual({
+    const result5 = await iterator.next();
+    expectJSON(result5).toDeepEqual({
       value: undefined,
       done: true,
     });