PKG -- [sdk] Create SDK subscribe function

packages/sdk/src/account/account.js

@@ -3,7 +3,7 @@ import {atBlockId} from "../build/build-at-block-id.js"
 import {getAccount} from "../build/build-get-account.js"
 import {invariant} from "@onflow/util-invariant"
 import {decodeResponse as decode} from "../decode/decode.js"
-import {send} from "../send/send.js"
+import {send} from "../transport"
 
 /**
  * @typedef {import("@onflow/typedefs").Account} Account

packages/sdk/src/block/block.js

@@ -1,4 +1,4 @@
-import {send} from "../send/send.js"
+import {send} from "../transport/send/send"
 import {getBlock} from "../build/build-get-block"
 import {atBlockHeight} from "../build/build-at-block-height.js"
 import {atBlockId} from "../build/build-at-block-id.js"

packages/sdk/src/contract.test.js

@@ -2,7 +2,7 @@ import * as root from "./sdk"
 import * as decode from "./decode/decode.js"
 import * as encode from "./encode/encode"
 import * as interaction from "./interaction/interaction"
-import * as send from "./send/send.js"
+import * as send from "./transport"
 import * as template from "@onflow/util-template"
 
 const interfaceContract =

packages/sdk/src/node-version-info/node-version-info.ts

@@ -1,4 +1,4 @@
-import {send} from "../send/send.js"
+import {send} from "../transport/send/send"
 import {decodeResponse as decode} from "../decode/decode.js"
 import {getNodeVersionInfo} from "../build/build-get-node-version-info"
 import {NodeVersionInfo} from "@onflow/typedefs"

packages/sdk/src/sdk.ts

@@ -2,7 +2,7 @@ import * as logger from "@onflow/util-logger"
 // Base
 export {build} from "./build/build.js"
 export {resolve} from "./resolve/resolve.js"
-export {send} from "./send/send.js"
+export {send, subscribe, rawSubscribe} from "./transport"
 export {decode} from "./decode/sdk-decode.js"
 export {
   encodeTransactionPayload,

packages/sdk/src/transport/get-transport.test.ts

@@ -0,0 +1,127 @@
+import {SdkTransport} from "@onflow/typedefs"
+import {getTransport} from "./get-transport"
+import {httpTransport} from "@onflow/transport-http"
+import {config} from "@onflow/config"
+
+jest.mock("@onflow/transport-http", () => ({
+  httpTransport: {
+    send: jest.fn(),
+    subscribe: jest.fn(),
+  } as jest.Mocked<SdkTransport.Transport>,
+}))
+
+describe("getTransport", () => {
+  beforeEach(() => {
+    jest.resetAllMocks()
+  })
+
+  test("fallback to http transport", async () => {
+    const transport = await getTransport()
+    expect(transport).toBe(httpTransport)
+  })
+
+  test("override with custom transport", async () => {
+    const customTransport = {
+      send: jest.fn(),
+      subscribe: jest.fn(),
+    } as jest.Mocked<SdkTransport.Transport>
+
+    const transport = await getTransport({transport: customTransport})
+    expect(transport).toBe(customTransport)
+  })
+
+  test("override with custom send function", async () => {
+    const customSend = jest.fn()
+
+    const transport = await getTransport({send: customSend})
+    expect(transport).toEqual({
+      send: customSend,
+      subscribe: expect.any(Function),
+    })
+  })
+
+  test("override with both custom transport and send function", async () => {
+    await expect(
+      getTransport({
+        send: jest.fn(),
+        transport: {
+          send: jest.fn(),
+          subscribe: jest.fn(),
+        },
+      })
+    ).rejects.toThrow(
+      /Cannot provide both "transport" and legacy "send" options/
+    )
+  })
+
+  test("transport from global config - sdk.transport", async () => {
+    const customTransport = {
+      send: jest.fn(),
+      subscribe: jest.fn(),
+    } as jest.Mocked<SdkTransport.Transport>
+
+    const tranpsort = await config().overload(
+      {
+        "sdk.transport": customTransport,
+      },
+      async () => {
+        return await getTransport()
+      }
+    )
+
+    expect(tranpsort).toBe(customTransport)
+  })
+
+  test("send function from global config - sdk.transport", async () => {
+    const customSend = jest.fn()
+
+    const transport = await config().overload(
+      {
+        "sdk.transport": customSend,
+      },
+      async () => {
+        return await getTransport()
+      }
+    )
+    expect(transport).toEqual({
+      send: customSend,
+      subscribe: expect.any(Function),
+    })
+  })
+
+  test("send function from global config - sdk.send", async () => {
+    const customSend = jest.fn()
+
+    const transport = await config().overload(
+      {
+        "sdk.send": customSend,
+      },
+      async () => {
+        return await getTransport()
+      }
+    )
+
+    expect(transport).toEqual({
+      send: customSend,
+      subscribe: expect.any(Function),
+    })
+  })
+
+  test("custom transport has priority over global config", async () => {
+    const customTransport = {
+      send: jest.fn(),
+      subscribe: jest.fn(),
+    } as jest.Mocked<SdkTransport.Transport>
+
+    const transport = await config().overload(
+      {
+        "sdk.transport": httpTransport,
+      },
+      async () => {
+        return await getTransport({transport: customTransport})
+      }
+    )
+
+    expect(transport).toBe(customTransport)
+  })
+})

packages/sdk/src/transport/get-transport.ts

@@ -0,0 +1,54 @@
+import {config} from "@onflow/config"
+import {httpTransport as defaultTransport} from "@onflow/transport-http"
+import {SdkTransport} from "@onflow/typedefs"
+import {invariant} from "@onflow/util-invariant"
+
+/**
+ * Get the SDK transport object, either from the provided override or from the global config.
+ * @param overrides - Override default configuration with custom transport or send function.
+ * @returns The SDK transport object.
+ */
+export async function getTransport(
+  override: {
+    send?: SdkTransport.SendFn
+    transport?: SdkTransport.Transport
+  } = {}
+): Promise<SdkTransport.Transport> {
+  invariant(
+    override.send == null || override.transport == null,
+    `SDK Transport Error: Cannot provide both "transport" and legacy "send" options.`
+  )
+
+  const transportOrSend =
+    override.transport ||
+    override.send ||
+    (await config().first<SdkTransport.Transport | SdkTransport.SendFn>(
+      ["sdk.transport", "sdk.send"],
+      defaultTransport
+    ))
+
+  // Backwards compatibility with legacy send function
+  if (!isTransportObject(transportOrSend)) {
+    return {
+      send: transportOrSend,
+      subscribe: () => {
+        throw new Error(
+          "Subscribe not supported with legacy send function transport, please provide a transport object."
+        )
+      },
+    }
+  }
+
+  return transportOrSend
+}
+
+function isTransportObject(
+  transport: any
+): transport is SdkTransport.Transport {
+  return (
+    transport.send !== undefined &&
+    transport.subscribe !== undefined &&
+    typeof transport.send === "function" &&
+    typeof transport.subscribe === "function"
+  )
+}

packages/sdk/src/transport/index.ts

@@ -0,0 +1,3 @@
+export {send} from "./send/send"
+export {subscribe} from "./subscribe/subscribe"
+export {rawSubscribe} from "./subscribe/raw-subscribe"

packages/sdk/src/transport/send/send.ts

@@ -0,0 +1,41 @@
+import {Buffer} from "@onflow/rlp"
+import {initInteraction, pipe} from "../../interaction/interaction"
+import * as ixModule from "../../interaction/interaction"
+import {invariant} from "../../build/build-invariant"
+import {response} from "../../response/response"
+import {config} from "@onflow/config"
+import {resolve as defaultResolve} from "../../resolve/resolve"
+import {getTransport} from "../get-transport"
+
+/**
+ * @description - Sends arbitrary scripts, transactions, and requests to Flow
+ * @param args - An array of functions that take interaction and return interaction
+ * @param opts - Optional parameters
+ * @returns - A promise that resolves to a response
+ */
+export const send = async (
+  args: Function | Function[] = [],
+  opts: any = {}
+): Promise<any> => {
+  const transport = await getTransport(opts)
+  const sendFn = transport.send.bind(transport)
+
+  invariant(
+    sendFn,
+    `Required value for sdk.transport is not defined in config. See: ${"https://github.com/onflow/fcl-js/blob/master/packages/sdk/CHANGELOG.md#0057-alpha1----2022-01-21"}`
+  )
+
+  const resolveFn = await config.first(
+    ["sdk.resolve"],
+    opts.resolve || defaultResolve
+  )
+
+  opts.node = opts.node || (await config().get("accessNode.api"))
+
+  if (Array.isArray(args)) args = pipe(initInteraction(), args as any) as any
+  return sendFn(
+    await resolveFn(args),
+    {config, response, ix: ixModule, Buffer} as any,
+    opts
+  )
+}

packages/sdk/src/transport/subscribe/raw-subscribe.test.ts

@@ -0,0 +1,47 @@
+import {config} from "@onflow/config"
+import {rawSubscribe} from "./raw-subscribe"
+import {SdkTransport} from "@onflow/typedefs"
+import {getTransport} from "../get-transport"
+
+jest.mock("../get-transport")
+
+describe("subscribe", () => {
+  let mockTransport: jest.Mocked<SdkTransport.Transport>
+  let mockSub: jest.Mocked<SdkTransport.Subscription> = {
+    unsubscribe: jest.fn(),
+  }
+
+  beforeEach(() => {
+    jest.resetAllMocks()
+
+    mockTransport = {
+      subscribe: jest.fn().mockReturnValue(mockSub),
+      send: jest.fn(),
+    }
+    jest.mocked(getTransport).mockResolvedValue(mockTransport)
+  })
+
+  test("subscribes to a topic and returns subscription from transport", async () => {
+    const topic = "topic" as SdkTransport.SubscriptionTopic
+    const args = {foo: "bar"} as SdkTransport.SubscriptionArguments<any>
+    const onData = jest.fn()
+    const onError = jest.fn()
+
+    const sub = await config().overload(
+      {
+        "accessNode.api": "http://localhost:8080",
+      },
+      async () => {
+        return await rawSubscribe({topic, args, onData, onError})
+      }
+    )
+
+    expect(mockTransport.subscribe).toHaveBeenCalledTimes(1)
+    expect(mockTransport.subscribe).toHaveBeenCalledWith(
+      {topic, args, onData: onData, onError},
+      {node: "http://localhost:8080"}
+    )
+
+    expect(sub).toBe(mockSub)
+  })
+})

packages/sdk/src/transport/subscribe/raw-subscribe.ts

@@ -0,0 +1,41 @@
+import {config} from "@onflow/config"
+import {SdkTransport} from "@onflow/typedefs"
+import {getTransport} from "../get-transport"
+import {invariant} from "@onflow/util-invariant"
+import {SubscribeParams} from "./types"
+
+/**
+ * Subscribe to a topic without decoding the data.
+ * @param params - The parameters for the subscription.
+ * @param opts - Additional options for the subscription.
+ * @returns A promise that resolves once the subscription is active.
+ */
+export async function rawSubscribe<T extends SdkTransport.SubscriptionTopic>(
+  {topic, args, onData, onError}: SubscribeParams<T>,
+  opts: {
+    node?: string
+    transport?: SdkTransport.Transport
+  } = {}
+) {
+  const transport = await getTransport(opts)
+  const node = opts?.node || (await config().get("accessNode.api"))
+
+  invariant(
+    !!node,
+    `SDK Send Error: Either opts.node or "accessNode.api" in config must be defined.`
+  )
+
+  // Subscribe using the resolved transport
+  return transport.subscribe(
+    {
+      topic,
+      args,
+      onData,
+      onError,
+    },
+    {
+      node,
+      ...opts,
+    }
+  )
+}