WiP: add a subscription example

services/products/index.js

@@ -10,6 +10,8 @@ const http = require("http");
 const { json } = require("body-parser");
 const cors = require("cors");
 const { parse } = require("graphql");
+const { WebSocketServer } = require("ws");
+const { useServer } = require("graphql-ws/lib/use/ws");
 
 const rateLimitTreshold = process.env.LIMIT || 5000;
 
@@ -26,6 +28,10 @@ const typeDefs = parse(`#graphql
     createProduct(upc: ID!, name: String): Product
   }
 
+  type Subscription {
+    productUpdate: Product
+  }
+
   type Product @key(fields: "upc") {
     upc: String!
     name: String
@@ -84,6 +90,24 @@ const resolvers = {
       };
     },
   },
+  Subscription: {
+    productUpdate: {
+      subscribe: async function* () {
+        for(count = 0; count < 20; count++){
+          let product = products[Math.floor(Math.random()*products.length)];
+          let newProduct = {
+            upc: product.upc,
+            name: product.name,
+            price: Math.floor(Math.random() * 2000),
+            weight: Math.floor(Math.random() * 1000)
+          }
+
+          yield { productUpdate: newProduct };
+          await new Promise((resolve) => setTimeout(resolve, 3000));
+        }
+      },
+    },
+  },
 };
 
 async function startApolloServer(typeDefs, resolvers) {
@@ -95,17 +119,34 @@ async function startApolloServer(typeDefs, resolvers) {
     max: rateLimitTreshold,
   });
 
+  const schema = buildSubgraphSchema([
+    {
+      typeDefs,
+      resolvers,
+    },
+  ]);
   const httpServer = http.createServer(app);
+  const wsServer = new WebSocketServer({
+    server: httpServer,
+    path: "/subscriptions",
+  });
+  const serverCleanup = useServer({ schema }, wsServer);
 
   const server = new ApolloServer({
-    schema: buildSubgraphSchema([
+    schema,
+    allowBatchedHttpRequests: true,
+    plugins: [
+      ApolloServerPluginDrainHttpServer({ httpServer },
       {
-        typeDefs,
-        resolvers,
+        async serverWillStart() {
+          return {
+            async drainServer() {
+              await serverCleanup.dispose();
+            },
+          };
+        },
       },
-    ]),
-    allowBatchedHttpRequests: true,
-    plugins: [ApolloServerPluginDrainHttpServer({ httpServer })],
+    )],
   });
 
   await server.start();

services/products/package.json

@@ -10,6 +10,8 @@
     "cors": "^2.8.5",
     "express": "^4.17.1",
     "express-rate-limit": "^5.5.1",
-    "graphql": "^16.8.1"
+    "graphql": "^16.8.1",
+    "graphql-ws": "^5.12.1",
+    "ws": "^8.13.0"
   }
 }

supergraph.graphql

@@ -4,6 +4,7 @@ schema
 {
   query: Query
   mutation: Mutation
+  subscription: Subscription
 }
 
 directive @join__enumValue(graph: join__Graph!) repeatable on ENUM_VALUE
@@ -87,6 +88,12 @@ type Review
   product: Product
 }
 
+type Subscription
+  @join__type(graph: PRODUCTS)
+{
+  productUpdate: Product
+}
+
 type User
   @join__type(graph: ACCOUNTS, key: "id")
   @join__type(graph: REVIEWS, key: "id")