[Adding] new field to rename the remote subject and deprecate To

v2/imports.go

@@ -32,14 +32,21 @@ type Import struct {
 	Subject Subject `json:"subject,omitempty"`
 	Account string  `json:"account,omitempty"`
 	Token   string  `json:"token,omitempty"`
+	// Deprecated: use LocalSubject instead
 	// To field in an import is always from the perspective of the subscriber
 	// in the case of a stream it is the client of the stream (the importer),
 	// from the perspective of a service, it is the subscription waiting for
 	// requests (the exporter). If the field is empty, it will default to the
 	// value in the Subject field.
-	To    Subject    `json:"to,omitempty"`
-	Type  ExportType `json:"type,omitempty"`
-	Share bool       `json:"share,omitempty"`
+	To Subject `json:"to,omitempty"`
+	// Local subject used to subscribe (for streams) and publish (for services) to.
+	// This value only needs setting if you want to change the value of Subject.
+	// If the value of Subject ends in > then LocalSubject needs to end in > as well.
+	// LocalSubject can contain $<number> wildcard references where number references the nth wildcard in Subject.
+	// The sum of wildcard reference and * tokens needs to match the number of * token in Subject.
+	LocalSubject RenamingSubject `json:"local_subject,omitempty"`
+	Type         ExportType      `json:"type,omitempty"`
+	Share        bool            `json:"share,omitempty"`
 }
 
 // IsService returns true if the import is of type service
@@ -67,6 +74,12 @@ func (i *Import) Validate(actPubKey string, vr *ValidationResults) {
 	}
 
 	i.Subject.Validate(vr)
+	if i.LocalSubject != "" {
+		i.LocalSubject.Validate(i.Subject, vr)
+		if i.To != "" {
+			vr.AddError("Local Subject replaces To")
+		}
+	}
 
 	if i.Share && !i.IsService() {
 		vr.AddError("sharing information (for latency tracking) is only valid for services: %q", i.Subject)
@@ -120,17 +133,29 @@ type Imports []*Import
 
 // Validate checks if an import is valid for the wrapping account
 func (i *Imports) Validate(acctPubKey string, vr *ValidationResults) {
-	toSet := make(map[Subject]bool, len(*i))
+	toSet := make(map[Subject]struct{}, len(*i))
 	for _, v := range *i {
 		if v == nil {
 			vr.AddError("null import is not allowed")
 			continue
 		}
 		if v.Type == Service {
-			if _, ok := toSet[v.To]; ok {
-				vr.AddError("Duplicate To subjects for %q", v.To)
+			sub := v.To
+			if sub == "" {
+				sub = v.LocalSubject.ToSubject()
+			}
+			if sub == "" {
+				sub = v.Subject
+			}
+			for k := range toSet {
+				if sub.IsContainedIn(k) || k.IsContainedIn(sub) {
+					vr.AddError("overlapping subject namespace for %q and %q", sub, k)
+				}
+			}
+			if _, ok := toSet[sub]; ok {
+				vr.AddError("overlapping subject namespace for %q", v.To)
 			}
-			toSet[v.To] = true
+			toSet[sub] = struct{}{}
 		}
 		v.Validate(acctPubKey, vr)
 	}

v2/imports_test.go

@@ -20,6 +20,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"sort"
+	"strings"
 	"testing"
 	"time"
 )
@@ -335,6 +336,54 @@ func TestImportsValidation(t *testing.T) {
 	}
 }
 
+func TestImportsLocalSubjectExclusiveTo(t *testing.T) {
+	ak := createAccountNKey(t)
+	akp := publicKey(ak, t)
+	i := &Import{Subject: "foo", Account: akp, LocalSubject: "bar", Type: Stream}
+	i2 := &Import{Subject: "foo", Account: akp, LocalSubject: "bar", Type: Service}
+
+	imports := &Imports{}
+	imports.Add(i, i2)
+
+	vr := CreateValidationResults()
+	imports.Validate("", vr)
+
+	if !vr.IsEmpty() {
+		t.Errorf("no issues expected")
+	}
+
+	i.To = "bar"
+	i2.To = "bar"
+	imports = &Imports{}
+	imports.Add(i, i2)
+
+	vr = CreateValidationResults()
+	imports.Validate("", vr)
+
+	if vr.IsEmpty() {
+		t.Errorf("issues expected")
+	}
+	if !vr.IsBlocking(false) {
+		t.Errorf("issues expected to be blocking")
+	}
+}
+
+func TestImportsLocalSubjectVariants(t *testing.T) {
+	ak := createAccountNKey(t)
+	akp := publicKey(ak, t)
+	imports := &Imports{}
+	imports.Add(
+		&Import{Subject: "foo.*.bar.*.>", Account: akp, LocalSubject: "my.$2.$1.>", Type: Stream},
+		&Import{Subject: "baz.*.bar.*.>", Account: akp, LocalSubject: "bar.*.*.>", Type: Service},
+		&Import{Subject: "baz.*", Account: akp, LocalSubject: "my.$1", Type: Stream},
+		&Import{Subject: "bar.*", Account: akp, LocalSubject: "baz.*", Type: Service})
+	vr := CreateValidationResults()
+	imports.Validate("", vr)
+	if !vr.IsEmpty() {
+		t.Errorf("no issues expected")
+	}
+}
+
 func TestTokenURLImportValidation(t *testing.T) {
 	ak := createAccountNKey(t)
 	ak2 := createAccountNKey(t)
@@ -484,7 +533,6 @@ func TestWildcard(t *testing.T) {
 	if vr.IsBlocking(true) {
 		t.Fatalf("Expected no blocking validation errors")
 	}
-
 }
 
 func TestImport_Sorting(t *testing.T) {
@@ -501,3 +549,21 @@ func TestImport_Sorting(t *testing.T) {
 		t.Fatal("imports not sorted")
 	}
 }
+
+func TestImports_Validate(t *testing.T) {
+	var imports Imports
+	pk := publicKey(createAccountNKey(t), t)
+	imports.Add(&Import{Subject: "x", LocalSubject: "foo", Type: Service, Account: pk})
+	imports.Add(&Import{Subject: "z.*", LocalSubject: "*", Type: Service, Account: pk})
+	imports.Add(&Import{Subject: "y.>", LocalSubject: ">", Type: Service, Account: pk})
+	vr := ValidationResults{}
+	imports.Validate("", &vr)
+	if len(vr.Issues) != 3 || !vr.IsBlocking(false) {
+		t.Fatal("expected 3 blocking issues")
+	}
+	for _, v := range vr.Issues {
+		if !strings.HasPrefix(v.Description, "overlapping subject namespace") {
+			t.Fatalf("Expected every error to contain: overlapping subject namespace")
+		}
+	}
+}

v2/types.go

@@ -21,6 +21,7 @@ import (
 	"net"
 	"net/url"
 	"reflect"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -101,6 +102,73 @@ func (t *ExportType) UnmarshalJSON(b []byte) error {
 	return fmt.Errorf("unknown export type %q", j)
 }
 
+type RenamingSubject Subject
+
+func (s RenamingSubject) Validate(from Subject, vr *ValidationResults) {
+	v := Subject(s)
+	v.Validate(vr)
+	if from == "" {
+		vr.AddError("subject cannot be empty")
+	}
+	if strings.Contains(string(s), " ") {
+		vr.AddError("subject %q cannot have spaces", v)
+	}
+	matchesSuffix := func(s Subject) bool {
+		return s == ">" || strings.HasSuffix(string(s), ".>")
+	}
+	if matchesSuffix(v) != matchesSuffix(from) {
+		vr.AddError("both, renaming subject and subject, need to end or not end in >")
+	}
+	fromCnt := from.countTokenWildcards()
+	refCnt := 0
+	for _, tk := range strings.Split(string(v), ".") {
+		if tk == "*" {
+			refCnt++
+		}
+		if len(tk) < 2 {
+			continue
+		}
+		if tk[0] == '$' {
+			if idx, err := strconv.Atoi(tk[1:]); err == nil {
+				if idx > fromCnt {
+					vr.AddError("Reference $%d in %q reference * in %q that do not exist", idx, s, from)
+				} else {
+					refCnt++
+				}
+			}
+		}
+	}
+	if refCnt != fromCnt {
+		vr.AddError("subject does not contain enough * or reference wildcards $[0-9]")
+	}
+}
+
+// Replaces reference tokens with *
+func (s RenamingSubject) ToSubject() Subject {
+	if !strings.Contains(string(s), "$") {
+		return Subject(s)
+	}
+	bldr := strings.Builder{}
+	tokens := strings.Split(string(s), ".")
+	for i, tk := range tokens {
+		convert := false
+		if len(tk) > 1 && tk[0] == '$' {
+			if _, err := strconv.Atoi(tk[1:]); err == nil {
+				convert = true
+			}
+		}
+		if convert {
+			bldr.WriteString("*")
+		} else {
+			bldr.WriteString(tk)
+		}
+		if i != len(tokens)-1 {
+			bldr.WriteString(".")
+		}
+	}
+	return Subject(bldr.String())
+}
+
 // Subject is a string that represents a NATS subject
 type Subject string
 
@@ -115,6 +183,21 @@ func (s Subject) Validate(vr *ValidationResults) {
 	}
 }
 
+func (s Subject) countTokenWildcards() int {
+	v := string(s)
+	if v == "*" {
+		return 1
+	}
+	cnt := strings.Count(v, ".*.")
+	if strings.HasSuffix(v, ".*") {
+		cnt++
+	}
+	if strings.HasPrefix(v, "*.") {
+		cnt++
+	}
+	return cnt
+}
+
 // HasWildCards is used to check if a subject contains a > or *
 func (s Subject) HasWildCards() bool {
 	v := string(s)

v2/types_test.go

@@ -234,9 +234,17 @@ func TestSubjectContainment(t *testing.T) {
 	var o Subject
 
 	s = "one.two.three"
-	o = "one.two.three"
+	o = "one.*.three"
 	AssertEquals(true, s.IsContainedIn(o), t)
 
+	s = "one.*.three"
+	o = "one.*.three"
+	AssertEquals(true, s.IsContainedIn(o), t)
+
+	s = "one.*.three"
+	o = "one.two.three"
+	AssertEquals(false, s.IsContainedIn(o), t)
+
 	s = "one.two.three"
 	o = "one.two.*"
 	AssertEquals(true, s.IsContainedIn(o), t)
@@ -270,6 +278,45 @@ func TestSubjectContainment(t *testing.T) {
 	AssertEquals(false, s.IsContainedIn(o), t)
 }
 
+func TestRenamingSubject_ToSubject(t *testing.T) {
+	AssertEquals(RenamingSubject("foo.$2.$1.bar").ToSubject(), Subject("foo.*.*.bar"), t)
+	AssertEquals(RenamingSubject("foo.*.bar").ToSubject(), Subject("foo.*.bar"), t)
+	AssertEquals(RenamingSubject("foo.$2.*.bar").ToSubject(), Subject("foo.*.*.bar"), t)
+}
+
+func TestRenamigSubject_Validate(t *testing.T) {
+	for from, to := range map[string]string{
+		"foo":">",
+		"bar":"*",
+		"foo.*":"*.*",
+		"foo.>":"*.*",
+		"bar.>":"*.>",
+		"bar.*.*>":"*.>",
+		"*.bar":"$2",
+	} {
+		vr := ValidationResults{}
+		RenamingSubject(to).Validate(Subject(from), &vr)
+		if !vr.IsBlocking(false) {
+			t.Fatalf("expected blocking issue %q:%q", to, from)
+		}
+	}
+	for from, to := range map[string]string{
+		"foo":"bar",
+		"foo.bar":"baz",
+		"x":"x.y.z",
+		">":"foo.>",
+		"*":"$1.foo",
+		"*.*":"$1.foo.$2",
+		"*.bar":"$1",
+	} {
+		vr := ValidationResults{}
+		RenamingSubject(to).Validate(Subject(from), &vr)
+		if !vr.IsEmpty() {
+			t.Fatalf("expected no issue %q:%q got: %v", to, from, vr.Issues)
+		}
+	}
+}
+
 func TestInvalidInfo(t *testing.T) {
 	for _, info := range []Info{{
 		Description: "",