package natsio import ( "errors" "fmt" "github.com/nats-io/nats.go" ) // GetConsumerInfo ... func (js JetStream) GetConsumerInfo(consumerName string) (*nats.ConsumerInfo, error) { return js.instance.ConsumerInfo(js.streamName, consumerName) } // AddConsumer ... func (js JetStream) AddConsumer(consumerName, filterSubject string) error { // get consumer first, return if existed consumer, _ := js.GetConsumerInfo(consumerName) if consumer != nil { return nil } // add _, err := js.instance.AddConsumer(js.streamName, &nats.ConsumerConfig{ Durable: consumerName, AckPolicy: nats.AckExplicitPolicy, FilterSubject: filterSubject, }) if err != nil { msg := fmt.Sprintf("[natsio.JetStream] add consumer %s to stream #%s error: %s", consumerName, js.streamName, err.Error()) return errors.New(msg) } return nil }