Support for new stream subject transform functionalities

cli/consumer_command.go

@@ -578,6 +578,7 @@ func (c *consumerCmd) showInfo(config api.ConsumerConfig, state api.ConsumerInfo
 	} else {
 		cols.AddRow("Pull Mode", true)
 	}
+
 	if config.FilterSubject != "" {
 		cols.AddRow("Filter Subject", config.FilterSubject)
 	} else if len(config.FilterSubjects) > 0 {
@@ -672,6 +673,7 @@ func (c *consumerCmd) showInfo(config api.ConsumerConfig, state api.ConsumerInfo
 	}
 
 	cols.AddRow("Unprocessed Messages", state.NumPending)
+
 	if config.DeliverSubject == "" {
 		if config.MaxWaiting > 0 {
 			cols.AddRowf("Waiting Pulls", "%s of maximum %s", f(state.NumWaiting), f(config.MaxWaiting))
@@ -698,6 +700,7 @@ func (c *consumerCmd) infoAction(_ *fisk.ParseContext) error {
 
 	var err error
 	consumer := c.selectedConsumer
+
 	if consumer == nil {
 		consumer, err = c.mgr.LoadConsumer(c.stream, c.consumer)
 		fisk.FatalIfError(err, "could not load Consumer %s > %s", c.stream, c.consumer)

cli/kv_command.go

@@ -58,6 +58,7 @@ type kvCommand struct {
 	repubHeadersOnly      bool
 	mirror                string
 	mirrorDomain          string
+	sources               []string
 }
 
 func configureKVCommand(app commandHost) {
@@ -88,6 +89,7 @@ for an indefinite period or a per-bucket configured TTL.
 	add.Flag("republish-headers", "Republish only message headers, no bodies").UnNegatableBoolVar(&c.repubHeadersOnly)
 	add.Flag("mirror", "Creates a mirror of a different bucket").StringVar(&c.mirror)
 	add.Flag("mirror-domain", "When mirroring find the bucket in a different domain").StringVar(&c.mirrorDomain)
+	add.Flag("source", "Source from a different bucket").PlaceHolder("BUCKET").StringsVar(&c.sources)
 
 	add.PreAction(c.parseLimitStrings)
 
@@ -464,6 +466,12 @@ func (c *kvCommand) addAction(_ *fisk.ParseContext) error {
 		}
 	}
 
+	for _, source := range c.sources {
+		cfg.Sources = append(cfg.Sources, &nats.StreamSource{
+			Name: source,
+		})
+	}
+
 	store, err := js.CreateKeyValue(cfg)
 	if err != nil {
 		return err
@@ -774,14 +782,33 @@ func (c *kvCommand) showStatus(store nats.KeyValue) error {
 			s := nfo.Mirror
 			cols.AddSectionTitle("Mirror Information")
 			cols.AddRow("Origin Bucket", strings.TrimPrefix(s.Name, "KV_"))
-			cols.AddRowIf("External API", s.External.APIPrefix, s.External != nil)
+			if s.External != nil {
+				cols.AddRow("External API", s.External.APIPrefix)
+			}
 			if s.Active > 0 && s.Active < math.MaxInt64 {
 				cols.AddRow("Last Seen", s.Active)
 			} else {
 				cols.AddRowf("Last Seen", "never")
 			}
 			cols.AddRow("Lag", s.Lag)
 		}
+
+		if len(nfo.Sources) > 0 {
+			cols.AddSectionTitle("Sources Information")
+			for _, source := range nfo.Sources {
+				cols.AddRow("Source Bucket", strings.TrimPrefix(source.Name, "KV_"))
+				if source.External != nil {
+					cols.AddRow("External API", source.External.APIPrefix)
+				}
+				if source.Active > 0 && source.Active < math.MaxInt64 {
+					cols.AddRow("Last Seen", source.Active)
+				} else {
+					cols.AddRow("Last Seen", "never")
+				}
+				cols.AddRow("Lag", source.Lag)
+			}
+		}
+
 		if nfo.Cluster != nil {
 			cols.AddSectionTitle("Cluster Information")
 			renderNatsGoClusterInfo(cols, nfo)