Use subscription.preProcessor to compose topic ; update tests

src/engine/subscription/Subscription.spec.ts

@@ -337,41 +337,56 @@ describe('Subscription', () => {
   });
 
   describe('preProcessor', () => {
-    it('should pass through preProcessor if it is declared', async () => {
-      const testEntity = new Entity({
-        name: 'SomeTestsEntityName',
-        description: 'Just some description',
-        attributes: {
-          someAttribute: {
-            type: DataTypeString,
-            description: 'Just some description',
-            required: true,
-          },
-          anotherAttribute: {
-            type: DataTypeString,
-            description: 'Just some description',
-          },
+    const testEntity = new Entity({
+      name: 'SomeTestsEntityName',
+      description: 'Just some description',
+      attributes: {
+        someAttribute: {
+          type: DataTypeString,
+          description: 'Just some description',
+          required: false,
         },
-        subscriptions: [
-          new Subscription({
-            name: 'SomeSubWithPreProcessor',
-            type: SUBSCRIPTION_TYPE_CREATE,
-            description: 'build item',
-            attributes: ['someAttribute'],
-            delimiter: '/',
-            wildCard: '',
-            pattern: '',
-            // pattern: 'someAttribute/anotherAttribute',
-            preProcessor: (_entity, _source, payload) => {
-              if (payload === 13) {
-                throw new Error('13 brings bad luck');
-              }
-              return payload;
-            },
-          }),
-        ],
-      });
+        anotherAttribute: {
+          type: DataTypeString,
+          description: 'Just some description',
+        },
+      },
+      subscriptions: [
+        new Subscription({
+          name: 'SomeSubWithPreProcessor',
+          type: SUBSCRIPTION_TYPE_CREATE,
+          description: 'build item',
+          attributes: ['someAttribute'],
+          delimiter: '/',
+          // wildCard: '',
+          // pattern: '',
+          preProcessor: (
+            _entity,
+            _id,
+            _source,
+            input,
+            typeName,
+            entitySubscription,
+          ) => {
+            if (entitySubscription.attributes && Object.keys(input).length) {
+              const delimiter = entitySubscription.delimiter;
+              const filled = entitySubscription.attributes
+                .map(attribute => input[attribute])
+                .reduce((acc, curr) => `${acc + delimiter + curr}`, '');
+
+              const topic = `${entitySubscription.name}${_entity.name}${filled}`;
+              return topic;
+            }
+
+            return null;
+          },
+        }),
+      ],
+    });
 
+    const someAttribute = 'test';
+
+    it('should pass through preProcessor if it is declared', async () => {
       const subscriptionByName = testEntity.getSubscriptionByName(
         'SomeSubWithPreProcessor',
       );
@@ -386,19 +401,19 @@ describe('Subscription', () => {
         subscription someSubWithPreProcessorSomeTestsEntityName($input: SomeSubWithPreProcessorSomeTestsEntityNameInput!) {
           someSubWithPreProcessorSomeTestsEntityName(input: $input) {
             someTestsEntityName {
-              someAttribute
+              # someAttribute
+              anotherAttribute
             }
           }
         }`);
 
-      const subscription = (await subscribe({
+      // subscription topic will be automatically generated by shyft
+      let subscription = (await subscribe({
         schema: graphqlSchema,
         document: subscriptionDoc,
         variableValues: {
           input: {
-            someTestsEntityName: {
-              someAttribute: 'test',
-            },
+            someTestsEntityName: {},
           },
         },
         contextValue: { pubsub },
@@ -408,20 +423,45 @@ describe('Subscription', () => {
       //   console.log({ errors: subscription.errors });
       // }
 
-      const pending = subscription.next();
+      let pending = subscription.next();
 
       await pubsub.publish('SomeSubWithPreProcessorSomeTestsEntityName', {
         // someTestsEntityName: {
-        someAttribute: 'hello',
+        someAttribute,
         anotherAttribute: 'world',
         // },
       });
 
-      const result = await pending;
-      // console.log('should pass through preProcessor if it is declared', result);
-
+      let result = await pending;
       expect(result).toMatchSnapshot();
+      expect(await subscription.return()).toMatchSnapshot();
+
+      // subscription topic will be generated in preProcessor
+      subscription = (await subscribe({
+        schema: graphqlSchema,
+        document: subscriptionDoc,
+        variableValues: {
+          input: {
+            someTestsEntityName: {
+              someAttribute,
+            },
+          },
+        },
+        contextValue: { pubsub },
+      })) as AsyncIterableIterator<any>;
 
+      pending = subscription.next();
+
+      await pubsub.publish(
+        `SomeSubWithPreProcessorSomeTestsEntityName/${someAttribute}`,
+        {
+          someAttribute,
+          anotherAttribute: 'world',
+        },
+      );
+
+      result = await pending;
+      expect(result).toMatchSnapshot();
       expect(await subscription.return()).toMatchSnapshot();
     });
   });

src/engine/subscription/Subscription.ts

@@ -56,7 +56,7 @@ export type SubscriptionSetup = {
     entitySubscription?: Subscription,
     context?: object,
     info?: any,
-  ) => Promise<object | null> | object | null;
+  ) => Promise<string | null> | string | null;
   postProcessor?: (
     entity?: Entity,
     // id,

src/engine/subscription/__snapshots__/Subscription.spec.ts.snap

@@ -23,7 +23,7 @@ Object {
     "data": Object {
       "someSubWithPreProcessorSomeTestsEntityName": Object {
         "someTestsEntityName": Object {
-          "someAttribute": "hello",
+          "anotherAttribute": "world",
         },
       },
     },
@@ -38,6 +38,28 @@ Object {
 }
 `;
 
+exports[`Subscription preProcessor should pass through preProcessor if it is declared 4`] = `
+Object {
+  "done": false,
+  "value": Object {
+    "data": Object {
+      "someSubWithPreProcessorSomeTestsEntityName": Object {
+        "someTestsEntityName": Object {
+          "anotherAttribute": "world",
+        },
+      },
+    },
+  },
+}
+`;
+
+exports[`Subscription preProcessor should pass through preProcessor if it is declared 5`] = `
+Object {
+  "done": true,
+  "value": undefined,
+}
+`;
+
 exports[`Subscription processEntitySubscriptions should throw if provided with an invalid list of subscriptions 1`] = `"Entity 'SomeEntityName' subscriptions definition needs to be an array of subscriptions"`;
 
 exports[`Subscription processEntitySubscriptions should throw if provided with an invalid subscription 1`] = `"Invalid subscription definition for entity 'SomeEntityName' at position '0'"`;

src/graphqlProtocol/resolver.ts

@@ -498,19 +498,6 @@ export const getSubscriptionResolver = (
 
     const id = idResolver({ args });
 
-    if (entitySubscription.preProcessor) {
-      args.input[typeName] = await entitySubscription.preProcessor(
-        entity,
-        id,
-        source,
-        args.input[typeName],
-        typeName,
-        entitySubscription,
-        context,
-        info,
-      );
-    }
-
     if (
       entitySubscription.type === SUBSCRIPTION_TYPE_CREATE ||
       entitySubscription.type === SUBSCRIPTION_TYPE_UPDATE
@@ -545,6 +532,11 @@ export const getSubscriptionResolver = (
 
     let topic;
     if (entitySubscription.pattern) {
+      // const delimiter = entitySubscription.delimiter;
+      // const filled = entitySubscription.attributes
+      //   .map(attribute => input[attribute])
+      //   .reduce((acc, curr) => `${acc + delimiter + curr}`, '');
+
       const params = entitySubscription.pattern
         .split(entitySubscription.delimiter)
         .reduce((acc, curr) => (acc[curr] = args.input[typeName][curr]), {});
@@ -559,7 +551,19 @@ export const getSubscriptionResolver = (
           ? entitySubscription.delimiter + entitySubscription.wildCard
           : ''
       }`;
-    } else {
+    } else if (entitySubscription.preProcessor) {
+      topic = await entitySubscription.preProcessor(
+        entity,
+        id,
+        source,
+        args.input[typeName],
+        typeName,
+        entitySubscription,
+        context,
+        info,
+      );
+    }
+    if (!topic) {
       topic = `${entitySubscription.name}${entity.name}${
         entitySubscription.wildCard
           ? entitySubscription.delimiter + entitySubscription.wildCard
@@ -612,7 +616,7 @@ export const getSubscriptionPayloadResolver = (
       ret[typeName] = result;
     }
 
-    // console.log('getSubscriptionPayloadResolver', { ret });
+    // console.log('getSubscriptionPayloadResolver', JSON.stringify(ret, null, 2));
 
     return ret;
   };