[IS-6/feat]: add publisher support (#26)

* feat: add publisher support

* fix: sender error test and remove ts-node

* fix: skipped test

---------

Co-authored-by: magne <[email protected]>

Author: l4mby

package-lock.json

@@ -55,7 +55,6 @@
     "globals": "^16.2.0",
     "got": "^14.4.7",
     "prettier": "3.5.3",
-    "ts-node": "^10.9.2",
     "tsup": "^8.5.0",
     "typescript": "^5.8.3",
     "typescript-eslint": "^8.33.1",

package.json

@@ -1,14 +1,20 @@
 import { ConnectionEvents, create_container, Connection as RheaConnection } from "rhea"
 import { AmqpManagement, Management } from "./management.js"
 import { EnvironmentParams } from "./environment.js"
+import { AmqpPublisher, Publisher } from "./publisher.js"
+import { DestinationOptions } from "./message.js"
 
 export interface Connection {
   close(): Promise<boolean>
   isOpen(): boolean
   management(): Management
+  createPublisher(options?: DestinationOptions): Promise<Publisher>
+  get publishers(): Map<string, Publisher>
 }
 
 export class AmqpConnection implements Connection {
+  private _publishers: Map<string, Publisher> = new Map<string, Publisher>()
+
   static async create(params: EnvironmentParams) {
     const connection = await AmqpConnection.open(params)
     const topologyManagement = await AmqpManagement.create(connection)
@@ -43,6 +49,7 @@ export class AmqpConnection implements Connection {
         return rej(new Error("Connection error: " + context.connection.error))
       })
 
+      this._publishers.forEach((p) => p.close())
       this.connection.close()
     })
   }
@@ -51,6 +58,16 @@ export class AmqpConnection implements Connection {
     return this.topologyManagement
   }
 
+  async createPublisher(options?: DestinationOptions): Promise<Publisher> {
+    const publisher = await AmqpPublisher.createFrom(this.connection, this._publishers, options)
+    this._publishers.set(publisher.id, publisher)
+    return publisher
+  }
+
+  public get publishers(): Map<string, Publisher> {
+    return this._publishers
+  }
+
   public isOpen(): boolean {
     return this.connection ? this.connection.is_open() : false
   }

src/connection.ts

@@ -15,7 +15,7 @@ export enum AmqpEndpoints {
 
 export const ME = "$me"
 
-export class MessageBuilder {
+export class LinkMessageBuilder {
   private messageId: string = generate_uuid()
   private to: string = ""
   private replyTo: string = ME