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