jetstream golang API

[carr123]

jetstream golang API is hard to understand and use.
is there any complete demonstration on how to invoke those APIs in different use scenerios, not just the "basic usage" ?
the jetstream doc on the official website using CLI to explains concepts. that doesn't help me too much. I want to talk to jetstream server programmably through golang API.

[wallyqs]

Thanks for the feedback , have you seen some of the examples from here? https://pkg.go.dev/github.com/nats-io/nats.go#example-JetStream

[aperrot42]

Hi, same issue here, betwwen the (outdated ?) official documentation that only references using the CLI, and the various implementations we can find, I am also lost, and it seems like my colleagues are as well.

As a go developper interacting with Jetstream nats server, should I use :

https://github.com/nats-io/nats.go (it contains jsm.go)
https://github.com/nats-io/jsm.go (it seems to contain some (a lot of) helpers)
The whole model and logic seems to differ, between pull-push consumers / subcriptions : for instance, using this library, it feels like pull based consumer are kind of "deprecated".

Bref, If you can help me find the current reference implementation of jetstream go client that would be great.

Best,

EDIT : it seems like jsm.go is not to be used, that leaves go developpers with nats.go library : nats-io/jsm.go#258

@wallyqs it is true that we lack examples illustrating the concept of streams for instance.

[aperrot42]

Also no mention of the "STREAM" concept in jetstream exemples from nats.go

[syacko-sote]

@aperrot42
You should use https://github.com/nats-io/nats.go

We use Pull for the most part and they are not deprecated.

Hope this helps,
Scott

[carr123]

sorry ro bother.
I create a working queue (using WorkQueuePolicy), then i create two pull-based consumers, i expect them to consume one subject(named "ORDERS.created") concurrently, but error occurs.
"nats: filtered consumer not unique on workqueue stream"

does that mean in WorkQueuePolicy one subject only support one single consumer ？

[derekcollison]

Could you share your code?

[carr123]

1. create stream

_, err = js.AddStream(&nats.StreamConfig{
		Name:      "ORDERS",
		Subjects:  []string{"ORDERS.*"},
		Storage:   nats.FileStorage,
		Replicas:  3,
		Retention: nats.WorkQueuePolicy,
		Discard:   nats.DiscardNew,
		MaxMsgs:   -1,
		MaxAge:    time.Hour * 24 * 365,
	})
	if err != nil {
		log.Println(err)
		return
	}

2. send messages

	for i := 0; i < 10000; i++ {
		szMsg := fmt.Sprintf("js msg %d", i)
		js.PublishAsync(subjectName, []byte(szMsg))
	}
	<-js.PublishAsyncComplete()

3. read message

    func recvmsg(nWhich int) {
        nc, err := nats.Connect(.........)
        js, err := nc.JetStream()

        durableName := fmt.Sprintf("consumer%d", nWhich)
      	sub, err := js.PullSubscribe("ORDERS.created", durableName, nats.BindStream("ORDERS"))
	if err != nil {
		fmt.Println("PullSubscribe:", err)
		return
	}
	for {
		msgs, _ := sub.Fetch(10, nats.Context(context.Background()))
		for _, msg := range msgs {
			log.Println(durableName+" recv data:", string(msg.Data))
			msg.Ack()
		}
	}
   }
  
  for i:=0; i<2; i++ {
    go recvmsg(i) 
 }

[derekcollison]

You should create a single durable consumer.

Then bind subscribers to that consumer.

 sub, err := js.PullSubscribe("ORDERS.created", durableName, nats.Bind("ORDERS", durableName))

[carr123]

thanks derekcollison , I follow your instruction, and got correct results !

[carr123]

is there any convienent function like js.AddStreamIfNotExist, js.AddConsumerIfNotExist ?

[wallyqs]

@carr123 you can just call js.AddStream and js.AddConsumer and should be idempotent if already created the strea/consumer.

[carr123]

_, err = js.AddStream(&nats.StreamConfig{
Name: "ORDERS",
Subjects: []string{"ORDERS.*"},
Storage: nats.FileStorage,
Replicas: 3,
Retention: nats.WorkQueuePolicy,
Discard: nats.DiscardNew,
MaxMsgs: -1,
MaxAge: time.Hour * 24 * 365,
})
if err != nil {
log.Println("AddStream fail:", err)
return
}

the code above run a second time, error printed : "AddStream fail: stream name already in use"

[wallyqs]

Ah that's right... There is no AddStreamIfNotExist API, maybe we should add... The workaround then would be to check error like:

	_, err = js.AddStream(&nats.StreamConfig{
		Name:     "ORDERS",
		Replicas: 3,
	})
	if err != nil && !strings.Contains(err.Error(), "already in use") {
		log.Fatal(err)
	}

[carr123]

thanks to your advice

[carr123]

_, err = js.AddStream(&nats.StreamConfig{
	Name:      streamName,
	Subjects:  []string{streamSubjects},
	Storage:   nats.FileStorage,
	Replicas:  3,
	Retention: nats.InterestPolicy,
	MaxMsgs:   -1,
	Discard:   nats.DiscardOld,
	MaxAge:    time.Hour * 24 * 30,
	NoAck:     true,
})

  _, err := js.AddConsumer(streamName, &nats.ConsumerConfig{
	Durable:       durableName,
	DeliverPolicy: nats.DeliverAllPolicy,
	AckPolicy:     nats.AckNonePolicy,
	ReplayPolicy:  nats.ReplayInstantPolicy,
	FilterSubject: subjectName,
})

 ERROR report:  "consumer in pull mode requires explicit ack policy"

how can I create a consumer in push mode using  js.AddConsumer ?

[wallyqs]

to create a consumer in push mode you have to set a deliver subject:

	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)
	}