Add in deliver status for consumer info

Signed-off-by: Derek Collison <[email protected]>

Author: derekcollison

server/consumer.go

@@ -43,6 +43,7 @@ type ConsumerInfo struct {
 	NumRedelivered int             `json:"num_redelivered"`
 	NumWaiting     int             `json:"num_waiting"`
 	NumPending     uint64          `json:"num_pending"`
+	Active         *PushActive     `json:"active,omitempty"`
 	Cluster        *ClusterInfo    `json:"cluster,omitempty"`
 }
 
@@ -69,6 +70,11 @@ type ConsumerConfig struct {
 	Direct bool `json:"direct,omitempty"`
 }
 
+type PushActive struct {
+	Subject string `json:"subject,omitempty"`
+	Queue   string `json:"queue,omitempty"`
+}
+
 type CreateConsumerRequest struct {
 	Stream string         `json:"stream_name"`
 	Config ConsumerConfig `json:"config"`
@@ -190,6 +196,7 @@ type consumer struct {
 	asflr             uint64
 	sgap              uint64
 	dsubj             string
+	qgroup            string
 	lss               *lastSeqSkipList
 	rlimit            *rate.Limiter
 	reqSub            *subscription
@@ -688,6 +695,27 @@ func (o *consumer) setConsumerAssignment(ca *consumerAssignment) {
 	}
 }
 
+// checkInterest will check on our interest's queue group status.
+// Lock should be held.
+func (o *consumer) checkQueueInterest() {
+	if !o.active || o.cfg.DeliverSubject == _EMPTY_ {
+		return
+	}
+	subj := o.dsubj
+	if subj == _EMPTY_ {
+		subj = o.cfg.DeliverSubject
+	}
+
+	if rr := o.acc.sl.Match(subj); len(rr.qsubs) > 0 {
+		// Just grab first
+		if qsubs := rr.qsubs[0]; len(qsubs) > 0 {
+			if sub := rr.qsubs[0][0]; len(sub.queue) > 0 {
+				o.qgroup = string(sub.queue)
+			}
+		}
+	}
+}
+
 // Lock should be held.
 func (o *consumer) isLeader() bool {
 	if o.node != nil {
@@ -763,6 +791,8 @@ func (o *consumer) setLeader(isLeader bool) {
 					stopAndClearTimer(&o.gwdtmr)
 					o.gwdtmr = time.AfterFunc(time.Second, func() { o.watchGWinterest() })
 				}
+			} else {
+				o.checkQueueInterest()
 			}
 		}
 
@@ -968,7 +998,12 @@ func (o *consumer) updateDeliveryInterest(localInterest bool) bool {
 	if interest && !o.active {
 		o.signalNewMessages()
 	}
-	o.active = interest
+	// Update active status, if not active clear any queue group we captured.
+	if o.active = interest; !o.active {
+		o.qgroup = _EMPTY_
+	} else {
+		o.checkQueueInterest()
+	}
 
 	// If the delete timer has already been set do not clear here and return.
 	if o.dtmr != nil && !o.isDurable() && !interest {
@@ -1433,13 +1468,21 @@ func (o *consumer) writeStoreState() error {
 
 // Info returns our current consumer state.
 func (o *consumer) info() *ConsumerInfo {
+	var pa *PushActive
+
 	o.mu.RLock()
 	mset := o.mset
 	if mset == nil || mset.srv == nil {
 		o.mu.RUnlock()
 		return nil
 	}
 	js := o.js
+	if o.active {
+		pa = &PushActive{
+			Subject: o.dsubj,
+			Queue:   o.qgroup,
+		}
+	}
 	o.mu.RUnlock()
 
 	if js == nil {
@@ -1468,6 +1511,7 @@ func (o *consumer) info() *ConsumerInfo {
 		NumAckPending:  len(o.pending),
 		NumRedelivered: len(o.rdc),
 		NumPending:     o.adjustedPending(),
+		Active:         pa,
 		Cluster:        ci,
 	}
 	// If we are a pull mode consumer, report on number of waiting requests.

server/jetstream_test.go

@@ -11801,10 +11801,6 @@ func TestJetStreamDomainInPubAck(t *testing.T) {
 	s, _ := RunServerWithConfig(conf)
 	defer s.Shutdown()
 
-	if !s.JetStreamEnabled() {
-		t.Fatalf("Expected JetStream to be enabled")
-	}
-
 	config := s.JetStreamConfig()
 	if config != nil {
 		defer removeDir(t, config.StoreDir)
@@ -11814,10 +11810,9 @@ func TestJetStreamDomainInPubAck(t *testing.T) {
 	defer nc.Close()
 
 	cfg := &nats.StreamConfig{
-		Name:         "TEST",
-		Storage:      nats.MemoryStorage,
-		Subjects:     []string{"foo"},
-		MaxConsumers: 1,
+		Name:     "TEST",
+		Storage:  nats.MemoryStorage,
+		Subjects: []string{"foo"},
 	}
 	if _, err := js.AddStream(cfg); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -11835,6 +11830,106 @@ func TestJetStreamDomainInPubAck(t *testing.T) {
 	}
 }
 
+func TestJetStreamPushConsumerInfo(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	config := s.JetStreamConfig()
+	if config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	cfg := &nats.StreamConfig{
+		Name:     "TEST",
+		Storage:  nats.MemoryStorage,
+		Subjects: []string{"foo"},
+	}
+	if _, err := js.AddStream(cfg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// We want to test extended consumer info for push based consumers.
+	// We need to do these by hand for now.
+	createConsumer := func(name, deliver string) {
+		t.Helper()
+		creq := CreateConsumerRequest{
+			Stream: "TEST",
+			Config: ConsumerConfig{
+				Durable:        name,
+				DeliverSubject: deliver,
+			},
+		}
+		req, err := json.Marshal(creq)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		resp, err := nc.Request(fmt.Sprintf(JSApiDurableCreateT, "TEST", name), req, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var ccResp JSApiConsumerCreateResponse
+		if err := json.Unmarshal(resp.Data, &ccResp); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ccResp.ConsumerInfo == nil || ccResp.Error != nil {
+			t.Fatalf("Got a bad response %+v", ccResp)
+		}
+	}
+
+	consumerInfo := func(name string) *ConsumerInfo {
+		t.Helper()
+		resp, err := nc.Request(fmt.Sprintf(JSApiConsumerInfoT, "TEST", name), nil, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var cinfo JSApiConsumerInfoResponse
+		if err := json.Unmarshal(resp.Data, &cinfo); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if cinfo.ConsumerInfo == nil || cinfo.Error != nil {
+			t.Fatalf("Got a bad response %+v", cinfo)
+		}
+		return cinfo.ConsumerInfo
+	}
+
+	// First create a durable push and make sure we show now active status.
+	createConsumer("dlc", "d.X")
+	if ci := consumerInfo("dlc"); ci.Active != nil {
+		t.Fatalf("Expected active to be nil, got %+v\n", ci.Active)
+	}
+	// Now bind the deliver subject.
+	sub, _ := nc.SubscribeSync("d.X")
+	nc.Flush() // Make sure it registers.
+	// Check that its reported.
+	if ci := consumerInfo("dlc"); ci.Active == nil || ci.Active.Subject != "d.X" {
+		t.Fatalf("Expected active to be set and have subject %q, got %+v\n", "d.X", ci.Active)
+	}
+	sub.Unsubscribe()
+	nc.Flush() // Make sure it registers.
+	if ci := consumerInfo("dlc"); ci.Active != nil {
+		t.Fatalf("Expected active to be nil, got %+v\n", ci.Active)
+	}
+
+	// Now make sure we have queue groups indictated as needed.
+	createConsumer("ik", "d.Z")
+	// Now bind the deliver subject with a queue group.
+	sub, _ = nc.QueueSubscribeSync("d.Z", "g22")
+	defer sub.Unsubscribe()
+	nc.Flush() // Make sure it registers.
+	// Check that queue group reported.
+	if ci := consumerInfo("ik"); ci.Active == nil || ci.Active.Subject != "d.Z" || ci.Active.Queue != "g22" {
+		t.Fatalf("Expected active to be set and have subject %q and queue %q, got %+v\n", "d.Z", "g22", ci.Active)
+	}
+	sub.Unsubscribe()
+	nc.Flush() // Make sure it registers.
+	if ci := consumerInfo("ik"); ci.Active != nil {
+		t.Fatalf("Expected active to be nil, got %+v\n", ci.Active)
+	}
+}
+
 // Issue #2213
 func TestJetStreamDirectConsumersBeingReported(t *testing.T) {
 	s := RunBasicJetStreamServer()