chore: revert onMessage to iterators

src/index.ts

@@ -1,7 +1,7 @@
 import { IBaseComponent } from "@well-known-components/interfaces"
 import { connect, NatsConnection } from "nats"
 import mitt from "mitt"
-import { natsComponent, INatsComponent, NatsEvents, Subscription, NatsMsg } from "./types"
+import { natsComponent, INatsComponent, NatsEvents, Subscription } from "./types"
 
 /**
  * Create a NATS component (https://nats.io/)
@@ -25,14 +25,8 @@ export async function createNatsComponent(
     natsConnection.publish(topic, message)
   }
 
-  function subscribe(topic: string, onMessage: (_: NatsMsg) => Promise<void>): Subscription {
+  function subscribe(topic: string): Subscription {
     const sub = natsConnection.subscribe(topic)
-    ;(async () => {
-      for await (const message of sub) {
-        await onMessage({ data: message.data, subject: message.subject })
-      }
-    })().catch((err: any) => logger.error(`error processing subscription message; ${err.message}`))
-
     sub.closed
       .then(() => {
         logger.info(`subscription closed for ${topic}`)
@@ -42,6 +36,7 @@ export async function createNatsComponent(
       })
     return {
       unsubscribe: () => sub.unsubscribe(),
+      generator: sub,
     }
   }
 

src/test-component.ts

@@ -1,6 +1,6 @@
 import { IBaseComponent, IConfigComponent } from "@well-known-components/interfaces"
 const { connect } = require("mock-nats-client")
-import { natsComponent, INatsComponent, Subscription, NatsEvents, NatsMsg } from "./types"
+import { natsComponent, INatsComponent, Subscription, NatsEvents } from "./types"
 import mitt from "mitt"
 
 export async function createLocalNatsComponent(
@@ -13,14 +13,12 @@ export async function createLocalNatsComponent(
     message ? client.publish(topic, message) : client.publish(topic, [])
   }
 
-  function subscribe(topic: string, onMessage: (_: NatsMsg) => Promise<void>): Subscription {
-    const sid = client.subscribe(topic, (delivery: any, _replyTo: any, subject: string) => {
-      onMessage({ data: delivery, subject })
-    })
-    const unsubscribe = () => {
-      client.unsubscribe(sid)
+  function subscribe(topic: string): Subscription {
+    const sub = client.subscribe(topic)
+    return {
+      unsubscribe: () => client.unsubscribe(sub),
+      generator: sub,
     }
-    return { unsubscribe }
   }
 
   async function start() {

src/types.ts

@@ -14,6 +14,7 @@ export type NatsMsg = {
 }
 
 export type Subscription = {
+  generator: AsyncIterable<NatsMsg>
   unsubscribe: () => void
 }
 
@@ -23,7 +24,7 @@ export type NatsEvents = {
 
 export type INatsComponent = {
   publish(topic: string, message?: Uint8Array): void
-  subscribe(topic: string, onMessage: (_: NatsMsg) => Promise<void>): Subscription
+  subscribe(topic: string): Subscription
 
   start(): Promise<void>
   stop(): Promise<void>