Allow us to add queue sub to existing consumer

[the1mills]

Is your feature request related to a problem? Please describe.

the problem is that dynamically adding more subscribers (more queues) to a Rabbit ch.Consume call is not possible. We have to call ch.Consume for every time we have more topics/queues to subscribe to. Whereas in the 2nd example, the pattern for Redis seems more performant and sustainable for real/large systems

Here is the minimum viable example:

package main

import (
  "fmt"
  amqp "github.com/rabbitmq/amqp091-go"
  "log"
)

func main() {
  conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
  defer conn.Close()

  ch, err := conn.Channel()
  defer ch.Close()

  // Function to dynamically declare and consume from a new queue
  consumeFromQueue := func(queueName string) {
    q, err := ch.QueueDeclare(
       // ....
    )

    msgs, err := ch.Consume(
      q.Name, // queue
        // ...
    )

    go func() {  
      for d := range msgs {   // storing extra memory here with this "permanent" goroutine
        log.Printf("Received a message: %s", d.Body)
      }
    }()
    
    fmt.Println("Subscribed to:", queueName)
  }

  // Subscribe to the initial queue
  consumeFromQueue("queue1")

  // Dynamically subscribe (in future) to more queues as needed
 go func(){
  consumeFromQueue("queue2")
  consumeFromQueue("queue3")
  }()

  // Block forever (for this example)
  forever := make(chan bool)
  <-forever
}

But the above ^^ seems suboptimal performance-wise - I am looking for something more like this Redis code:

package main

import (
  "context"
  "fmt"
  "github.com/go-redis/redis/v8"
  "time"
)

func main() {
  rdb := redis.NewClient(&redis.Options{
    Addr:     "localhost:6379", // Redis server address
  })

  ctx := context.Background()

  pubsub := rdb.Subscribe(ctx)
  defer pubsub.Close()

  // Function to dynamically add channels to the subscription
  addChannel := func(channel string) {
    err := pubsub.Subscribe(ctx, channel)
    if err != nil {
      fmt.Println("Subscribe error:", err)
      return
    }
    fmt.Println("Subscribed to:", channel)
  }

  // Add initial channel
  addChannel("channel1")

  // Simulate dynamically adding more channels
  go func() {
    time.Sleep(5 * time.Second)
    addChannel("channel2")
    time.Sleep(5 * time.Second)
    addChannel("channel3")
  }()

  // read all messages for all channels/topics
  for msg := range pubsub.Channel() {
    fmt.Println(msg.Channel, msg.Payload)
  }
}

with Redis, we can add new channels/topics to an existing subscriber (saving memory and cpu, in theory).
But with RMQ, we need to create a new goroutine and for-loop to read messages to a new queue.

We should be able to add the queue to existing subscriber (I suppose, to share a channel).

Describe the solution you'd like

The redis code (above) is what I want, that's a good system

Describe alternatives you've considered

Not sure, open to alternatives

Additional context

No response

[lukebakken]

But the above ^^ seems suboptimal performance-wise

Is it, though? Have you benchmarked it to demonstrate that the current behavior does not meet your needs?

[the1mills]

But the above ^^ seems suboptimal performance-wise

Is it, though? Have you benchmarked it to demonstrate that the current behavior does not meet your needs?

It's more of a memory issue, if I have to call that 10,000 times for 10K queues, it will blow up memory unnecessarily with closures - I could provide hard data on it if you want - but since I have nothing to compare it to - I cant yet. You logicing with me still? So the API should be flexible to accommodate it - so we could actually do a real comparison. Lets solve this together, my fellow knights.

[lukebakken]

All of the AMQP clients work in pretty much the same manner, and without a very compelling reason to operate differently here, I doubt this API suggestion would be implemented.

You are always free to submit a PR for review, of course.

[the1mills]

yes I understand that such an API change might be incompatible with the way RabbitMQ operates, it feels antiquated to me. I honestly hope it gets major maintenance - I don't know erlang yet, so I doubt I could help anytime soon, but maybe one I day I will get there. Again, the Redis design makes sense, it's flexible in an important way that RMQ API is not.

[lukebakken]

yes I understand that such an API change might be incompatible with the way RabbitMQ operates

This idea should be compatible with RabbitMQ, it's just very different from other AMQP client API design (which is what I said). If enough people were to request such a change, or PAY for it to be implemented, we would consider it, especially if evidence could be provided that the current design causes issues in someone's production application.

I honestly hope it gets major maintenance

Yes, RabbitMQ and it's client libraries are professionally maintained by Broadcom / VMware. We can't just jump on everyone's suggestions, especially when they are not backed by hard evidence as to the improvement, or with a clear demonstration of the current problems.

[michaelklishin]

RabbitMQ is 17 years old and supports N protocols which are not perfect but can accomplish a lot if you understand how all their features are supposed to be used in combination.

@the1mills throwing around generalizations and accusations like "perplexing", "antiquated", and so on does not help your case at all. You have been previously told that the way you communicate with RabbitMQ maintainers is disrespectful, counterproductive and you come around as the most entitled person on GitHub. Who has not contributed much to open source projects on top of that but boy do you want others to do stuff for your for free!

Unless you are 16, this attitude towards folks whose work you use for free, including updates and a certain degree of support, is not something Team RabbitMQ is going to just accept or ignore.

Your second account will be banned just like your first account was.

[lukebakken]

I also just remembered that the AMQP protocol allows a single channel to consume from multiple queues. That seems like it fits your requirement. Give that a try.

[michaelklishin]

The protocol and client library design assumes a few things:

• That channels are long lived
• That you can add or remove consumers (and reuse a consumer implementation, like a Java object or a function in some other clients) between them
• That your topology is not completely chaotic and does not change multiple times a second

Specifically topic exchanges can support a wide range of complex patterns, and in many cases help reduce the topology churn.

Instead of making generalized claims that something is "perplexing", "antiquated", and other completely unbacked bs like that, perhaps you should present your problem, explain you decisions, and ask where you may be missing something important.

Or just go use Redis or whatever else you like, and abuse their maintainers when something isn't up to your liking.

[michaelklishin]

What the protocol does not assume is that a consumer (a subscription) is reused across queues. Reusing an object or function in an application is not the same as reusing a subscription: AMQP 0-9-1 does not allow you to do that. In fact, neither do MQTT, STOMP, and to my knowledge AMQP 1.0.

This piece of code in a loop

  consumeFromQueue := func(queueName string) {
    q, err := ch.QueueDeclare(
       // ....
    )

    msgs, err := ch.Consume(
      q.Name, // queue
        // ...
    )

would be a loop of subscriptions in MQTT, STOMP (if the destinations are known to exist, otherwise they will have to be declared by the broker), AMQP 1.0, and the RabbitMQ stream protocol. Not a whole lot of difference with AMQP 0-9-1 semantically, although one protocol method (a round trip in this case, to declare a queue) would be avoided.

So your options are (pick some or do all of them):

• Use a smaller and stable number of queues
• Use a stable number of consumers
• If you need a mass fanout (e.g. if a large group of consumers can come up online all at once), use a stream with a dedicated stream protocol client for Go

Each consumer consumes from only one queue at a time in AMQP 0-9-1. It is also roughly the same story in other protocols but the step of declaring a queue (destination) may be implicit.