Add support for stream source subject transforms
(requires companion jsm.go changes)
@@ -1008,6 +1008,10 @@ func (c *streamCmd) reportAction(_ *fisk.ParseContext) error {
if source.FilterSubject != "" {
edge.Label(source.FilterSubject)
}
+ if source.SubjectTransform != "" {
+ edge2 := dg.Edge(snode, node).Attr("color", "red")
+ edge2.Label(source.SubjectTransform)
+ }
}
}
@@ -1593,7 +1597,13 @@ func (c *streamCmd) renderSource(s *api.StreamSource) string {
parts = append(parts, fmt.Sprintf("Start Time: %v", s.OptStartTime))
}
if s.FilterSubject != "" {
- parts = append(parts, fmt.Sprintf("Subject: %s", s.FilterSubject))
+ parts = append(parts, fmt.Sprintf("Subject filter: %s", s.FilterSubject))
+ }
+ if s.SubjectTransform != "" {
+ if s.FilterSubject == "" {
+ parts = append(parts, fmt.Sprintf("Subject filter: %s", ">"))
+ }
+ parts = append(parts, fmt.Sprintf("Subject transform: %s", s.SubjectTransform))
}
if s.External != nil {
if s.External.ApiPrefix != "" {
@@ -1674,6 +1684,20 @@ func (c *streamCmd) showStreamInfo(info *api.StreamInfo) {
showSource := func(s *api.StreamSourceInfo) {
fmt.Printf(" Stream Name: %s\n", s.Name)
+
+ if s.SubjectTransform != "" {
+ if s.FilterSubject == "" {
+ fmt.Printf(" Subject Filter: %s\n", ">")
+ } else {
+ fmt.Printf(" Subject Filter: %s\n", s.FilterSubject)
+ }
+ fmt.Printf(" Subject Transform: %s \n", s.SubjectTransform)
+ } else {
+ if s.FilterSubject != "" {
+ fmt.Printf(" Subject Filter: %s\n", s.FilterSubject)
+ }
+ }
+
fmt.Printf(" Lag: %s\n", humanize.Comma(int64(s.Lag)))
if s.Active > 0 && s.Active < math.MaxInt64 {
fmt.Printf(" Last Seen: %v\n", humanizeDuration(s.Active))
@@ -2225,6 +2249,12 @@ func (c *streamCmd) askSource(name string, prefix string) *api.StreamSource {
}, &cfg.FilterSubject)
fisk.FatalIfError(err, "could not request filter")
+ err = askOne(&survey.Input{
+ Message: fmt.Sprintf("%s Subject mapping transform", prefix),
+ Help: "Map matching subjects according to this destination transform",
+ }, &cfg.SubjectTransform)
+ fisk.FatalIfError(err, "could not request subject mapping destination transform")
+
ok, err = askConfirmation(fmt.Sprintf("Import %q from a different JetStream domain", name), false)
fisk.FatalIfError(err, "Could not request source details")
if ok {
