fix: publisher not linked can publish messages with address correctly (#31)

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

Author: l4mby

src/consumer.ts

@@ -49,6 +49,8 @@ export class AmqpConsumer implements Consumer {
   ) {
     const id = generate_uuid()
     const address = createAddressFrom({ queue: { name: queueName } })
+    if (!address) throw new Error("Consumer must have an address")
+
     const receiverLink = await AmqpConsumer.openReceiver(connection, address, id)
     return new AmqpConsumer(id, connection, consumersList, receiverLink, params)
   }

src/message.ts

@@ -20,14 +20,19 @@ type MessageOptions = {
 
 export function createAmqpMessage(options: MessageOptions): RheaMessage {
   if (options.destination) {
-    return { message_id: generate_uuid(), body: options.body, to: createAddressFrom(options.destination) }
+    return {
+      message_id: generate_uuid(),
+      body: options.body,
+      to: createAddressFrom(options.destination),
+      durable: true,
+    }
   }
 
-  return { message_id: generate_uuid(), body: options.body }
+  return { message_id: generate_uuid(), body: options.body, durable: true }
 }
 
-export function createAddressFrom(options?: DestinationOptions): string {
-  if (!options) return ""
+export function createAddressFrom(options?: DestinationOptions): string | undefined {
+  if (!options) return undefined
   if ("queue" in options) return `/${AmqpEndpoints.Queues}/${options.queue.name}`
   if ("exchange" in options) {
     return options.exchange.routingKey

src/publisher.ts

@@ -5,15 +5,15 @@ import { inspect } from "util"
 import { createAddressFrom, DestinationOptions } from "./message.js"
 
 const getPublisherSenderLinkConfigurationFrom = (
-  address: string,
-  publisherId: string
+  publisherId: string,
+  address?: string
 ): SenderOptions | ReceiverOptions => ({
   snd_settle_mode: 0,
   rcv_settle_mode: 0,
   name: publisherId,
   target: { address, expiry_policy: "SESSION_END", durable: 0, dynamic: false },
   source: {
-    address,
+    address: address ?? "",
     expiry_policy: "LINK_DETACH",
     timeout: 0,
     dynamic: false,
@@ -49,17 +49,17 @@ export class AmqpPublisher implements Publisher {
   ): Promise<Publisher> {
     const address = createAddressFrom(options)
     const id = randomUUID()
-    const senderLink = await AmqpPublisher.openSender(connection, address, id)
+    const senderLink = await AmqpPublisher.openSender(connection, id, address)
     return new AmqpPublisher(connection, senderLink, id, publishersList)
   }
 
-  private static async openSender(connection: Connection, address: string, publisherId: string): Promise<Sender> {
+  private static async openSender(connection: Connection, publisherId: string, address?: string): Promise<Sender> {
     return openLink<Sender>(
       connection,
       SenderEvents.senderOpen,
       SenderEvents.senderError,
       connection.open_sender.bind(connection),
-      getPublisherSenderLinkConfigurationFrom(address, publisherId)
+      getPublisherSenderLinkConfigurationFrom(publisherId, address)
     )
   }
 

test/e2e/consumer.test.ts

@@ -60,7 +60,7 @@ describe("Consumer", () => {
     })
   })
 
-  test.skip("consumer can handle message on exchange, destination on message", async () => {
+  test("consumer can handle message on exchange, destination on message", async () => {
     const publisher = await connection.createPublisher()
     const expectedBody = "ciao"
     await publisher.publish(

test/e2e/publisher.test.ts

@@ -115,6 +115,6 @@ describe("Publisher", () => {
       })
     )
 
-    expect(publishResult.outcome).to.eql(OutcomeState.ACCEPTED)
+    expect(publishResult.outcome).to.eql(OutcomeState.RELEASED)
   })
 })

test/support/rhea_utils.ts

@@ -47,13 +47,13 @@ const MANAGEMENT_NODE_CONFIGURATION: SenderOptions | ReceiverOptions = {
   properties: { paired: true },
 }
 
-const getPublisherNodeConfigurationFrom = (address: string): SenderOptions => ({
+const getPublisherNodeConfigurationFrom = (address?: string): SenderOptions => ({
   snd_settle_mode: 0,
   rcv_settle_mode: 0,
   name: "publisher-sender-link",
   target: { address, expiry_policy: "SESSION_END", durable: 0, dynamic: false },
   source: {
-    address,
+    address: address ?? "",
     expiry_policy: "LINK_DETACH",
     timeout: 0,
     dynamic: false,
@@ -116,7 +116,7 @@ async function openLink<T extends Sender | Receiver>(
   })
 }
 
-export async function openPublisherSender(connection: Connection, address: string) {
+export async function openPublisherSender(connection: Connection, address?: string) {
   return openLink<Sender>(
     connection,
     SenderEvents.senderOpen,

test/unit/rhea/publisher.test.ts

@@ -123,4 +123,47 @@ describe("Creating a publisher through rhea", () => {
       })
     })
   })
+
+  describe("anonymous", () => {
+    test("create a publisher", async () => {
+      const publisher = await openPublisherSender(connection)
+
+      expect(publisher).to.not.eql(null)
+    })
+
+    test("send a message through a publisher", async () => {
+      const publisher = await openPublisherSender(connection)
+      let test = false
+      publisher.on(SenderEvents.accepted, () => {
+        console.log("accepted")
+        test = true
+      })
+      publisher.on(SenderEvents.sendable, () => {
+        console.log("sendable")
+        test = true
+      })
+      publisher.on(SenderEvents.modified, () => {
+        console.log("modified")
+        test = true
+      })
+      publisher.on(SenderEvents.rejected, () => {
+        console.log("rejected")
+        test = true
+      })
+      publisher.on(SenderEvents.released, () => {
+        console.log("released")
+        test = true
+      })
+      publisher.on(SenderEvents.settled, () => {
+        console.log("settled")
+        test = true
+      })
+
+      publisher.send({ message_id: randomUUID(), body: "Hello world!", to: `/queues/${testQueueName}` })
+
+      await eventually(async () => {
+        expect(test).eql(true)
+      })
+    })
+  })
 })