feat: implement NATS component

package.json

@@ -23,11 +23,19 @@
   "devDependencies": {
     "@microsoft/api-extractor": "^7.17.0",
     "@types/jest": "^28.1.0",
+    "@well-known-components/env-config-provider": "^1.1.1",
+    "@well-known-components/test-helpers": "^1.3.0",
     "jest": "^28.1.0",
     "ts-jest": "^28.0.4",
     "typescript": "^4.7.3"
   },
   "files": [
     "dist"
-  ]
+  ],
+  "dependencies": {
+    "@well-known-components/interfaces": "^1.1.1",
+    "mitt": "^3.0.0",
+    "mock-nats-client": "^0.2.0",
+    "nats": "^2.7.1"
+  }
 }

src/index.ts

@@ -1,7 +1,64 @@
-/**
- * A function that does something
- * @public
- */
-export function example(){
-  return true
-}
+import { IBaseComponent } from "@well-known-components/interfaces"
+import { connect, NatsConnection } from "nats"
+import mitt from "mitt"
+import { natsComponent, INatsComponent, NatsEvents, Subscription } from "./types"
+
+export async function createNatsComponent(
+  components: natsComponent.NeededComponents
+): Promise<INatsComponent & IBaseComponent> {
+  const { config, logs } = components
+  const logger = logs.getLogger("NATS")
+
+  // config
+  const natsUrl = (await config.getString("NATS_URL")) || "localhost:4222"
+  const natsConfig = { servers: `${natsUrl}` }
+  let natsConnection: NatsConnection
+
+  const events = mitt<NatsEvents>()
+
+  function publish(topic: string, message?: Uint8Array): void {
+    natsConnection.publish(topic, message)
+  }
+
+  function subscribe(topic: string): Subscription {
+    const sub = natsConnection.subscribe(topic)
+    sub.closed
+      .then(() => {
+        logger.info(`subscription closed for ${topic}`)
+      })
+      .catch((err) => {
+        logger.error(`subscription closed with an error ${err.message}`)
+      })
+    return {
+      unsubscribe: () => sub.unsubscribe(),
+      generator: sub,
+    }
+  }
+
+  async function start() {
+    try {
+      natsConnection = await connect(natsConfig)
+      events.emit("connected")
+      logger.info(`Connected to NATS: ${natsUrl}`)
+    } catch (error) {
+      logger.error(`An error occurred trying to connect to the NATS server: ${natsUrl}`)
+      throw error
+    }
+  }
+
+  async function stop() {
+    try {
+      await natsConnection.close()
+    } catch (error) {
+      logger.error(`An error occurred trying to close the connection to the NATS server: ${natsUrl}`)
+    }
+  }
+
+  return {
+    publish,
+    subscribe,
+    start,
+    stop,
+    events,
+  }
+}

src/test-component.ts

@@ -0,0 +1,37 @@
+import { IBaseComponent, IConfigComponent } from "@well-known-components/interfaces"
+const { connect } = require("mock-nats-client")
+import { natsComponent, INatsComponent, Subscription, NatsEvents } from "./types"
+import mitt from "mitt"
+
+export async function createLocalNatsComponent(
+  components: natsComponent.NeededComponents
+): Promise<INatsComponent & IBaseComponent> {
+  const events = mitt<NatsEvents>()
+  const client = connect({ preserveBuffers: true })
+
+  function publish(topic: string, message: any): void {
+    message ? client.publish(topic, message) : client.publish(topic, [])
+  }
+
+  function subscribe(topic: string): Subscription {
+    const sub = client.subscribe(topic)
+    return {
+      unsubscribe: () => client.unsubscribe(sub),
+      generator: sub,
+    }
+  }
+
+  async function start() {
+    events.emit("connected")
+  }
+
+  async function stop() {}
+
+  return {
+    publish,
+    subscribe,
+    start,
+    stop,
+    events,
+  }
+}

src/types.ts

@@ -0,0 +1,33 @@
+import { IConfigComponent, ILoggerComponent } from "@well-known-components/interfaces"
+import { Emitter } from "mitt"
+
+export namespace natsComponent {
+  export type NeededComponents = {
+    logs: ILoggerComponent
+    config: IConfigComponent
+  }
+}
+
+export type NatsMsg = {
+  subject: string
+  data: Uint8Array
+}
+
+export type Subscription = {
+  generator: AsyncIterable<NatsMsg>
+  unsubscribe: () => void
+}
+
+export type NatsEvents = {
+  connected: void
+}
+
+export type INatsComponent = {
+  publish(topic: string, message?: Uint8Array): void
+  subscribe(topic: string): Subscription
+
+  start(): Promise<void>
+  stop(): Promise<void>
+
+  events: Emitter<NatsEvents>
+}

test/components.spec.ts

@@ -0,0 +1,8 @@
+import { test } from "./components"
+
+test("smoke test", function ({ components }) {
+  it("smoke test", async () => {
+    const { nats } = components
+    expect(nats).toBeTruthy()
+  })
+})

test/components.ts

@@ -0,0 +1,49 @@
+// This file is the "test-environment" analogous for src/components.ts
+// Here we define the test components to be used in the testing environment
+
+import { ILoggerComponent } from "@well-known-components/interfaces"
+import { createRunner } from "@well-known-components/test-helpers"
+import { createConfigComponent } from "@well-known-components/env-config-provider"
+import { INatsComponent, natsComponent } from "../src/types"
+import { createLocalNatsComponent } from "../src/test-component"
+
+export type TestComponents = natsComponent.NeededComponents & { nats: INatsComponent }
+
+export const logger = {
+  log: jest.fn(),
+  debug: jest.fn(),
+  error: jest.fn(),
+  warn: jest.fn(),
+  info: jest.fn(),
+}
+function createTestConsoleLogComponent(): ILoggerComponent {
+  return {
+    getLogger: () => logger,
+  }
+}
+
+/**
+ * Behaves like Jest "describe" function, used to describe a test for a
+ * use case, it creates a whole new program and components to run an
+ * isolated test.
+ *
+ * State is persistent within the steps of the test.
+ */
+export const test = createRunner<TestComponents>({
+  async main({ startComponents }) {
+    await startComponents()
+  },
+  async initComponents(): Promise<TestComponents> {
+    const config = createConfigComponent(process.env)
+
+    const logs = createTestConsoleLogComponent()
+
+    const nats = await createLocalNatsComponent({ config, logs })
+
+    return {
+      config,
+      logs,
+      nats,
+    }
+  },
+})