jetstream_consumer.go

@@ -14,10 +14,17 @@ func (js JetStream) GetConsumerInfo(stream, name string) (*nats.ConsumerInfo, er
 
 // AddConsumer ...
 func (js JetStream) AddConsumer(stream, subject, name string) error {
+	// Get consumer first, return if existed
+	consumer, err := js.GetConsumerInfo(stream, name)
+	if consumer != nil {
+		return nil
+	}
+
+	// Generate channel name
 	channel := combineStreamAndSubjectName(stream, subject)
 
 	// Add
-	_, err := js.instance.AddConsumer(stream, &nats.ConsumerConfig{
+	_, err = js.instance.AddConsumer(stream, &nats.ConsumerConfig{
 		Durable:       name,
 		AckPolicy:     nats.AckExplicitPolicy,
 		FilterSubject: channel,

server_reqres.go

@@ -22,11 +22,7 @@ func (sv Server) Request(subject string, payload []byte) (*nats.Msg, error) {
 
 // Reply ...
 func (sv Server) Reply(msg *nats.Msg, payload []byte) error {
-	err := sv.instance.Publish(msg.Reply, payload)
+	return sv.instance.Publish(msg.Reply, payload)
-
-	// Ack message
-	msg.Ack()
-	return err
 }
 
 // Subscribe ...