feat: use callbacks instead of async generators (#30)

etc/nats-component.api.md

@@ -27,14 +27,9 @@ export function encodeJson<T = unknown>(d: T): Uint8Array;
 
 // @public
 export type Subscription = {
-    generator: AsyncIterable<NatsMsg>;
     unsubscribe: () => void;
 };
 
-// Warnings were encountered during analysis:
-//
-// src/types.ts:21:3 - (ae-forgotten-export) The symbol "NatsMsg" needs to be exported by the entry point index.d.ts
-
 // (No @packageDocumentation comment for this package)
 
 ```

src/index.ts

@@ -1,8 +1,7 @@
 import { IBaseComponent } from "@well-known-components/interfaces"
-import { connect, NatsConnection, JSONCodec } from "nats"
-// import * as nats from "nats"
+import { connect, Msg, NatsConnection, NatsError } from "nats"
 import mitt from "mitt"
-import { natsComponent, INatsComponent, NatsEvents, Subscription } from "./types"
+import { natsComponent, INatsComponent, NatsEvents, Subscription, SubscriptionCallback } from "./types"
 
 export { createLocalNatsComponent } from "./test-component"
 export { encodeJson, decodeJson } from "./codecs"
@@ -33,30 +32,33 @@ export async function createNatsComponent(
     natsConnection.publish(topic, message)
   }
 
-  function subscribe(topic: string): Subscription {
+  function subscribe(topic: string, cb: SubscriptionCallback): Subscription {
     if (!natsConnection) {
       throw new Error("NATS component was not started yet")
     }
 
-    const sub = natsConnection.subscribe(topic)
+    const sub = natsConnection.subscribe(topic, {
+      callback: (err: NatsError | null, msg: Msg) => {
+        cb(err, msg)
+      },
+    })
+
     sub.closed
       .then(() => {
         logger.debug(`Subscription closed for topic`, { topic })
       })
       .catch((err: any) => {
         logger.error(`Subscription closed with an error`, err)
       })
-    return {
-      unsubscribe: () => sub.unsubscribe(),
-      generator: sub,
-    }
+
+    return sub
   }
 
   let didStop = false
 
   async function printStatus(connection: NatsConnection) {
     for await (const s of connection.status()) {
-      logger.info(`Status change`, s as any);
+      logger.info(`Status change`, s as any)
     }
   }
 
@@ -65,20 +67,18 @@ export async function createNatsComponent(
       natsConnection = await connect({ servers: `${natsUrl}` })
       printStatus(natsConnection).catch(logger.error)
 
-      natsConnection
-        .closed()
-        .then((err) => {
-          if (!didStop) {
-            logger.error(`NATS connection lost`)
-            if (err) {
-              logger.error(err)
-            }
-            // TODO: gracefully quit, this is an unrecoverable state
-            process.exit(1)
+      natsConnection.closed().then((err) => {
+        if (!didStop) {
+          logger.error(`NATS connection lost`)
+          if (err) {
+            logger.error(err)
           }
-        })
+          // TODO: gracefully quit, this is an unrecoverable state
+          process.exit(1)
+        }
+      })
       events.emit("connected")
-      logger.info(`Connected`, { server: natsConnection.getServer() });
+      logger.info(`Connected`, { server: natsConnection.getServer() })
     } catch (error) {
       logger.error(`An error occurred trying to connect to the NATS server: ${natsUrl}`)
       throw error

src/test-component.ts

@@ -1,22 +1,17 @@
 import { IBaseComponent } from "@well-known-components/interfaces"
-import { pushableChannel } from "@well-known-components/pushable-channel"
 import mitt from "mitt"
-import { INatsComponent, Subscription, NatsEvents, NatsMsg } from "./types"
-
-type PushableChannel = {
-  push: (value: NatsMsg, resolve: (err?: any) => void) => void
-}
+import { INatsComponent, Subscription, NatsEvents, SubscriptionCallback } from "./types"
 
 /**
  * Create a local NATS component, for testing purposes
  * @public
  */
 export async function createLocalNatsComponent(): Promise<INatsComponent & IBaseComponent> {
-  const channels = new Map<string, PushableChannel>()
+  const callbacks = new Map<string, Set<SubscriptionCallback>>()
   const events = mitt<NatsEvents>()
 
   function publish(topic: string, data: Uint8Array): void {
-    channels.forEach((ch, pattern) => {
+    callbacks.forEach((cbs, pattern) => {
       const sPattern = pattern.split(".")
       const sTopic = topic.split(".")
 
@@ -30,18 +25,20 @@ export async function createLocalNatsComponent(): Promise<INatsComponent & IBase
         }
       }
 
-      ch.push({ subject: topic, data }, console.log)
+      for (const cb of cbs) {
+        cb(null, { subject: topic, data })
+      }
     })
   }
 
-  function subscribe(pattern: string): Subscription {
-    const channel = pushableChannel<NatsMsg>(function deferCloseChannel() {
-      channels.delete(pattern)
-    })
-    channels.set(pattern, channel)
+  function subscribe(pattern: string, cb: SubscriptionCallback): Subscription {
+    const cbs = callbacks.get(pattern) || new Set<SubscriptionCallback>()
+    cbs.add(cb)
+    callbacks.set(pattern, cbs)
     return {
-      unsubscribe: () => channel.close(),
-      generator: channel.iterable,
+      unsubscribe: () => {
+        cbs.delete(cb)
+      },
     }
   }
 

src/types.ts

@@ -18,17 +18,22 @@ export type NatsMsg = {
  * @public
  */
 export type Subscription = {
-  generator: AsyncIterable<NatsMsg>
   unsubscribe: () => void
 }
 
+/**
+ * Subscription Callback
+ * @public
+ */
+export type SubscriptionCallback = (error: Error | null, msg: NatsMsg) => void
+
 export type NatsEvents = {
   connected: void
 }
 
 export type INatsComponent = {
   publish(topic: string, message?: Uint8Array): void
-  subscribe(topic: string): Subscription
+  subscribe(topic: string, cb: SubscriptionCallback): Subscription
 
   start(): Promise<void>
   stop(): Promise<void>