feat(subscriptions): trying to figure out subs

I added a streamCreated sub to try it out,
but I can't get it to connect 😅

app.js

@@ -22,7 +22,6 @@ let graphqlServer
  * @return {[type]} an express applicaiton and the graphql server
  */
 exports.init = async ( ) => {
-
   const app = express( )
 
   await knex.migrate.latest( )
@@ -65,6 +64,9 @@ exports.init = async ( ) => {
 
 const setupCheck = require( `${appRoot}/setupcheck` )
 const { createProxyMiddleware } = require( 'http-proxy-middleware' )
+const { PubSub } = require( 'graphql-subscriptions' )
+const { SubscriptionServer } = require( 'subscriptions-transport-ws' )
+const { execute, subscribe } = require( 'graphql' )
 
 exports.startHttp = async ( app ) => {
   let port = process.env.PORT || 3000
@@ -85,7 +87,6 @@ exports.startHttp = async ( app ) => {
     debug( 'speckle:http-startup' )( `👉 setup application: http://localhost:${port}/setup` )
     debug( 'speckle:hint' )( `ℹ️  Don't forget to run "npm run dev:frontend" in a different terminal to start the vue application.` )
   } else {
-
     app.use( '/', express.static( `${appRoot}/frontend/dist` ) )
 
     app.all( '/auth*', async ( req, res ) => {
@@ -97,9 +98,8 @@ exports.startHttp = async ( app ) => {
     } )
 
     app.all( '*', async ( req, res ) => {
-
       try {
-        // refrehsing this variable on every request only if it's false  
+        // refrehsing this variable on every request only if it's false
         if ( !setupComplete ) {
           setupComplete = await setupCheck( )
         }
@@ -115,6 +115,7 @@ exports.startHttp = async ( app ) => {
     } );
   }
 
+  const pubsub = new PubSub()
   let server = http.createServer( app )
 
   graphqlServer.installSubscriptionHandlers( server )
@@ -123,7 +124,16 @@ exports.startHttp = async ( app ) => {
     debug( `speckle:startup` )( `Listening on ${server.address().port}` )
   } )
 
-  server.listen( port )
+  server.listen( port, () => {
+    new SubscriptionServer( {
+      execute,
+      subscribe,
+      //schema: graphqlSchema -- not sure how to get the full schema?
+    }, {
+      server: server,
+      path: '/subscriptions'
+    } )
+  } )
 
   return { server }
-}
+}

frontend/src/vue-apollo.js

@@ -1,6 +1,7 @@
 import Vue from 'vue'
 import VueApollo from 'vue-apollo'
 import { createApolloClient, restartWebsockets } from 'vue-cli-plugin-apollo/graphql-client'
+import { SubscriptionClient } from 'subscriptions-transport-ws';
 
 // Install the vue plugin
 Vue.use( VueApollo )
@@ -11,6 +12,11 @@ const AUTH_TOKEN = 'AuthToken'
 // Http endpoint
 const httpEndpoint = process.env.VUE_APP_GRAPHQL_HTTP || 'http://localhost:3000/graphql'
 
+// Subscription Client
+const subscriptionClient = new SubscriptionClient(httpEndpoint, {
+  reconnect: true
+})
+
 // Config
 const defaultOptions = {
   // You can use `https` for secure connection (recommended in production)
@@ -27,6 +33,8 @@ const defaultOptions = {
   websocketsOnly: false,
   // Is being rendered on the server?
   ssr: false,
+  // Subscription Client
+  networkInterface: subscriptionClient
 
   // Override default apollo link
   // note: don't override httpLink here, specify httpLink options in the

modules/core/graph/resolvers/streams.js

@@ -13,6 +13,9 @@ const {
   revokePermissionsStream
 } = require( '../../services/streams' )
 const { validateServerRole, validateScopes, authorizeResolver } = require( `${appRoot}/modules/shared` )
+const { pubsub } = require( `${appRoot}/app` )
+
+const STREAM_CREATED = 'STREAM_CREATED'
 
 module.exports = {
   Query: {
@@ -29,10 +32,10 @@ module.exports = {
       if ( args.limit && args.limit > 100 )
         throw new UserInputError( 'Cannot return more than 100 items, please use pagination.' )
 
-      let totalCount = await getUserStreamsCount( {userId: context.userId, publicOnly: false, searchQuery: args.query} )
+      let totalCount = await getUserStreamsCount( { userId: context.userId, publicOnly: false, searchQuery: args.query } )
 
-      let {cursor, streams} = await getUserStreams( {userId: context.userId, limit: args.limit, cursor: args.cursor, publicOnly: false, searchQuery: args.query} )
-      return {totalCount, cursor: cursor, items: streams}
+      let { cursor, streams } = await getUserStreams( { userId: context.userId, limit: args.limit, cursor: args.cursor, publicOnly: false, searchQuery: args.query } )
+      return { totalCount, cursor: cursor, items: streams }
     }
   },
   Stream: {
@@ -62,6 +65,7 @@ module.exports = {
       await validateServerRole( context, 'server:user' )
       await validateScopes( context.scopes, 'streams:write' )
 
+      await pubsub.publish( STREAM_CREATED, { streamCreated: args } )
       let id = await createStream( { ...args.stream, ownerId: context.userId } )
       return id
     },
@@ -101,5 +105,10 @@ module.exports = {
 
       return await revokePermissionsStream( { ...args } )
     }
+  },
+  Subscription: {
+    streamCreated: {
+      subscribe: () => pubsub.asyncIterator( [ STREAM_CREATED ] )
+    }
   }
 }

modules/core/graph/schemas/streams.graphql

@@ -60,6 +60,13 @@ extend type Mutation {
 
 }
 
+extend type Subscription {
+  """
+  Subscribes to new stream creation.
+  """
+  streamCreated: Stream!
+}
+
 input StreamCreateInput {
   name: String
   description: String

modules/index.js

@@ -9,7 +9,6 @@ const debug = require( 'debug' )( 'speckle:modules' )
 const { scalarResolvers, scalarSchemas } = require( './core/graph/scalars' )
 
 exports.init = async ( app ) => {
-
   let dirs = fs.readdirSync( `${appRoot}/modules` )
   let moduleDirs = [ ]
 
@@ -27,7 +26,6 @@ exports.init = async ( app ) => {
   moduleDirs.forEach( async dir => {
     await require( dir ).init( app )
   } )
-
 }
 
 exports.graph = ( ) => {
@@ -47,8 +45,14 @@ exports.graph = ( ) => {
       The void stares back.
       """
       _: String
+      }
+      type Subscription{
+        """
+        It's lonely in the void.
+        """
+        _: String
       }`
-    ]
+  ]
 
   let resolverObjs = [ ]
   // let directiveDirs = [ ]
@@ -74,5 +78,4 @@ exports.graph = ( ) => {
   } )
 
   return { resolvers, typeDefs }
-
-}
+}