Subscribe

.gitignore

@@ -6,4 +6,5 @@ node_modules
 *.iml
 *.code-workspace
 .DS_Store
-sakila.db
+sakila.db
+src/jeffNotes.md

package.json

@@ -36,6 +36,7 @@
   "scripts": {
     "build": "rimraf dist && tsc",
     "test": "jest --runInBand --coverage",
+    "testOnly": "jest --runInBand",
     "coveralls": "cat ./coverage/lcov.info | coveralls",
     "lint": "eslint src",
     "format": "prettier --write \"src/**/*.ts\"",
@@ -79,6 +80,7 @@
     "graphql": "15.1.0",
     "graphql-middleware": "4.0.2",
     "graphql-scalars": "1.1.5",
+    "graphql-subscriptions": "^1.1.0",
     "graphql-tag": "2.10.3",
     "husky": "4.2.5",
     "jest": "26.0.1",
@@ -95,4 +97,4 @@
   "peerDependencies": {
     "graphql": "^14.0.0 || ^15.0.0"
   }
-}
+}

src/__tests__/postgres/client.ts

@@ -1,4 +1,5 @@
 import { createSqlmancerClient } from '../../'
 import { SqlmancerClient } from './sqlmancer'
+import { PubSub } from 'graphql-subscriptions'
 
-export const client = createSqlmancerClient<SqlmancerClient>(__dirname + '/schema.ts', require('./knex'))
+export const client = createSqlmancerClient<SqlmancerClient>(__dirname + '/schema.ts', require('./knex'), new PubSub())

src/__tests__/postgres/integration.test.ts

@@ -1,7 +1,10 @@
 /* eslint-disable jest/require-top-level-describe */
 /* eslint-disable no-useless-escape */
-import { graphql, validateSchema } from 'graphql'
-import { schema, client } from './schema'
+import { graphql, validateSchema, subscribe, parse, ExecutionResult } from 'graphql'
+import { schema, client, pubsub } from './schema'
+import { EventEmitter } from 'events'
+import { mockResolveInfo } from '../utilities'
+
 
 const describeMaybeSkip = process.env.DB && !process.env.DB.split(' ').includes('postgres') ? describe.skip : describe
 
@@ -394,4 +397,55 @@ describeMaybeSkip('integration (postgres)', () => {
     expect(errors).toBeUndefined()
     expect(data?.films.some((f: any) => f.sequel && f.sequel.id)).toBe(true)
   })
+
+  test('subscriptions', async () => {
+    const document = parse(`
+      subscription {
+        create 
+      }
+    `)
+
+    const sub = <AsyncIterator<ExecutionResult>>await subscribe(schema, document);
+    expect(sub.next).toBeDefined()
+    setImmediate(() => pubsub.publish('CREATE_ONE', { create: "FLUM!" }))
+
+    const { value: { errors, data } } = await sub.next()
+
+    expect(errors).toBeUndefined()
+    expect(data).toBeDefined()
+    expect(data.create).toBe('FLUM!')
+  })
+
+  // this test times out (sub.next() does not return)
+  test.skip('subscription triggered by mutation', async () => {
+    const document = parse(`
+      subscription {
+        create 
+      }
+    `)
+
+
+    const query = `mutation {
+      createCustomerWithPayload {
+        customer {
+          id
+          email
+        }
+      }
+    }`
+
+    const sub = <AsyncIterator<ExecutionResult>>await subscribe(schema, document);
+    expect(sub.next).toBeDefined()
+
+    const info = await mockResolveInfo(schema, 'Mutation', 'createCustomerWithPayload', query)
+    expect(info).toBeDefined()
+    expect(info.operation).toBeDefined()
+    expect(info.operation.operation).toBeDefined()
+    expect(info.operation.operation).toBe('mutation')
+
+    const { value: { errors, data } } = await sub.next()
+
+    expect(errors).toBeUndefined()
+    expect(data.event).toBeDefined()
+  }, 10000)
 })

src/__tests__/postgres/schema.ts

@@ -5,11 +5,16 @@ import { IResolvers } from 'graphql-tools'
 import knex from './knex'
 import { createSqlmancerClient, makeSqlmancerSchema } from '../..'
 import { SqlmancerClient } from './sqlmancer'
+import { PubSub } from 'graphql-subscriptions'
 
 const typeDefs = gql`
   scalar DateTime
   scalar JSON
   scalar JSONObject
+
+  type Subscription {
+      create: String!
+  }
 
   type Query
     @sqlmancer(
@@ -243,14 +248,24 @@ const typeDefs = gql`
   }
 `
 
-export const client = createSqlmancerClient<SqlmancerClient>(__filename, knex)
+const pubsub = new PubSub()
+export const client = createSqlmancerClient<SqlmancerClient>(__filename, knex, pubsub)
 
 const { Film, Actor, Customer, Address, Movie, Person } = client.models
 
 const resolvers: IResolvers = {
   DateTime: GraphQLDateTime,
   JSON: GraphQLJSON,
   JSONObject: GraphQLJSONObject,
+
+  Subscription: {
+    create: {
+      subscribe: () => {
+        return pubsub.asyncIterator('CREATE_ONE');
+      }
+    }
+  },
+
   Query: {
     actors: (_root, _args, _ctx, info) => {
       return Actor.findMany().resolveInfo(info).execute()
@@ -291,7 +306,7 @@ const resolvers: IResolvers = {
   },
   Mutation: {
     createCustomer: async (_root, args, _ctx, info) => {
-      const id = await Customer.createOne(args.input).execute()
+      const id = await Customer.createOne(args.input).publish().execute()
       return Customer.findById(id).resolveInfo(info).execute()
     },
     createCustomers: async (_root, args, _ctx, info) => {
@@ -339,3 +354,4 @@ export const schema = makeSqlmancerSchema({
   typeDefs,
   resolvers,
 })
+export { pubsub };

src/client/createSqlmancerClient.ts

@@ -38,7 +38,8 @@ export type GenericSqlmancerClient = Knex & {
 
 export function createSqlmancerClient<T extends GenericSqlmancerClient = GenericSqlmancerClient>(
   glob: string,
-  knex: Knex
+  knex: Knex,
+  pubsub?: any
 ): T {
   const typeDefs = getTypeDefsFromGlob(glob)
 
@@ -51,7 +52,7 @@ export function createSqlmancerClient<T extends GenericSqlmancerClient = Generic
 
   return Object.assign(knex, {
     models: _.mapValues(models, (model: Model) => {
-      const options = { knex, dialect }
+      const options = { knex, dialect, pubsub }
       const { builders, readOnly } = model
       return {
         findById: (id: ID) => new builders.findById(options, id),

src/queryBuilder/createOne.ts

@@ -10,6 +10,16 @@ export class CreateOneBuilder<TCreateFields extends Record<string, any>> extends
     this._data = data
   }
 
+  /**
+   * Publishes and event notifying subscriber of Change
+   */
+  public publish(): this {
+    if (this._options.pubsub) {
+      this._options.pubsub.publish('CREATE_ONE');
+    }
+    return this;
+  }
+
   /**
    * Executes the query and returns a Promise that will resolve to the ID of the created row.
    */