Merge pull request #2218 from nats-io/si_ext

Fix stream source lookup and add in optional External to StreamSource

Author: derekcollison

server/jetstream_cluster_test.go

@@ -6367,6 +6367,51 @@ func TestJetStreamClusterDomainsAndSameNameSources(t *testing.T) {
 	if si.State.Msgs != 2 {
 		t.Fatalf("Expected 2 msgs, got %d", si.State.Msgs)
 	}
+
+	// Make sure we can see our external information.
+	// This not in the Go client yet so manual for now.
+	resp, err := nc.Request(fmt.Sprintf(JSApiStreamInfoT, "S"), nil, time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	var ssi StreamInfo
+	if err = json.Unmarshal(resp.Data, &ssi); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if len(ssi.Sources) != 2 {
+		t.Fatalf("Expected 2 source streams, got %d", len(ssi.Sources))
+	}
+	if ssi.Sources[0].External == nil {
+		t.Fatalf("Expected a non-nil external designation")
+	}
+	if ssi.Sources[0].External.ApiPrefix != "$JS.SPOKE-1.API" {
+		t.Fatalf("Expected external api of %q, got %q", "$JS.SPOKE-1.API", ssi.Sources[0].External.ApiPrefix)
+	}
+
+	// Also create a mirror.
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name: "M",
+		Mirror: &nats.StreamSource{
+			Name:     "TEST",
+			External: &nats.ExternalStream{APIPrefix: "$JS.SPOKE-1.API"},
+		},
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	resp, err = nc.Request(fmt.Sprintf(JSApiStreamInfoT, "M"), nil, time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if err = json.Unmarshal(resp.Data, &ssi); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if ssi.Mirror == nil || ssi.Mirror.External == nil {
+		t.Fatalf("Expected a non-nil external designation for our mirror")
+	}
+	if ssi.Mirror.External.ApiPrefix != "$JS.SPOKE-1.API" {
+		t.Fatalf("Expected external api of %q, got %q", "$JS.SPOKE-1.API", ssi.Sources[0].External.ApiPrefix)
+	}
 }
 
 func TestJetStreamClusterLeafDifferentAccounts(t *testing.T) {

server/stream.go

@@ -103,10 +103,11 @@ type PeerInfo struct {
 
 // StreamSourceInfo shows information about an upstream stream source.
 type StreamSourceInfo struct {
-	Name   string        `json:"name"`
-	Lag    uint64        `json:"lag"`
-	Active time.Duration `json:"active"`
-	Error  *ApiError     `json:"error,omitempty"`
+	Name     string          `json:"name"`
+	External *ExternalStream `json:"external,omitempty"`
+	Lag      uint64          `json:"lag"`
+	Active   time.Duration   `json:"active"`
+	Error    *ApiError       `json:"error,omitempty"`
 }
 
 // StreamSource dictates how streams can source from other streams.
@@ -117,7 +118,7 @@ type StreamSource struct {
 	FilterSubject string          `json:"filter_subject,omitempty"`
 	External      *ExternalStream `json:"external,omitempty"`
 
-	// Internal
+	// Internale
 	iname string // For indexing when stream names are the same for multiple sources.
 }
 
@@ -1151,7 +1152,20 @@ func (mset *stream) sourceInfo(si *sourceInfo) *StreamSourceInfo {
 	if si == nil {
 		return nil
 	}
-	return &StreamSourceInfo{Name: si.name, Lag: si.lag, Active: time.Since(si.last), Error: si.err}
+	ssi := &StreamSourceInfo{Name: si.name, Lag: si.lag, Active: time.Since(si.last), Error: si.err}
+	var ext *ExternalStream
+	if mset.cfg.Mirror != nil {
+		ext = mset.cfg.Mirror.External
+	} else if ss := mset.streamSource(si.iname); ss != nil && ss.External != nil {
+		ext = ss.External
+	}
+	if ext != nil {
+		ssi.External = &ExternalStream{
+			ApiPrefix:     ext.ApiPrefix,
+			DeliverPrefix: ext.DeliverPrefix,
+		}
+	}
+	return ssi
 }
 
 // Return our source info for our mirror.
@@ -1563,9 +1577,9 @@ func (mset *stream) setupMirrorConsumer() error {
 	return nil
 }
 
-func (mset *stream) streamSource(sname string) *StreamSource {
+func (mset *stream) streamSource(iname string) *StreamSource {
 	for _, ssi := range mset.cfg.Sources {
-		if ssi.Name == sname {
+		if ssi.iname == iname {
 			return ssi
 		}
 	}
@@ -1626,7 +1640,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 
 	si.sseq, si.dseq = seq, 0
 	si.last = time.Now()
-	ssi := mset.streamSource(si.name)
+	ssi := mset.streamSource(iname)
 
 	// Determine subjects etc.
 	var deliverSubject string