Repository: nats-io/nats-server
Pull request #5337 (2024-04-23 23:50:39)

Added Left and Right Mapping Operations
<!-- Please make sure to read CONTRIBUTING.md, then delete this notice and replace it with your PR description. The below sign-off certifies that the contribution is your original work and that you license the work to the project under the Apache-2.0 license. We cannot accept contributions without it. -->
Adding Left and Right operations to Mappings enabling the ability to group by token beginnings and endings.  

Example: 
foo.*: foo.{{Right(1,2)}} -> foo.1234 -> foo.34
foo.*:foo.{{Left(1,2)}} -> foo.1234:  -> foo.12

Signed-off-by: Stephen Spates <sspates@starbucks.com>



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 650af5f2..fe994d71 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2257,6 +2257,7 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 		sm = new(StoreMsg)
 	}
 
+	var result SublistResult
 	for seq := start; seq <= lseq; seq++ {
 		llseq := mb.llseq
 		fsm, err := mb.cacheLookup(seq, sm)
@@ -2265,7 +2266,7 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 		}
 		expireOk := seq == lseq && mb.llseq == seq
 
-		if sl.HasInterest(fsm.subj) {
+		if r := sl.MatchWithResult(fsm.subj, &result); len(r.psubs) > 0 {
 			return fsm, expireOk, nil
 		}
 		// If we are here we did not match, so put the llseq back.
diff --git a/server/memstore.go b/server/memstore.go
index 723520fa..7a81847d 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -1055,12 +1055,13 @@ func (ms *memStore) LoadNextMsgMulti(sl *Sublist, start uint64, smp *StoreMsg) (
 	// Initial setup.
 	fseq, lseq := start, ms.state.LastSeq
 
+	var result SublistResult
 	for nseq := fseq; nseq <= lseq; nseq++ {
 		sm, ok := ms.msgs[nseq]
 		if !ok {
 			continue
 		}
-		if sl.HasInterest(sm.subj) {
+		if r := sl.MatchWithResult(sm.subj, &result); len(r.psubs) > 0 {
 			if smp == nil {
 				smp = new(StoreMsg)
 			}
diff --git a/server/subject_transform.go b/server/subject_transform.go
index e7d72576..8292d487 100644
--- a/server/subject_transform.go
+++ b/server/subject_transform.go
@@ -31,6 +31,8 @@ var (
 	sliceFromLeftMappingFunctionRegEx  = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[lL]eft\s*\((.*)\)\s*}}`)
 	sliceFromRightMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[rR]ight\s*\((.*)\)\s*}}`)
 	splitMappingFunctionRegEx          = regexp.MustCompile(`{{\s*[sS]plit\s*\((.*)\)\s*}}`)
+	leftMappingFunctionRegEx           = regexp.MustCompile(`{{\s*[lL]eft\s*\((.*)\)\s*}}`)
+	rightMappingFunctionRegEx          = regexp.MustCompile(`{{\s*[rR]ight\s*\((.*)\)\s*}}`)
 )
 
 // Enum for the subject mapping subjectTransform function types
@@ -44,6 +46,8 @@ const (
 	SliceFromLeft
 	SliceFromRight
 	Split
+	Left
+	Right
 )
 
 // Transforms for arbitrarily mapping subjects from one to another for maps, tees and filters.
@@ -297,6 +301,18 @@ func indexPlaceHolders(token string) (int16, []int, int32, string, error) {
 				return transformIndexIntArgsHelper(token, args, SliceFromRight)
 			}
 
+			// Right(token, length)
+			args = getMappingFunctionArgs(rightMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, Right)
+			}
+
+			// Left(token, length)
+			args = getMappingFunctionArgs(leftMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, Left)
+			}
+
 			// split(token, deliminator)
 			args = getMappingFunctionArgs(splitMappingFunctionRegEx, token)
 			if args != nil {
@@ -515,6 +531,24 @@ func (tr *subjectTransform) TransformTokenizedSubject(tokens []string) string {
 						b.WriteString(tsep)
 					}
 				}
+			case Left:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				sliceSize := int(tr.dtokmfintargs[i])
+				if sliceSize > 0 && sliceSize < sourceTokenLen {
+					b.WriteString(sourceToken[0:sliceSize])
+				} else { // too small to slice at the requested size: don't slice
+					b.WriteString(sourceToken)
+				}
+			case Right:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				sliceSize := int(tr.dtokmfintargs[i])
+				if sliceSize > 0 && sliceSize < sourceTokenLen {
+					b.WriteString(sourceToken[sourceTokenLen-sliceSize : sourceTokenLen])
+				} else { // too small to slice at the requested size: don't slice
+					b.WriteString(sourceToken)
+				}
 			}
 		}
 
diff --git a/server/sublist.go b/server/sublist.go
index 71c256e4..8d5859f4 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -527,27 +527,30 @@ var emptyResult = &SublistResult{}
 // Match will match all entries to the literal subject.
 // It will return a set of results for both normal and queue subscribers.
 func (s *Sublist) Match(subject string) *SublistResult {
-	return s.match(subject, true)
+	return s.match(subject, true, nil)
 }
 
-// HasInterest will return whether or not there is any interest in the subject.
-// In cases where more detail is not required, this may be faster than Match.
-func (s *Sublist) HasInterest(subject string) bool {
-	return s.hasInterest(subject, true)
+// MatchWithResult will match all entries to the literal subject, reusing the
+// supplied SublistResult to reduce allocations on hot paths.
+// It will return a set of results for both normal and queue subscribers.
+func (s *Sublist) MatchWithResult(subject string, result *SublistResult) *SublistResult {
+	return s.match(subject, true, result)
 }
 
 func (s *Sublist) matchNoLock(subject string) *SublistResult {
-	return s.match(subject, false)
+	return s.match(subject, false, nil)
 }
 
-func (s *Sublist) match(subject string, doLock bool) *SublistResult {
+func (s *Sublist) match(subject string, doLock bool, result *SublistResult) *SublistResult {
 	atomic.AddUint64(&s.matches, 1)
 
 	// Check cache first.
 	if doLock {
 		s.RLock()
 	}
-	cacheEnabled := s.cache != nil
+	// Writing to the cache is only allowed if not supplying our
+	// own SublistResult, i.e. via call to MatchWithResult.
+	cacheEnabled := result == nil && s.cache != nil
 	r, ok := s.cache[subject]
 	if doLock {
 		s.RUnlock()
@@ -575,7 +578,11 @@ func (s *Sublist) match(subject string, doLock bool) *SublistResult {
 	tokens = append(tokens, subject[start:])
 
 	// FIXME(dlc) - Make shared pool between sublist and client readLoop?
-	result := &SublistResult{}
+	if result == nil {
+		result = &SublistResult{}
+	}
+	result.psubs = result.psubs[:0]
+	result.qsubs = result.qsubs[:0]
 
 	// Get result from the main structure and place into the shared cache.
 	// Hold the read lock to avoid race between match and store.
@@ -614,47 +621,6 @@ func (s *Sublist) match(subject string, doLock bool) *SublistResult {
 	return result
 }
 
-func (s *Sublist) hasInterest(subject string, doLock bool) bool {
-	// Check cache first.
-	if doLock {
-		s.RLock()
-	}
-	var matched bool
-	if s.cache != nil {
-		_, matched = s.cache[subject]
-	}
-	if doLock {
-		s.RUnlock()
-	}
-	if matched {
-		atomic.AddUint64(&s.cacheHits, 1)
-		return true
-	}
-
-	tsa := [32]string{}
-	tokens := tsa[:0]
-	start := 0
-	for i := 0; i < len(subject); i++ {
-		if subject[i] == btsep {
-			if i-start == 0 {
-				return false
-			}
-			tokens = append(tokens, subject[start:i])
-			start = i + 1
-		}
-	}
-	if start >= len(subject) {
-		return false
-	}
-	tokens = append(tokens, subject[start:])
-
-	if doLock {
-		s.RLock()
-		defer s.RUnlock()
-	}
-	return matchLevelForAny(s.root, tokens)
-}
-
 // Remove entries in the cache until we are under the maximum.
 // TODO(dlc) this could be smarter now that its not inline.
 func (s *Sublist) reduceCacheCount() {
@@ -767,36 +733,6 @@ func matchLevel(l *level, toks []string, results *SublistResult) {
 	}
 }
 
-func matchLevelForAny(l *level, toks []string) bool {
-	var pwc, n *node
-	for i, t := range toks {
-		if l == nil {
-			return false
-		}
-		if l.fwc != nil {
-			return true
-		}
-		if pwc = l.pwc; pwc != nil {
-			if match := matchLevelForAny(pwc.next, toks[i+1:]); match {
-				return true
-			}
-		}
-		n = l.nodes[t]
-		if n != nil {
-			l = n.next
-		} else {
-			l = nil
-		}
-	}
-	if n != nil {
-		return true
-	}
-	if pwc != nil {
-		return true
-	}
-	return false
-}
-
 // lnt is used to track descent into levels for a removal for pruning.
 type lnt struct {
 	l *level


Test patch:
diff --git a/server/subject_transform_test.go b/server/subject_transform_test.go
index 68d18328..dbbfdc28 100644
--- a/server/subject_transform_test.go
+++ b/server/subject_transform_test.go
@@ -71,6 +71,20 @@ func TestPlaceHolderIndex(t *testing.T) {
 	if err != nil || transformType != SliceFromLeft || len(indexes) != 1 || indexes[0] != 2 || sliceSize != 2 {
 		t.Fatalf("Error parsing %s", testString)
 	}
+
+	testString = "{{Left(3,2)}}"
+	transformType, indexes, position, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != Left || len(indexes) != 1 || indexes[0] != 3 || position != 2 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	testString = "{{Right(3,2)}}"
+	transformType, indexes, position, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != Right || len(indexes) != 1 || indexes[0] != 3 || position != 2 {
+		t.Fatalf("Error parsing %s", testString)
+	}
 }
 
 func TestSubjectTransformHelpers(t *testing.T) {
@@ -200,4 +214,10 @@ func TestSubjectTransforms(t *testing.T) {
 	shouldMatch("*", "{{split(1,-)}}", "-abc-def--ghi-", "abc.def.ghi")
 	shouldMatch("*", "{{split(1,-)}}", "abc-def--ghi-", "abc.def.ghi")
 	shouldMatch("*.*", "{{split(2,-)}}.{{splitfromleft(1,2)}}", "foo.-abc-def--ghij-", "abc.def.ghij.fo.o") // combo + checks split for multiple instance of deliminator and deliminator being at the start or end
+	shouldMatch("*", "{{right(1,1)}}", "1234", "4")
+	shouldMatch("*", "{{right(1,3)}}", "1234", "234")
+	shouldMatch("*", "{{right(1,6)}}", "1234", "1234")
+	shouldMatch("*", "{{left(1,1)}}", "1234", "1")
+	shouldMatch("*", "{{left(1,3)}}", "1234", "123")
+	shouldMatch("*", "{{left(1,6)}}", "1234", "1234")
 }
diff --git a/server/sublist_test.go b/server/sublist_test.go
index 16444b92..1e2e0288 100644
--- a/server/sublist_test.go
+++ b/server/sublist_test.go
@@ -1602,28 +1602,6 @@ func TestMain(m *testing.M) {
 	os.Exit(m.Run())
 }
 
-func TestSublistHasInterest(t *testing.T) {
-	sl := NewSublistWithCache()
-	fooSub := newSub("foo")
-	sl.Insert(fooSub)
-
-	// Expect to find that "foo" matches but "bar" doesn't.
-	// At this point nothing should be in the cache.
-	require_True(t, sl.HasInterest("foo"))
-	require_False(t, sl.HasInterest("bar"))
-	require_Equal(t, sl.cacheHits, 0)
-
-	// Now call Match(), which will populate the cache.
-	sl.Match("foo")
-	require_Equal(t, sl.cacheHits, 0)
-
-	// Future calls to HasInterest() should hit the cache now.
-	for i := uint64(1); i <= 5; i++ {
-		require_True(t, sl.HasInterest("foo"))
-		require_Equal(t, sl.cacheHits, i)
-	}
-}
-
 func subsInit(pre string, toks []string) {
 	var sub string
 	for _, t := range toks {