Repository: nats-io/nats.go
Pull request #1237 (2023-03-23 18:51:54)

[CHANGED] Use different subjects per PR when using Fetch and FetchBatch
This PR adds separate reply subjects for each pull request sent with `Fetch()` or `FetchBatch()`. This allows us to ignore timeouts from different calls.


Patch:
diff --git a/js.go b/js.go
index 1f52b63..99ee347 100644
--- a/js.go
+++ b/js.go
@@ -1610,6 +1610,8 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 	if isPullMode {
 		nms = fmt.Sprintf(js.apiSubj(apiRequestNextT), stream, consumer)
 		deliver = nc.NewInbox()
+		// for pull consumers, create a wildcard subscription to differentiate pull requests
+		deliver += ".*"
 	}
 
 	// In case this has a context, then create a child context that
@@ -2629,7 +2631,7 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 
 	nc := sub.conn
 	nms := sub.jsi.nms
-	rply := sub.jsi.deliver
+	rply, reqID := newFetchInbox(jsi.deliver)
 	js := sub.jsi.js
 	pmc := len(sub.mch) > 0
 
@@ -2751,10 +2753,17 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 					// wait this time.
 					noWait = false
 					err = sendReq()
-				} else if err == ErrTimeout && len(msgs) == 0 {
+				} else if err == ErrTimeout {
 					// If we get a 408, we will bail if we already collected some
-					// messages, otherwise ignore and go back calling NextMsg.
-					err = nil
+					// messages, otherwise ignore and go back calling nextMsg.
+					if len(msgs) == 0 {
+						err = nil
+						continue
+					}
+					// ignore timeout message from server if it comes from a different pull request
+					if reqID != "" && !subjectMatchesReqID(msg.Subject, reqID) {
+						err = nil
+					}
 				}
 			}
 		}
@@ -2766,6 +2775,28 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 	return msgs, nil
 }
 
+// newFetchInbox returns subject used as reply subject when sending pull requests
+// as well as request ID. For non-wildcard subject, request ID is empty and
+// passed subject is not transformed
+func newFetchInbox(subj string) (string, string) {
+	if !strings.HasSuffix(subj, ".*") {
+		return subj, ""
+	}
+	reqID := nuid.Next()
+	var sb strings.Builder
+	sb.WriteString(subj[:len(subj)-1])
+	sb.WriteString(reqID)
+	return sb.String(), reqID
+}
+
+func subjectMatchesReqID(subject, reqID string) bool {
+	subjectParts := strings.Split(subject, ".")
+	if len(subjectParts) < 2 {
+		return false
+	}
+	return subjectParts[len(subjectParts)-1] == reqID
+}
+
 // MessageBatch provides methods to retrieve messages consumed using [Subscribe.FetchBatch].
 type MessageBatch interface {
 	// Messages returns a channel on which messages will be published.
@@ -2835,7 +2866,7 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 
 	nc := sub.conn
 	nms := sub.jsi.nms
-	rply := sub.jsi.deliver
+	rply, reqID := newFetchInbox(sub.jsi.deliver)
 	js := sub.jsi.js
 	pmc := len(sub.mch) > 0
 
@@ -2964,6 +2995,10 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 			usrMsg, err = checkMsg(msg, true, false)
 			if err != nil {
 				if err == ErrTimeout {
+					if reqID != "" && !subjectMatchesReqID(msg.Subject, reqID) {
+						// ignore timeout message from server if it comes from a different pull request
+						continue
+					}
 					err = nil
 				}
 				break
diff --git a/micro/service.go b/micro/service.go
index ed151b8..79f0cc9 100644
--- a/micro/service.go
+++ b/micro/service.go
@@ -108,15 +108,14 @@ type (
 
 	// EndpointStats contains stats for a specific endpoint.
 	EndpointStats struct {
-		Name                  string            `json:"name"`
-		Subject               string            `json:"subject"`
-		Metadata              map[string]string `json:"metadata"`
-		NumRequests           int               `json:"num_requests"`
-		NumErrors             int               `json:"num_errors"`
-		LastError             string            `json:"last_error"`
-		ProcessingTime        time.Duration     `json:"processing_time"`
-		AverageProcessingTime time.Duration     `json:"average_processing_time"`
-		Data                  json.RawMessage   `json:"data,omitempty"`
+		Name                  string          `json:"name"`
+		Subject               string          `json:"subject"`
+		NumRequests           int             `json:"num_requests"`
+		NumErrors             int             `json:"num_errors"`
+		LastError             string          `json:"last_error"`
+		ProcessingTime        time.Duration   `json:"processing_time"`
+		AverageProcessingTime time.Duration   `json:"average_processing_time"`
+		Data                  json.RawMessage `json:"data,omitempty"`
 	}
 
 	// Ping is the response type for PING monitoring endpoint.
@@ -466,9 +465,8 @@ func addEndpoint(s *service, name, subject string, handler Handler, schema *Sche
 	endpoint.subscription = sub
 	s.endpoints = append(s.endpoints, endpoint)
 	endpoint.stats = EndpointStats{
-		Name:     name,
-		Subject:  subject,
-		Metadata: endpoint.Metadata,
+		Name:    name,
+		Subject: subject,
 	}
 	return nil
 }
@@ -733,7 +731,6 @@ func (s *service) Stats() Stats {
 		endpointStats := &EndpointStats{
 			Name:                  endpoint.stats.Name,
 			Subject:               endpoint.stats.Subject,
-			Metadata:              endpoint.stats.Metadata,
 			NumRequests:           endpoint.stats.NumRequests,
 			NumErrors:             endpoint.stats.NumErrors,
 			LastError:             endpoint.stats.LastError,
@@ -848,9 +845,8 @@ func (e *Endpoint) stop() error {
 
 func (e *Endpoint) reset() {
 	e.stats = EndpointStats{
-		Name:     e.stats.Name,
-		Subject:  e.stats.Subject,
-		Metadata: e.stats.Metadata,
+		Name:    e.stats.Name,
+		Subject: e.stats.Subject,
 	}
 }
 


Test patch:
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index 2dae0e1..d168fbe 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -1159,9 +1159,8 @@ func TestServiceStats(t *testing.T) {
 				Version: "0.1.0",
 				APIURL:  "http://someapi.com/v1",
 				Endpoint: &micro.EndpointConfig{
-					Subject:  "test.func",
-					Handler:  micro.HandlerFunc(handler),
-					Metadata: map[string]string{"test": "value"},
+					Subject: "test.func",
+					Handler: micro.HandlerFunc(handler),
 					Schema: &micro.Schema{
 						Request:  "some_request",
 						Response: "some_response",
@@ -1279,11 +1278,6 @@ func TestServiceStats(t *testing.T) {
 			if stats.Type != micro.StatsResponseType {
 				t.Errorf("Invalid response type; want: %s; got: %s", micro.StatsResponseType, stats.Type)
 			}
-			if test.config.Endpoint != nil && test.config.Endpoint.Metadata != nil {
-				if !reflect.DeepEqual(test.config.Endpoint.Metadata, stats.Endpoints[0].Metadata) {
-					t.Errorf("invalid endpoint metadata: %v", stats.Endpoints[0].Metadata)
-				}
-			}
 
 			if test.expectedStats != nil {
 				var data map[string]interface{}