js: Ephemeral consumers with js.QueueSubscribe act like Subscribe

[wallyqs]

Currently js.QueueSubscribe unless using a durable or binding to a consumer with a durable name, no load balancing would occur when other members call js.QueueSubscribe("foo", "bar") since each call would have its own ephemeral consumer.

[aricart]

@wallyqs this is the same behaviour as the javascript clients - the change would require that the (maybe generated) name of the consumer be passed, OR would need some mechanism to filter all consumers based on a subject. Perhaps I am missing something here.

[wallyqs]

@aricart Yes that's the way it currently works. A few users have run into this expecting load balancing a la core NATS QueueSubscribe so maybe should improve somehow.

[derekcollison]

We should be able to bind to any consumer. All are named (ConsumerInfo shows it), its just their lifetime changes whether or not its durable or ephemeral.

[rwrz]

I tried to use the QueueSubscribe. but even using durable, it is not working.
Can you provide an example, please?

[derekcollison]

When subscribing if we can bind to an existing consumer we should, meaning js.QueueSubscribe should work they way its intended.

Does this not behave this way for the JS clients?

[wallyqs]

@rwrz Below you can find a couple of examples of using js.QueueSubscribe:

QueueSubscribe with Durable

package main

import (
	"fmt"
	"log"

	"github.com/nats-io/nats.go"
)

func init() {
	log.SetFlags(log.LstdFlags | log.Lshortfile)
}

func main() {
	nc, err := nats.Connect("localhost")
	if err != nil {
		log.Fatal(err)
	}
	js, err := nc.JetStream()
	if err != nil {
		log.Fatal(err)
	}

	js.AddStream(&nats.StreamConfig{
		Name:     "foo",
		Subjects: []string{"foo"},
	})

	for i := 0; i < 20; i++ {
		n := i
		js.QueueSubscribe("foo", "bar", func(msg *nats.Msg) {
			log.Printf("[Received n=%-2d]:\t%+v", n, string(msg.Data))
		}, nats.Durable("consumer"))
	}

	for i := 0; i < 1000; i++ {
		js.Publish("foo", []byte(fmt.Sprintf("i:%v", i)))
	}
	nc.Drain()
}

QueueSubscribe with ephemeral

package main

import (
	"fmt"
	"log"

	"github.com/nats-io/nats.go"
)

func init() {
	log.SetFlags(log.LstdFlags | log.Lshortfile)
}

func main() {
	nc, err := nats.Connect("localhost")
	if err != nil {
		log.Fatal(err)
	}
	js, err := nc.JetStream()
	if err != nil {
		log.Fatal(err)
	}

	js.AddStream(&nats.StreamConfig{
		Name:     "foo",
		Subjects: []string{"foo"},
	})

	inbox := nats.NewInbox()
	// Avoid: 'consumer requires interest for delivery subject when ephemeral'
	qsub, err := nc.QueueSubscribeSync(inbox, "bar")
	if err != nil {
		log.Fatal(err)
	}

	consumer, err := js.AddConsumer("foo", &nats.ConsumerConfig{
		DeliverSubject: inbox,
		AckPolicy:      nats.AckExplicitPolicy,
	})
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("%+v", consumer)

	for i := 0; i < 20; i++ {
		n := i
		js.QueueSubscribe("foo", "bar", func(msg *nats.Msg) {
			log.Printf("[Received n=%-2d]:\t%+v", n, string(msg.Data))
		}, nats.Bind("foo", consumer.Name))
	}

	// Remove original subscription.
	qsub.Unsubscribe()

	for i := 0; i < 1000; i++ {
		js.Publish("foo", []byte(fmt.Sprintf("i:%v", i)))
	}
	nc.Drain()
}

[stepanbujnak]

@wallyqs your example with ephemeral consumer is only valid when running in a single process. You can't do this reliably from multiple processes or multiple machines, because you wouldn't know the name of the ephemeral consumer and even if you could query it you'll end up with race conditions.

[derekcollison]

There are ways around that but you do raise a good point. Several folks have suggested being able to name ephemerals as a potential workaround.