How to pull from a subject?

[AhmedAbouelkher]

Hi,

I'm testing pull based subscriptions and I noticed a very strange behavior.

I have a stream called new which has three subjects sub, node and oak and I'm trying to process every subject's messages separately, but when using PullSubscribe the consumer function keeps fetching messages from all the 3 subjects (in a robin round) which is not what I want clearly.

var nc *nats.Conn
nc, err := nats.Connect("nats://127.0.0.1:4222")
if err != nil {
    log.Fatal(err)
}
defer nc.Close()

js, err := nc.JetStream()
if err != nil {
    log.Fatal(err)
}

_, err = js.AddConsumer("new", &nats.ConsumerConfig{
    Durable:   "new_consumer",
    AckPolicy: nats.AckExplicitPolicy,
})
if err != nil {
    log.Fatalln(err)
}

go func() {
    for {

        select {
        case <-signals:
            return
        default:
        }

        sub, err := js.PullSubscribe(
            "sub",
            "new_consumer",
            nats.Bind("new", "new_consumer"),
        )
        if err != nil {
            log.Println(err)
            continue
        }

        messages, err := sub.Fetch(1)
        if err != nil {
            log.Println(err)
            continue
        }

        msg := messages[0]
        log.Printf("consuming message %s\n" msg.Data)
        msg.Ack()

        time.Sleep(time.Second)
    }
}()

//... rest of the code

When subscription replaced with js.Subscribe() the issue disappeared and only consumes from the givin subject, but even then I will loose durability.

var nc *nats.Conn
nc, err := nats.Connect("nats://127.0.0.1:4222")
if err != nil {
    log.Fatal(err)
}
defer nc.Close()

js, err := nc.JetStream()
if err != nil {
    log.Fatal(err)
}

_, err = js.AddConsumer("new", &nats.ConsumerConfig{
    Durable:   "new_consumer",
    AckPolicy: nats.AckExplicitPolicy,
})
if err != nil {
    log.Fatalln(err)
}

js.Subscribe("sub", func(msg *nats.Msg) {
    log.Printf("consuming message %s\n" msg.Data)
    msg.Ack()
})

//... rest of the code

Side Question

How to make the consumer like Kafka ignores the old messages if I have hundreds of them and the consumer is newly created?

[wallyqs]

Hi, this is because when you are creating the consumer via AddConsumer it is being created without a FilterSubject so it would match all the messages, and then later js.PullSubscribe("sub", "new_consumer", nats.Bind("new", "new_consumer") the Bind will reuse the created consumer without having to use the sub for the lookup to find the consumer so it becomes ignored.
If you remove AddConsumer and then do js.PullSubscribe("sub", "new_consumer") without Bind, then the consumer would be auto created with sub as a filter subject to narrow down the messages.