feat(javascript_client): implement ActionCableGraphqlWsClient

javascript_client/package.json

@@ -22,6 +22,7 @@
     "@types/zen-observable": "^0.8.2",
     "ably": "1.2.6",
     "graphql": "^15.0.0",
+    "graphql-ws": "^5.5.5",
     "jest": "^25.0.0",
     "nock": "^11.0.0",
     "pako": "^2.0.3",
@@ -41,7 +42,8 @@
   },
   "peerDependencies": {
     "@apollo/client": "^3.3.6",
-    "graphql": "^14.3.1 || ^15.0.0"
+    "graphql": "^14.3.1 || ^15.0.0",
+    "graphql-ws": "^5.5.5"
   },
   "prettier": {
     "semi": false,

javascript_client/src/subscriptions/ActionCableGraphqlWsClient.ts

@@ -1,33 +1,29 @@
 import ActionCable from 'actioncable'
+import assert from 'assert'
 import GraphqlWs from 'graphql-ws'
+import { ActionCableUtil } from '../utils/ActionCableUtil'
 import GraphqlWsClient from './GraphqlWsClient'
 
-/**
- * Create a Relay Modern-compatible subscription handler.
- *
- * @param {ActionCable.Consumer} cable - An ActionCable consumer from `.createConsumer`
- * @param {String} channelName - ActionCable Channel name. Defaults to "GraphqlChannel"
- * @param {OperationStoreClient} operations - A generated OperationStoreClient for graphql-pro's OperationStore
- * @return {Function}
- */
-interface ActionCableGraphqlWsClientOptions {
+export interface ActionCableGraphqlWsClientOptions {
+  /** An ActionCable consumer from `.createConsumer` */
   cable: ActionCable.Cable
+  /** ActionCable Channel name. Defaults to "GraphqlChannel" */
   channelName: string
+  /** A generated OperationStoreClient for graphql-pro's OperationStore */
   operations?: { getOperationId: Function}
   // connectionParams: ConnectionParams
 }
 
-const getChannelId = () => Math.round(Date.now() + Math.random() * 100000).toString(16) // TODO: extract
-
 interface ChannelNameWithParams extends ActionCable.ChannelNameWithParams {
   channel: string
   channelId: string
 }
 
 class ActionCableGraphqlWsClient extends GraphqlWsClient {
   cable: ActionCable.Cable
-  actionCableChannel: ActionCable.Channel
+  channel: ActionCable.Channel
   cleanup: () => void
+  sink?: GraphqlWs.Sink
   // operations // TODO:
 
   constructor(options: ActionCableGraphqlWsClientOptions) {
@@ -36,61 +32,63 @@ class ActionCableGraphqlWsClient extends GraphqlWsClient {
     this.cable = options.cable
     const channelNameWithParams: ChannelNameWithParams = {
       channel: options.channelName || 'GraphqlChannel',
-      channelId: getChannelId()
+      channelId: ActionCableUtil.getUniqueChannelId()
     }
 
-    this.actionCableChannel = this.cable.subscriptions.create(channelNameWithParams, {
-      connected: this._action_cable_connected,
-      disconnected: this._action_cable_disconnected,
-      received: this._action_cable_received
-    })
-    this.cleanup = () => { /* TODO */}
-  }
-
-  on(event: GraphqlWs.Event) {
-    switch (event) {
-      case 'connecting':
-        break
-      case 'opened':
-        break
-      case 'connected':
-        break
-      case 'ping':
-        break
-      case 'pong':
-        break
-      case 'message':
-        break
-      case 'closed':
-        break
-      case 'error':
-        break
+    this.channel = this.cable.subscriptions.create(channelNameWithParams, {
+      connected: this.action_cable_connected.bind(this),
+      disconnected: this.action_cable_disconnected.bind(this),
+      received: this.action_cable_received.bind(this)
+    })// TODO: support connectionParams like `ActionCableLink.ts` ?
+    this.cleanup = () => {
+      this.channel.unsubscribe()
     }
-    return () => {}
   }
 
+  // TODO: Should we do anything with `event` here ?
+  on(_event: GraphqlWs.Event) { return () => {} }
+
   subscribe(payload: GraphqlWs.SubscribePayload, sink: GraphqlWs.Sink) {
+    this.sink = sink
+    const {
+      operationName,
+      query,
+      variables,
+    } = payload
+
+    const channelParams = {
+      variables,
+      operationName,
+      query
+    }
+
+    this.channel.perform('execute', channelParams)
+    // TODO: Why another 'send' in `createActionCableHandler` ?
+    // this.channel.perform('send', channelParams)
+
     return this.cleanup
   }
 
   dispose() {
     return this.cleanup()
   }
 
-  _connect() {
+  private action_cable_connected() {}
 
-  }
-
-  _action_cable_connected() {
-    // TODO
-  }
+  private action_cable_disconnected() {}
 
-  _action_cable_disconnected() {
-    // TODO
-  }
+  private action_cable_received(payload: any) {
+    assert.ok(this.sink) // subscribe() should have been called first, right ?
+    const result = payload.result
 
-  _action_cable_received() {
-    // TODO
+    if (result?.errors) {
+      this.sink.error(result.errors)
+    } else if (result) {
+      this.sink.next(result)
+    }
+    if (!payload.more) {
+      this.sink.complete()
+    }
   }
 }
 

javascript_client/src/subscriptions/GraphqlWsClient.ts

@@ -1,16 +1,14 @@
 import GraphqlWs from 'graphql-ws'
 
 class GraphqlWsClient implements GraphqlWs.Client {
-  on(event: GraphqlWs.Event) {
-    return () => {
-      console.error('Subclass responsibility')
-    }
+  on(_event: GraphqlWs.Event) {
+    console.error('Subclass responsibility')
+    return () => {}
   }
 
-  subscribe(payload: GraphqlWs.SubscribePayload, sink: GraphqlWs.Sink) {
-    return () => {
-      console.error('Subclass responsibility')
-    }
+  subscribe(_payload: GraphqlWs.SubscribePayload, _sink: GraphqlWs.Sink) {
+    console.error('Subclass responsibility')
+    return () => {}
   }
 
   dispose() {

javascript_client/src/subscriptions/__tests__/ActionCableGraphqlWsClientTest.ts

@@ -0,0 +1,129 @@
+import { Cable  } from "actioncable"
+import { Sink, SubscribePayload } from "graphql-ws"
+import ActionCableGraphqlWsClient, { ActionCableGraphqlWsClientOptions } from "../ActionCableGraphqlWsClient"
+
+describe("ActionCableGraphqlWsClient", () => {
+  let log: any[]
+  let cable: any
+  let cableReceived: Function
+  let options: ActionCableGraphqlWsClientOptions
+  let query: string
+  let subscribePayload: SubscribePayload
+  let sink: Sink
+
+  beforeEach(() => {
+    log = []
+    cable = {
+      subscriptions: {
+        create: function(channelName: string | object, options: {connected: Function, received: Function}) {
+          let channel = channelName
+          let params = typeof channel === "object" ? channel : { channel }
+          let alreadyConnected = false
+          cableReceived = options.received
+          let subscription = Object.assign(
+            Object.create({
+              perform: function(actionName: string, options: object) {
+                log.push(["cable perform", { actionName: actionName, options: options }])
+              },
+              unsubscribe: function() {
+                log.push(["cable unsubscribe"])
+              }
+            }),
+            { params },
+            options
+          )
+
+          subscription.connected = subscription.connected.bind(subscription)
+          let received = subscription.received
+          subscription.received = function(data: any) {
+            if (!alreadyConnected) {
+              alreadyConnected = true
+              subscription.connected()
+            }
+            received(data)
+          }
+          subscription.__proto__.unsubscribe = subscription.__proto__.unsubscribe.bind(subscription)
+          return subscription
+        }
+      }
+    }
+    options = {
+      cable: (cable as unknown) as Cable,
+      channelName: 'GraphQLChannel',
+      operations: undefined
+    }
+
+    query = "subscription { foo { bar } }"
+
+    subscribePayload = {
+      operationName: 'myOperationName',
+      variables: { a: 1 },
+      query: query
+    }
+
+    sink = {
+      next(value) {
+        log.push(["sink next", value])
+      },
+      error(error) {
+        log.push(["sink error", error])
+      },
+      complete() {
+        log.push(["sink complete"])
+      }
+    }
+  })
+
+  it("delegates to the cable", () => {
+    const client = new ActionCableGraphqlWsClient(options)
+
+    client.subscribe(subscribePayload, sink)
+    cableReceived({ result: { data: null }, more: true })
+    cableReceived({ result: { data: "data 1" }, more: true })
+    cableReceived({ result: { data: "data 2" }, more: false })
+
+    expect(log).toEqual([
+      [
+        "cable perform",
+        {
+          actionName: "execute",
+          options: {
+            operationName: "myOperationName",
+            query: "subscription { foo { bar } }",
+            variables: { a: 1 }
+          }
+        }
+      ],
+      ["sink next", { data: null} ],
+      ["sink next", { data: "data 1"} ],
+      ["sink next", { data: "data 2"} ],
+      ["sink complete"],
+    ])
+  })
+
+  it("delegates a manual unsubscribe to the cable", () => {
+    const client = new ActionCableGraphqlWsClient(options)
+
+    client.subscribe(subscribePayload, sink)
+    cableReceived({ result: { data: null }, more: true })
+    cableReceived({ result: { data: "data 1" }, more: true })
+    client.dispose()
+
+    expect(log).toEqual([
+      [
+        "cable perform",
+        {
+          actionName: "execute",
+          options: {
+            operationName: "myOperationName",
+            query: "subscription { foo { bar } }",
+            variables: { a: 1 }
+          }
+        }
+      ],
+      ["sink next", { data: null }],
+      ["sink next", { data: "data 1" }],
+      ["cable unsubscribe"]
+    ])
+  })
+})