From 7f897d308f2428f5d564e809dc3f9520cccb0e20 Mon Sep 17 00:00:00 2001 From: Aditya Agarwal Date: Tue, 24 Dec 2024 15:37:54 +0530 Subject: [PATCH 1/9] Support For Custom Subject Mapping --- server/subject_transform.go | 43 ++++++++++++++++++++++++++++---- server/subject_transform_test.go | 12 +++++++++ 2 files changed, 50 insertions(+), 5 deletions(-) diff --git a/server/subject_transform.go b/server/subject_transform.go index 8292d48732a..6de6928436e 100644 --- a/server/subject_transform.go +++ b/server/subject_transform.go @@ -1,4 +1,3 @@ -// Copyright 2023 The NATS Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at @@ -21,6 +20,17 @@ import ( "strings" ) + +var customMappingFunction func(token string) string + +func RegisterCustomMappingFunction(f func(token string) string) error{ + if customMappingFunction == nil { + customMappingFunction = f + return nil + } + return fmt.Errorf("custom mapping function already registered") +} + // Subject mapping and transform setups. var ( commaSeparatorRegEx = regexp.MustCompile(`,\s*`) @@ -33,6 +43,7 @@ var ( splitMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit\s*\((.*)\)\s*}}`) leftMappingFunctionRegEx = regexp.MustCompile(`{{\s*[lL]eft\s*\((.*)\)\s*}}`) rightMappingFunctionRegEx = regexp.MustCompile(`{{\s*[rR]ight\s*\((.*)\)\s*}}`) + customtMappingFunctionRegEx = regexp.MustCompile(`{{\s*[cC]ustom\s*\((.*)\)\s*}}`) ) // Enum for the subject mapping subjectTransform function types @@ -48,6 +59,7 @@ const ( Split Left Right + Custom ) // Transforms for arbitrarily mapping subjects from one to another for maps, tees and filters. @@ -116,9 +128,8 @@ func NewSubjectTransformWithStrict(src, dest string, strict bool) (*subjectTrans if err != nil { return nil, err } - if strict { - if tranformType != NoTransform && tranformType != Wildcard { + if tranformType != NoTransform && tranformType != Wildcard && tranformType != Custom { return nil, &mappingDestinationErr{token, ErrMappingDestinationNotSupportedForImport} } } @@ -313,6 +324,23 @@ func indexPlaceHolders(token string) (int16, []int, int32, string, error) { return transformIndexIntArgsHelper(token, args, Left) } + // Custom(token) + args = getMappingFunctionArgs(customtMappingFunctionRegEx, token) + if args != nil { + if len(args) == 1 && args[0] == _EMPTY_ { + return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrMappingDestinationNotEnoughArgs} + } + if len(args) == 1 { + tokenIndex, err := strconv.Atoi(strings.Trim(args[0], " ")) + if err != nil { + return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrMappingDestinationInvalidArg} + } + return Custom, []int{tokenIndex}, -1, _EMPTY_, nil + } else { + return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrMappingDestinationTooManyArgs} + } + } + // split(token, deliminator) args = getMappingFunctionArgs(splitMappingFunctionRegEx, token) if args != nil { @@ -458,8 +486,13 @@ func (tr *subjectTransform) TransformTokenizedSubject(tokens []string) string { keyForHashing = append(keyForHashing, []byte(tokens[sourceToken])...) } b.WriteString(tr.getHashPartition(keyForHashing, int(tr.dtokmfintargs[i]))) + case Custom: + if(customMappingFunction == nil) { + b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]]) + } + b.WriteString(customMappingFunction(tokens[tr.dtokmftokindexesargs[i][0]])) case Wildcard: // simple substitution - b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]]) + b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]]) case SplitFromLeft: sourceToken := tokens[tr.dtokmftokindexesargs[i][0]] sourceTokenLen := len(sourceToken) @@ -620,4 +653,4 @@ func subjectInfo(subject string) (bool, []string, int, bool) { } } return true, tokens, npwcs, sfwc -} +} \ No newline at end of file diff --git a/server/subject_transform_test.go b/server/subject_transform_test.go index dbbfdc28a1d..0afb4968a8e 100644 --- a/server/subject_transform_test.go +++ b/server/subject_transform_test.go @@ -133,6 +133,13 @@ func TestSubjectTransformHelpers(t *testing.T) { } func TestSubjectTransforms(t *testing.T) { + + var function = func (value string) string { + return "custom" + value + } + + RegisterCustomMappingFunction(function) + shouldErr := func(src, dest string, strict bool) { t.Helper() if _, err := NewSubjectTransformWithStrict(src, dest, strict); err != ErrBadSubject && !errors.Is(err, ErrInvalidMappingDestination) { @@ -185,6 +192,9 @@ func TestSubjectTransforms(t *testing.T) { shouldBeOK("*.*", "{{partition(10,1,2)}}", false) shouldBeOK("foo.*.*", "foo.{{wildcard(1)}}.{{wildcard(2)}}.{{partition(5,1,2)}}", false) + shouldBeOK("foo.*", "bar.{{custom(1)}}", false) + shouldBeOK("foo.*", "bar.{{custom(1)}}", true) + shouldMatch := func(src, dest, sample, expected string) { t.Helper() tr := shouldBeOK(src, dest, false) @@ -220,4 +230,6 @@ func TestSubjectTransforms(t *testing.T) { shouldMatch("*", "{{left(1,1)}}", "1234", "1") shouldMatch("*", "{{left(1,3)}}", "1234", "123") shouldMatch("*", "{{left(1,6)}}", "1234", "1234") + shouldMatch("foo.*", "bar.{{custom(1)}}", "foo.hello","bar.customhello") + shouldMatch("foo.*.*.bar", "bar.{{custom(2)}}.{{custom(1)}}", "foo.1.2.bar","bar.custom2.") } From 1ac74f38b58bfdaac1fcf49938afa10c225cb210 Mon Sep 17 00:00:00 2001 From: Aditya Agarwal Date: Tue, 24 Dec 2024 15:52:32 +0530 Subject: [PATCH 2/9] Fmt --- server/subject_transform.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/server/subject_transform.go b/server/subject_transform.go index 6de6928436e..dc87baa515b 100644 --- a/server/subject_transform.go +++ b/server/subject_transform.go @@ -1,3 +1,4 @@ +// Copyright 2023 The NATS Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at @@ -128,6 +129,7 @@ func NewSubjectTransformWithStrict(src, dest string, strict bool) (*subjectTrans if err != nil { return nil, err } + if strict { if tranformType != NoTransform && tranformType != Wildcard && tranformType != Custom { return nil, &mappingDestinationErr{token, ErrMappingDestinationNotSupportedForImport} @@ -492,7 +494,7 @@ func (tr *subjectTransform) TransformTokenizedSubject(tokens []string) string { } b.WriteString(customMappingFunction(tokens[tr.dtokmftokindexesargs[i][0]])) case Wildcard: // simple substitution - b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]]) + b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]]) case SplitFromLeft: sourceToken := tokens[tr.dtokmftokindexesargs[i][0]] sourceTokenLen := len(sourceToken) From 3d9b570b58b057b448399ad3e0f3481a91f6c4fb Mon Sep 17 00:00:00 2001 From: Aditya Agarwal Date: Tue, 24 Dec 2024 15:54:21 +0530 Subject: [PATCH 3/9] Fmt --- server/subject_transform.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/subject_transform.go b/server/subject_transform.go index dc87baa515b..ac8ae3a81e6 100644 --- a/server/subject_transform.go +++ b/server/subject_transform.go @@ -494,7 +494,7 @@ func (tr *subjectTransform) TransformTokenizedSubject(tokens []string) string { } b.WriteString(customMappingFunction(tokens[tr.dtokmftokindexesargs[i][0]])) case Wildcard: // simple substitution - b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]]) + b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]]) case SplitFromLeft: sourceToken := tokens[tr.dtokmftokindexesargs[i][0]] sourceTokenLen := len(sourceToken) From 7e90c8daeb42136838889c6f50e1f944b5b07703 Mon Sep 17 00:00:00 2001 From: Aditya Agarwal Date: Tue, 24 Dec 2024 15:55:17 +0530 Subject: [PATCH 4/9] Fmt --- server/subject_transform.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/subject_transform.go b/server/subject_transform.go index ac8ae3a81e6..7b0bcd9ceac 100644 --- a/server/subject_transform.go +++ b/server/subject_transform.go @@ -494,7 +494,7 @@ func (tr *subjectTransform) TransformTokenizedSubject(tokens []string) string { } b.WriteString(customMappingFunction(tokens[tr.dtokmftokindexesargs[i][0]])) case Wildcard: // simple substitution - b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]]) + b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]]) case SplitFromLeft: sourceToken := tokens[tr.dtokmftokindexesargs[i][0]] sourceTokenLen := len(sourceToken) From eab1c4b23f80d1a01ec64c19020530dbeb862060 Mon Sep 17 00:00:00 2001 From: Aditya Agarwal Date: Tue, 24 Dec 2024 15:57:27 +0530 Subject: [PATCH 5/9] Fixed Test --- server/subject_transform_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/subject_transform_test.go b/server/subject_transform_test.go index 0afb4968a8e..febd9d72379 100644 --- a/server/subject_transform_test.go +++ b/server/subject_transform_test.go @@ -231,5 +231,5 @@ func TestSubjectTransforms(t *testing.T) { shouldMatch("*", "{{left(1,3)}}", "1234", "123") shouldMatch("*", "{{left(1,6)}}", "1234", "1234") shouldMatch("foo.*", "bar.{{custom(1)}}", "foo.hello","bar.customhello") - shouldMatch("foo.*.*.bar", "bar.{{custom(2)}}.{{custom(1)}}", "foo.1.2.bar","bar.custom2.") + shouldMatch("foo.*.*.bar", "bar.{{custom(2)}}.{{custom(1)}}", "foo.1.2.bar","bar.custom2.custom1") } From 54059802977097501c75b3644733add8643061a5 Mon Sep 17 00:00:00 2001 From: Aditya Agarwal Date: Tue, 24 Dec 2024 16:22:29 +0530 Subject: [PATCH 6/9] Added More Test --- server/subject_transform_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/server/subject_transform_test.go b/server/subject_transform_test.go index febd9d72379..b28178fbb72 100644 --- a/server/subject_transform_test.go +++ b/server/subject_transform_test.go @@ -232,4 +232,5 @@ func TestSubjectTransforms(t *testing.T) { shouldMatch("*", "{{left(1,6)}}", "1234", "1234") shouldMatch("foo.*", "bar.{{custom(1)}}", "foo.hello","bar.customhello") shouldMatch("foo.*.*.bar", "bar.{{custom(2)}}.{{custom(1)}}", "foo.1.2.bar","bar.custom2.custom1") + shouldMatch("user.details.*", "user.details.{{custom(1)}}", "user.details.11","user.details.custom11") } From 110db0363deb216526e05dd79b0d28088206b624 Mon Sep 17 00:00:00 2001 From: Aditya Agarwal Date: Thu, 2 Jan 2025 17:09:57 +0530 Subject: [PATCH 7/9] Updating the Unknown Mapping Issue --- server/sublist.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/server/sublist.go b/server/sublist.go index d34bcde1ad1..5fc39577082 100644 --- a/server/sublist.go +++ b/server/sublist.go @@ -1292,7 +1292,8 @@ func ValidateMapping(src string, dest string) error { !splitFromRightMappingFunctionRegEx.MatchString(t) && !sliceFromLeftMappingFunctionRegEx.MatchString(t) && !sliceFromRightMappingFunctionRegEx.MatchString(t) && - !splitMappingFunctionRegEx.MatchString(t) { + !splitMappingFunctionRegEx.MatchString(t) && + !customtMappingFunctionRegEx.MatchString(t) { return &mappingDestinationErr{t, ErrUnknownMappingDestinationFunction} } else { continue From 17aed435b01e73d3bf1284e1e567b6417d861ffb Mon Sep 17 00:00:00 2001 From: Aditya Agarwal Date: Mon, 13 Jan 2025 11:32:20 +0530 Subject: [PATCH 8/9] Updating to add mapping during the import call --- server/client.go | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/server/client.go b/server/client.go index b5fc75a8380..c296161ea6e 100644 --- a/server/client.go +++ b/server/client.go @@ -600,6 +600,7 @@ type subscription struct { qw int32 closed int32 mqtt *mqttSub + mappedSub []byte } // Indicate that this subscription is closed. @@ -2902,6 +2903,13 @@ func (c *client) addShadowSubscriptions(acc *Account, sub *subscription, enact b // exact match or a superset match we should use the from field from // the import. If we are a subset or overlap, we have to dynamically calculate // the subject. On overlap, ime requires the overlap subject. + if (c.kind == CLIENT || c.kind == LEAF) && c.in.flags.isSet(hasMappings) { + mappedSub, _ := c.acc.selectMappedSubject(string(sub.subject)) + sub.mappedSub = sub.subject + sub.subject = []byte(mappedSub) + subj =mappedSub + } + for _, im := range acc.imports.streams { if im.invalid { continue @@ -2991,7 +2999,6 @@ func (c *client) addShadowSub(sub *subscription, ime *ime, enact bool) (*subscri im := ime.im nsub.im = im - if !im.usePub && ime.dyn && im.tr != nil { if im.rtr == nil { im.rtr = im.tr.reverse() @@ -4703,6 +4710,9 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver, dsubj = append(_dsubj[:0], sub.im.to...) } + if(sub.mappedSub != nil){ + dsubj = append(_dsubj[:0], sub.mappedSub...) + } if mt != nil { mt.addStreamExportEvent(sub.client, dsubj) // If allow_trace is false... From b8a4b7a7b3865fea381758f41c3293c49390414f Mon Sep 17 00:00:00 2001 From: Aditya Agarwal Date: Tue, 14 Jan 2025 11:43:11 +0530 Subject: [PATCH 9/9] Minor --- server/client.go | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/server/client.go b/server/client.go index c296161ea6e..832e7c8b99d 100644 --- a/server/client.go +++ b/server/client.go @@ -2899,17 +2899,16 @@ func (c *client) addShadowSubscriptions(acc *Account, sub *subscription, enact b hasWC = true } } - // Loop over the import subjects. We have 4 scenarios. If we have an - // exact match or a superset match we should use the from field from - // the import. If we are a subset or overlap, we have to dynamically calculate - // the subject. On overlap, ime requires the overlap subject. if (c.kind == CLIENT || c.kind == LEAF) && c.in.flags.isSet(hasMappings) { mappedSub, _ := c.acc.selectMappedSubject(string(sub.subject)) sub.mappedSub = sub.subject sub.subject = []byte(mappedSub) subj =mappedSub } - + // Loop over the import subjects. We have 4 scenarios. If we have an + // exact match or a superset match we should use the from field from + // the import. If we are a subset or overlap, we have to dynamically calculate + // the subject. On overlap, ime requires the overlap subject. for _, im := range acc.imports.streams { if im.invalid { continue