search

onebeyond / rascal

A config driven wrapper for amqp.node supporting multi-host connections, automatic error recovery, redelivery flood protection, transparent encryption / decryption and channel pooling.
MIT License
451 stars 69 forks source link

Cannot override queue in publish() or subscribe() #153

Closed stevenhair closed 3 years ago

stevenhair commented 3 years ago

When a queue name is passed in the overrides to a publish() or a subscribe() call, it seems to be ignored and the original queue defined in the broker configuration will be used. This is particularly problematic for dynamically-generated response queues.

Here is a reproduction: https://github.com/stevenhair/ubiquitous-adventure

cressie176 commented 3 years ago

Hi @stevenhair,

Thanks for reporting and the excellent reproduction. Rascal was designed this way to ensure that the queues and exchanges exist when the broker starts, and that they can be optionally torn down on shutdown, but I agree it makes things problematic for dynamically generated response queues. I suspect making this change may be difficult, but I'll have a think and see if it's something Rascal can support. In the meantime your best option is to use broker.connect.

stevenhair commented 3 years ago

Thanks for the quick response. For context, I'm using a request-reply pattern, which was implemented using amqplib using a temporary queue for the response. I recently needed the ability to fail over to another RabbitMQ instance which is not supported by amqplib, so I decided to give Rascal a try. Of course, I tried to translate the original pattern, but since Rascal doesn't support this, I changed the pattern a bit.

Now I use a single response queue that lives for the life of the application. I subscribe to that and just push all of the messages to an rxjs Subject. After sending a request, the caller then subscribes to the response observable and filters out responses that don't have the same correlationId.

If you don't feel that this functionality is a fit for Rascal, feel free to close this ticket. I'll move this issue over to DefinitelyTyped, since the types imply that this functionality exists.

cressie176 commented 3 years ago

Hi @stevenhair,

Thanks for your response and the extra context. I took a walk through the code, and I think the feature will be difficult to implement, mainly because when I started Rascal I wanted it to be config driven. This led me to put quite a lot of logic in the configuration, initialisation and shutdown phases which would need to be either duplicated or extracted so they could be called dynamically. I'm not against the change, but not sure I'll get time to make it for a little while.

I do have one other idea...

Rascal has an undocumented feature where if you specify replyTo: true in the queue configuration, it appends a uuid to the queue name, making it unique per application. You may also need to configure the queue as exclusive: true, durable:false and autoDelete: true as well.

This would mean each instance of the application could be subscribe to a dedicated, temporary reply queue. To publish messages to the reply queue you would need to publish a message to the default exchange, using the unique queue name as the routing key. The queue name would have to be specified in the original message (probably via a header). One catch though - I'm not sure how easy it is to get the queue name, but I don't imagine it would be hard to expose this on the publication.

Would the above fit your needs?

stevenhair commented 3 years ago

I think that's basically what I ended up implementing - I just generate a random queue name before creating the broker and use that name in the configuration passed to the create() function. I'm not sure how much overhead queue creation added (my application is relatively low-volume), but this is probably a better solution than what I had before.