Add SDK SubscriptionManager

Author: jribbink

packages/sdk/src/subscribe/subscribe.test.ts

@@ -0,0 +1,48 @@
+import { config } from "@onflow/config"
+import {subscribe} from "./subscribe"
+import {SubscriptionManager} from "./subscription-manager"
+
+const MockSubscriptionManager = <jest.Mock<SubscriptionManager>>SubscriptionManager;
+
+jest.mock("./subscription-manager")
+
+describe("subscribe", () => {
+  let mockSubscriptionTransport: jest.Mock
+
+  beforeEach(() => {
+    jest.resetAllMocks()
+    mockSubscriptionTransport = jest.fn().mockImplementation(() => ({
+        subscribe: jest.fn().mockReturnValue({
+            unsubscribe: jest.fn()
+        })
+    }))
+  })
+
+  test("subscribes to a topic and returns a subscription", async () => {
+    const topic = "topic"
+    const args = {foo: "bar"}
+    const onData = jest.fn()
+    const onError = jest.fn()
+
+    const sub = await config().overload({
+        "sdk.subscriptionTransport": mockSubscriptionTransport,
+        "accessNode.api": "http://localhost:8080"
+    }, async () => {
+        await subscribe({topic, args, onData, onError})
+    })
+
+    expect(MockSubscriptionManager).toHaveBeenCalledTimes(1)
+    expect(mockSubscriptionTransport).toHaveBeenCalledTimes(1)
+    expect(mockSubscriptionTransport).toHaveBeenCalledWith({node: "http://localhost:8080"})
+
+    const manager = <jest.Mocked<SubscriptionManager>>MockSubscriptionManager.mock.instances[0]
+    expect(manager.subscribe).toHaveBeenCalledTimes(1)
+    expect(manager.subscribe).toHaveBeenCalledWith({
+        topic,
+        args,
+        onData,
+        onError
+    })
+    expect(sub).toBe(manager.subscribe.mock.results[0].value)
+  })
+})

packages/sdk/src/subscribe/subscribe.ts

@@ -0,0 +1,44 @@
+import {config} from "@onflow/config"
+import {
+  SubscriptionManager,
+  SubscriptionTransportCstr,
+} from "./subscription-manager"
+
+const managerMap: Map<
+  SubscriptionTransportCstr,
+  {
+    [node: string]: SubscriptionManager
+  }
+> = new Map()
+
+// TODO: OPTS FUNCTION
+export async function subscribe(
+  {
+    topic,
+    args,
+    onData,
+    onError,
+  }: {
+    topic: string
+    args: any
+    onData: (data: any) => void
+    onError: (error: Error) => void
+  }
+) {
+  // Find the SubscriptionManager instance corresponding to the current transport and node
+  const subscriptionTransport = (await config().get<SubscriptionTransportCstr>("sdk.subscriptionTransport"))
+  const node = await config().get<string>("accessNode.api")
+
+  const managersForConstructor = managerMap.get(subscriptionTransport)
+  const subscriptionManager =
+    managersForConstructor?.[node] ??
+    new SubscriptionManager(new subscriptionTransport({node}))
+  managerMap.set(subscriptionTransport, {
+    ...managersForConstructor,
+    [node]: subscriptionManager,
+  })
+
+  // TODO: handle onError
+  // Subscribe using the resolved transport
+  return subscriptionManager.subscribe({topic, args, onData, onError})
+}

packages/sdk/src/subscribe/subscription-manager.test.ts

@@ -0,0 +1,55 @@
+import { SubscriptionManager, SubscriptionTransport } from "./subscription-manager"
+
+describe("SubscriptionManager", () => {
+    let mockSubscriptionTransport: jest.Mocked<SubscriptionTransport>
+    let manager: SubscriptionManager
+
+    beforeEach(() => {
+        jest.resetAllMocks()
+        mockSubscriptionTransport = {
+            subscribe: jest.fn().mockReturnValue({
+                unsubscribe: jest.fn()
+            })
+        }
+        manager = new SubscriptionManager(mockSubscriptionTransport)
+    })
+
+    test("subscribes to a topic and returns a subscription", async () => {
+        const topic = "topic"
+        const args = {foo: "bar"}
+        const onData = jest.fn()
+        const onError = jest.fn()
+
+        const sub = manager.subscribe({topic, args, onData, onError})
+
+        expect(mockSubscriptionTransport.subscribe).toHaveBeenCalledTimes(1)
+        expect(mockSubscriptionTransport.subscribe).toHaveBeenCalledWith({
+            topic,
+            args,
+            onData: expect.any(Function),
+            onError
+        })
+        expect(sub).toStrictEqual({
+            unsubscribe: expect.any(Function)
+        })
+    })
+
+    test("unsubscribe from a subscription", async () => {
+        const topic = "topic"
+        const args = {foo: "bar"}
+        const onData = jest.fn()
+        const onError = jest.fn()
+
+        const sub = manager.subscribe({topic, args, onData, onError})
+        sub.unsubscribe()
+
+        expect(mockSubscriptionTransport.subscribe).toHaveBeenCalledTimes(1)
+        expect(mockSubscriptionTransport.subscribe).toHaveBeenCalledWith({
+            topic,
+            args,
+            onData: expect.any(Function),
+            onError
+        })
+        expect(mockSubscriptionTransport.subscribe.mock.results[0].value.unsubscribe).toHaveBeenCalledTimes(1)
+    })
+})

packages/sdk/src/subscribe/subscription-manager.ts

@@ -0,0 +1,57 @@
+export type Subscription = {
+  unsubscribe: () => void
+}
+
+export type SubscriptionTransportCstr = new (
+  config: SubscriptionTransportConfig
+) => SubscriptionTransport
+
+export type SubscriptionTransport = {
+  subscribe: ({
+    topic,
+    args,
+    onData,
+    onError,
+  }: {
+    topic: string
+    args: any
+    onData: (data: any) => void
+    onError: (error: Error) => void
+  }) => Subscription
+}
+
+export type SubscriptionTransportConfig = {
+  node: string
+}
+
+export class SubscriptionManager {
+  constructor(private readonly transport: SubscriptionTransport) {}
+
+  subscribe({
+    topic,
+    args,
+    onData,
+    onError,
+  }: {
+    topic: string
+    args: any
+    onData: (data: any) => void
+    onError: (error: Error) => void
+  }) {
+    const sub = this.transport.subscribe({
+      topic,
+      args,
+      onData: data => {
+        onData(this.decode(topic, data))
+      },
+      onError,
+    })
+
+    return sub
+  }
+
+  private decode(topic: string, data: any): any {
+    // TODO
+    return data
+  }
+}