Repository: nats-io/nats-server
Pull request #3814 (2023-01-25 20:17:35)

[ADDED] Stream subject mapping transforms
- [ ] Documentation added (if applicable)
- [X] Tests added
- [X] Branch rebased on top of current dev
- [x] Build is green in Travis CI
- [X] You have certified that the contribution is your original work and that you license the work to the project under the [Apache 2 license](https://github.com/nats-io/nats-server/blob/main/LICENSE)

Adds support for subject mapping transforms in streams

Introduces the following changes:

- Extracts subject transform code and tests from accounts.go to it's own separate set of files.
- Ability to add a subject mapping transform destination to sourced streams, allowing to map message subjects as they are received from the stream being sourced
- Ability to source more than once from the same stream, each time using a specific subject filter and destination transform
- Ability to add a stream input subject mapping transform to a stream which will not act as a filter but will transform the subject of messages matching the transform's source according to the transform's destination. This mapping happens regardless of how the messages are received into the stream (mirror, or subject(s) and/or source(s))

![Stream transform-2023-01-17-1401-3](https://user-images.githubusercontent.com/7819096/213826324-4ff402ce-160d-4220-b34c-a0b12844d730.png)

This enables many interesting use cases, such as KV bucket sourcing, or inserting a partition number token in the subject in order to then create a consumer per partition, directly at the stream definition level rather than having to resort to using the Core NATS subject mapping functionality (which requires administrative access to the server config file or account key to be configured).

/cc @nats-io/core


Patch:
diff --git a/server/accounts.go b/server/accounts.go
index 2d68641b..5bd07472 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -18,7 +18,6 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
-	"hash/fnv"
 	"hash/maphash"
 	"io"
 	"math"
@@ -26,7 +25,6 @@ import (
 	"net/http"
 	"net/textproto"
 	"reflect"
-	"regexp"
 	"sort"
 	"strconv"
 	"strings"
@@ -115,8 +113,8 @@ type streamImport struct {
 	acc     *Account
 	from    string
 	to      string
-	tr      *transform
-	rtr     *transform
+	tr      *subjectTransform
+	rtr     *subjectTransform
 	claim   *jwt.Import
 	usePub  bool
 	invalid bool
@@ -132,7 +130,7 @@ type serviceImport struct {
 	sid         []byte
 	from        string
 	to          string
-	tr          *transform
+	tr          *subjectTransform
 	ts          int64
 	rt          ServiceRespType
 	latency     *serviceLatency
@@ -166,28 +164,6 @@ const (
 	Chunked
 )
 
-var commaSeparatorRegEx = regexp.MustCompile(`,\s*`)
-var partitionMappingFunctionRegEx = regexp.MustCompile(`{{\s*[pP]artition\s*\((.*)\)\s*}}`)
-var wildcardMappingFunctionRegEx = regexp.MustCompile(`{{\s*[wW]ildcard\s*\((.*)\)\s*}}`)
-var splitFromLeftMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit[fF]rom[lL]eft\s*\((.*)\)\s*}}`)
-var splitFromRightMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit[fF]rom[rR]ight\s*\((.*)\)\s*}}`)
-var sliceFromLeftMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[lL]eft\s*\((.*)\)\s*}}`)
-var sliceFromRightMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[rR]ight\s*\((.*)\)\s*}}`)
-var splitMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit\s*\((.*)\)\s*}}`)
-
-// Enum for the subject mapping transform function types
-const (
-	NoTransform int16 = iota
-	BadTransform
-	Partition
-	Wildcard
-	SplitFromLeft
-	SplitFromRight
-	SliceFromLeft
-	SliceFromRight
-	Split
-)
-
 // String helper.
 func (rt ServiceRespType) String() string {
 	switch rt {
@@ -590,7 +566,7 @@ func NewMapDest(subject string, weight uint8) *MapDest {
 
 // destination is for internal representation for a weighted mapped destination.
 type destination struct {
-	tr     *transform
+	tr     *subjectTransform
 	weight uint8
 }
 
@@ -643,7 +619,7 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 		if err != nil {
 			return err
 		}
-		tr, err := newTransform(src, d.Subject)
+		tr, err := newSubjectTransform(src, d.Subject)
 		if err != nil {
 			return err
 		}
@@ -674,7 +650,7 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 				// We need to make the appropriate markers for the wildcards etc.
 				dest = transformTokenize(dest)
 			}
-			tr, err := newTransform(src, dest)
+			tr, err := newSubjectTransform(src, dest)
 			if err != nil {
 				return nil, err
 			}
@@ -730,38 +706,6 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 	return nil
 }
 
-// Helper function to tokenize subjects with partial wildcards into formal transform destinations.
-// e.g. foo.*.* -> foo.$1.$2
-func transformTokenize(subject string) string {
-	// We need to make the appropriate markers for the wildcards etc.
-	i := 1
-	var nda []string
-	for _, token := range strings.Split(subject, tsep) {
-		if token == "*" {
-			nda = append(nda, fmt.Sprintf("$%d", i))
-			i++
-		} else {
-			nda = append(nda, token)
-		}
-	}
-	return strings.Join(nda, tsep)
-}
-
-func transformUntokenize(subject string) (string, []string) {
-	var phs []string
-	var nda []string
-
-	for _, token := range strings.Split(subject, tsep) {
-		if len(token) > 1 && token[0] == '$' && token[1] >= '1' && token[1] <= '9' {
-			phs = append(phs, token)
-			nda = append(nda, "*")
-		} else {
-			nda = append(nda, token)
-		}
-	}
-	return strings.Join(nda, tsep), phs
-}
-
 // RemoveMapping will remove an existing mapping.
 func (a *Account) RemoveMapping(src string) bool {
 	a.mu.Lock()
@@ -869,8 +813,8 @@ func (a *Account) selectMappedSubject(dest string) (string, bool) {
 	if d != nil {
 		if len(d.tr.dtokmftokindexesargs) == 0 {
 			ndest = d.tr.dest
-		} else if nsubj, err := d.tr.transform(tts); err == nil {
-			ndest = nsubj
+		} else {
+			ndest = d.tr.TransformTokenizedSubject(tts)
 		}
 	}
 
@@ -1874,7 +1818,7 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 	// Check to see if we have a wildcard
 	var (
 		usePub bool
-		tr     *transform
+		tr     *subjectTransform
 		err    error
 	)
 	if subjectHasWildcard(to) {
@@ -1884,7 +1828,7 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 		} else {
 			to, _ = transformUntokenize(to)
 			// Create a transform. Do so in reverse such that $ symbols only exist in to
-			if tr, err = newTransform(to, transformTokenize(from)); err != nil {
+			if tr, err = newSubjectTransform(to, transformTokenize(from)); err != nil {
 				a.mu.Unlock()
 				return nil, fmt.Errorf("failed to create mapping transform for service import subject %q to %q: %v",
 					from, to, err)
@@ -2423,7 +2367,7 @@ func (a *Account) AddMappedStreamImportWithClaim(account *Account, from, to stri
 
 	var (
 		usePub bool
-		tr     *transform
+		tr     *subjectTransform
 		err    error
 	)
 	if subjectHasWildcard(from) {
@@ -2431,7 +2375,7 @@ func (a *Account) AddMappedStreamImportWithClaim(account *Account, from, to stri
 			usePub = true
 		} else {
 			// Create a transform
-			if tr, err = newTransform(from, transformTokenize(to)); err != nil {
+			if tr, err = newSubjectTransform(from, transformTokenize(to)); err != nil {
 				return fmt.Errorf("failed to create mapping transform for stream import subject %q to %q: %v",
 					from, to, err)
 			}
@@ -4286,439 +4230,3 @@ func (dr *CacheDirAccResolver) Start(s *Server) error {
 func (dr *CacheDirAccResolver) Reload() error {
 	return dr.DirAccResolver.Reload()
 }
-
-// Transforms for arbitrarily mapping subjects from one to another for maps, tees and filters.
-// These can also be used for proper mapping on wildcard exports/imports.
-// These will be grouped and caching and locking are assumed to be in the upper layers.
-type transform struct {
-	src, dest            string
-	dtoks                []string // destination tokens
-	stoks                []string // source tokens
-	dtokmftypes          []int16  // destination token mapping function types
-	dtokmftokindexesargs [][]int  // destination token mapping function array of source token index arguments
-	dtokmfintargs        []int32  // destination token mapping function int32 arguments
-	dtokmfstringargs     []string // destination token mapping function string arguments
-}
-
-func getMappingFunctionArgs(functionRegEx *regexp.Regexp, token string) []string {
-	commandStrings := functionRegEx.FindStringSubmatch(token)
-	if len(commandStrings) > 1 {
-		return commaSeparatorRegEx.Split(commandStrings[1], -1)
-	}
-	return nil
-}
-
-// Helper for mapping functions that take a wildcard index and an integer as arguments
-func transformIndexIntArgsHelper(token string, args []string, transformType int16) (int16, []int, int32, string, error) {
-	if len(args) < 2 {
-		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
-	}
-	if len(args) > 2 {
-		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
-	}
-	i, err := strconv.Atoi(strings.Trim(args[0], " "))
-	if err != nil {
-		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
-	}
-	mappingFunctionIntArg, err := strconv.Atoi(strings.Trim(args[1], " "))
-	if err != nil {
-		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
-	}
-
-	return transformType, []int{i}, int32(mappingFunctionIntArg), _EMPTY_, nil
-}
-
-// Helper to ingest and index the transform destination token (e.g. $x or {{}}) in the token
-// returns a transformation type, and three function arguments: an array of source subject token indexes, and a single number (e.g. number of partitions, or a slice size), and a string (e.g.a split delimiter)
-
-func indexPlaceHolders(token string) (int16, []int, int32, string, error) {
-	length := len(token)
-	if length > 1 {
-		// old $1, $2, etc... mapping format still supported to maintain backwards compatibility
-		if token[0] == '$' { // simple non-partition mapping
-			tp, err := strconv.Atoi(token[1:])
-			if err != nil {
-				// other things rely on tokens starting with $ so not an error just leave it as is
-				return NoTransform, []int{-1}, -1, _EMPTY_, nil
-			}
-			return Wildcard, []int{tp}, -1, _EMPTY_, nil
-		}
-
-		// New 'mustache' style mapping
-		if length > 4 && token[0] == '{' && token[1] == '{' && token[length-2] == '}' && token[length-1] == '}' {
-			// wildcard(wildcard token index) (equivalent to $)
-			args := getMappingFunctionArgs(wildcardMappingFunctionRegEx, token)
-			if args != nil {
-				if len(args) == 1 && args[0] == _EMPTY_ {
-					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
-				}
-				if len(args) == 1 {
-					tokenIndex, err := strconv.Atoi(strings.Trim(args[0], " "))
-					if err != nil {
-						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
-					}
-					return Wildcard, []int{tokenIndex}, -1, _EMPTY_, nil
-				} else {
-					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
-				}
-			}
-
-			// partition(number of partitions, token1, token2, ...)
-			args = getMappingFunctionArgs(partitionMappingFunctionRegEx, token)
-			if args != nil {
-				if len(args) < 2 {
-					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
-				}
-				if len(args) >= 2 {
-					mappingFunctionIntArg, err := strconv.Atoi(strings.Trim(args[0], " "))
-					if err != nil {
-						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
-					}
-					var numPositions = len(args[1:])
-					tokenIndexes := make([]int, numPositions)
-					for ti, t := range args[1:] {
-						i, err := strconv.Atoi(strings.Trim(t, " "))
-						if err != nil {
-							return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
-						}
-						tokenIndexes[ti] = i
-					}
-
-					return Partition, tokenIndexes, int32(mappingFunctionIntArg), _EMPTY_, nil
-				}
-			}
-
-			// SplitFromLeft(token, position)
-			args = getMappingFunctionArgs(splitFromLeftMappingFunctionRegEx, token)
-			if args != nil {
-				return transformIndexIntArgsHelper(token, args, SplitFromLeft)
-			}
-
-			// SplitFromRight(token, position)
-			args = getMappingFunctionArgs(splitFromRightMappingFunctionRegEx, token)
-			if args != nil {
-				return transformIndexIntArgsHelper(token, args, SplitFromRight)
-			}
-
-			// SliceFromLeft(token, position)
-			args = getMappingFunctionArgs(sliceFromLeftMappingFunctionRegEx, token)
-			if args != nil {
-				return transformIndexIntArgsHelper(token, args, SliceFromLeft)
-			}
-
-			// SliceFromRight(token, position)
-			args = getMappingFunctionArgs(sliceFromRightMappingFunctionRegEx, token)
-			if args != nil {
-				return transformIndexIntArgsHelper(token, args, SliceFromRight)
-			}
-
-			// split(token, deliminator)
-			args = getMappingFunctionArgs(splitMappingFunctionRegEx, token)
-			if args != nil {
-				if len(args) < 2 {
-					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
-				}
-				if len(args) > 2 {
-					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
-				}
-				i, err := strconv.Atoi(strings.Trim(args[0], " "))
-				if err != nil {
-					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
-				}
-				if strings.Contains(args[1], " ") || strings.Contains(args[1], tsep) {
-					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token: token, err: ErrorMappingDestinationFunctionInvalidArgument}
-				}
-
-				return Split, []int{i}, -1, args[1], nil
-			}
-
-			return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrUnknownMappingDestinationFunction}
-		}
-	}
-	return NoTransform, []int{-1}, -1, _EMPTY_, nil
-}
-
-// SubjectTransformer transforms subjects using mappings
-//
-// This API is not part of the public API and not subject to SemVer protections
-type SubjectTransformer interface {
-	Match(string) (string, error)
-}
-
-// NewSubjectTransformer creates a new SubjectTransformer
-//
-// This API is not part of the public API and not subject to SemVer protections
-func NewSubjectTransformer(src, dest string) (SubjectTransformer, error) {
-	return newTransform(src, dest)
-}
-
-// newTransform will create a new transform checking the src and dest subjects for accuracy.
-func newTransform(src, dest string) (*transform, error) {
-	// Both entries need to be valid subjects.
-	sv, stokens, npwcs, hasFwc := subjectInfo(src)
-	dv, dtokens, dnpwcs, dHasFwc := subjectInfo(dest)
-
-	// Make sure both are valid, match fwc if present and there are no pwcs in the dest subject.
-	if !sv || !dv || dnpwcs > 0 || hasFwc != dHasFwc {
-		return nil, ErrBadSubject
-	}
-
-	var dtokMappingFunctionTypes []int16
-	var dtokMappingFunctionTokenIndexes [][]int
-	var dtokMappingFunctionIntArgs []int32
-	var dtokMappingFunctionStringArgs []string
-
-	// If the src has partial wildcards then the dest needs to have the token place markers.
-	if npwcs > 0 || hasFwc {
-		// We need to count to make sure that the dest has token holders for the pwcs.
-		sti := make(map[int]int)
-		for i, token := range stokens {
-			if len(token) == 1 && token[0] == pwc {
-				sti[len(sti)+1] = i
-			}
-		}
-
-		nphs := 0
-		for _, token := range dtokens {
-			tranformType, transformArgWildcardIndexes, transfomArgInt, transformArgString, err := indexPlaceHolders(token)
-			if err != nil {
-				return nil, err
-			}
-
-			if tranformType == NoTransform {
-				dtokMappingFunctionTypes = append(dtokMappingFunctionTypes, NoTransform)
-				dtokMappingFunctionTokenIndexes = append(dtokMappingFunctionTokenIndexes, []int{-1})
-				dtokMappingFunctionIntArgs = append(dtokMappingFunctionIntArgs, -1)
-				dtokMappingFunctionStringArgs = append(dtokMappingFunctionStringArgs, _EMPTY_)
-			} else {
-				nphs++
-				// Now build up our runtime mapping from dest to source tokens.
-				var stis []int
-				for _, wildcardIndex := range transformArgWildcardIndexes {
-					if wildcardIndex > npwcs {
-						return nil, &mappingDestinationErr{fmt.Sprintf("%s: [%d]", token, wildcardIndex), ErrorMappingDestinationFunctionWildcardIndexOutOfRange}
-					}
-					stis = append(stis, sti[wildcardIndex])
-				}
-				dtokMappingFunctionTypes = append(dtokMappingFunctionTypes, tranformType)
-				dtokMappingFunctionTokenIndexes = append(dtokMappingFunctionTokenIndexes, stis)
-				dtokMappingFunctionIntArgs = append(dtokMappingFunctionIntArgs, transfomArgInt)
-				dtokMappingFunctionStringArgs = append(dtokMappingFunctionStringArgs, transformArgString)
-
-			}
-		}
-		if nphs < npwcs {
-			// not all wildcards are being used in the destination
-			return nil, &mappingDestinationErr{dest, ErrMappingDestinationNotUsingAllWildcards}
-		}
-	}
-
-	return &transform{src: src, dest: dest, dtoks: dtokens, stoks: stokens, dtokmftypes: dtokMappingFunctionTypes, dtokmftokindexesargs: dtokMappingFunctionTokenIndexes, dtokmfintargs: dtokMappingFunctionIntArgs, dtokmfstringargs: dtokMappingFunctionStringArgs}, nil
-}
-
-// Match will take a literal published subject that is associated with a client and will match and transform
-// the subject if possible.
-//
-// This API is not part of the public API and not subject to SemVer protections
-func (tr *transform) Match(subject string) (string, error) {
-	// TODO(dlc) - We could add in client here to allow for things like foo -> foo.$ACCOUNT
-
-	// Special case: matches any and no no-op transform. May not be legal config for some features
-	// but specific validations made at transform create time
-	if (tr.src == fwcs || tr.src == _EMPTY_) && (tr.dest == fwcs || tr.dest == _EMPTY_) {
-		return subject, nil
-	}
-
-	// Tokenize the subject. This should always be a literal subject.
-	tsa := [32]string{}
-	tts := tsa[:0]
-	start := 0
-	for i := 0; i < len(subject); i++ {
-		if subject[i] == btsep {
-			tts = append(tts, subject[start:i])
-			start = i + 1
-		}
-	}
-	tts = append(tts, subject[start:])
-	if !isValidLiteralSubject(tts) {
-		return _EMPTY_, ErrBadSubject
-	}
-
-	if (tr.src == _EMPTY_ || tr.src == fwcs) || isSubsetMatch(tts, tr.src) {
-		return tr.transform(tts)
-	}
-	return _EMPTY_, ErrNoTransforms
-}
-
-// transformSubject do not need to match, just transform.
-func (tr *transform) transformSubject(subject string) (string, error) {
-	// Tokenize the subject.
-	tsa := [32]string{}
-	tts := tsa[:0]
-	start := 0
-	for i := 0; i < len(subject); i++ {
-		if subject[i] == btsep {
-			tts = append(tts, subject[start:i])
-			start = i + 1
-		}
-	}
-	tts = append(tts, subject[start:])
-	return tr.transform(tts)
-}
-
-func (tr *transform) getHashPartition(key []byte, numBuckets int) string {
-	h := fnv.New32a()
-	h.Write(key)
-
-	return strconv.Itoa(int(h.Sum32() % uint32(numBuckets)))
-}
-
-// Do a transform on the subject to the dest subject.
-func (tr *transform) transform(tokens []string) (string, error) {
-	if len(tr.dtokmftypes) == 0 {
-		return tr.dest, nil
-	}
-
-	var b strings.Builder
-
-	// We need to walk destination tokens and create the mapped subject pulling tokens or mapping functions
-	// This is slow and that is ok, transforms should have caching layer in front for mapping transforms
-	// and export/import semantics with streams and services.
-	li := len(tr.dtokmftypes) - 1
-	for i, mfType := range tr.dtokmftypes {
-		if mfType == NoTransform {
-			// Break if fwc
-			if len(tr.dtoks[i]) == 1 && tr.dtoks[i][0] == fwc {
-				break
-			}
-			b.WriteString(tr.dtoks[i])
-		} else {
-			switch mfType {
-			case Partition:
-				var (
-					_buffer       [64]byte
-					keyForHashing = _buffer[:0]
-				)
-				for _, sourceToken := range tr.dtokmftokindexesargs[i] {
-					keyForHashing = append(keyForHashing, []byte(tokens[sourceToken])...)
-				}
-				b.WriteString(tr.getHashPartition(keyForHashing, int(tr.dtokmfintargs[i])))
-			case Wildcard: // simple substitution
-				b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]])
-			case SplitFromLeft:
-				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
-				sourceTokenLen := len(sourceToken)
-				position := int(tr.dtokmfintargs[i])
-				if position > 0 && position < sourceTokenLen {
-					b.WriteString(sourceToken[:position])
-					b.WriteString(tsep)
-					b.WriteString(sourceToken[position:])
-				} else { // too small to split at the requested position: don't split
-					b.WriteString(sourceToken)
-				}
-			case SplitFromRight:
-				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
-				sourceTokenLen := len(sourceToken)
-				position := int(tr.dtokmfintargs[i])
-				if position > 0 && position < sourceTokenLen {
-					b.WriteString(sourceToken[:sourceTokenLen-position])
-					b.WriteString(tsep)
-					b.WriteString(sourceToken[sourceTokenLen-position:])
-				} else { // too small to split at the requested position: don't split
-					b.WriteString(sourceToken)
-				}
-			case SliceFromLeft:
-				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
-				sourceTokenLen := len(sourceToken)
-				sliceSize := int(tr.dtokmfintargs[i])
-				if sliceSize > 0 && sliceSize < sourceTokenLen {
-					for i := 0; i+sliceSize <= sourceTokenLen; i += sliceSize {
-						if i != 0 {
-							b.WriteString(tsep)
-						}
-						b.WriteString(sourceToken[i : i+sliceSize])
-						if i+sliceSize != sourceTokenLen && i+sliceSize+sliceSize > sourceTokenLen {
-							b.WriteString(tsep)
-							b.WriteString(sourceToken[i+sliceSize:])
-							break
-						}
-					}
-				} else { // too small to slice at the requested size: don't slice
-					b.WriteString(sourceToken)
-				}
-			case SliceFromRight:
-				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
-				sourceTokenLen := len(sourceToken)
-				sliceSize := int(tr.dtokmfintargs[i])
-				if sliceSize > 0 && sliceSize < sourceTokenLen {
-					remainder := sourceTokenLen % sliceSize
-					if remainder > 0 {
-						b.WriteString(sourceToken[:remainder])
-						b.WriteString(tsep)
-					}
-					for i := remainder; i+sliceSize <= sourceTokenLen; i += sliceSize {
-						b.WriteString(sourceToken[i : i+sliceSize])
-						if i+sliceSize < sourceTokenLen {
-							b.WriteString(tsep)
-						}
-					}
-				} else { // too small to slice at the requested size: don't slice
-					b.WriteString(sourceToken)
-				}
-			case Split:
-				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
-				splits := strings.Split(sourceToken, tr.dtokmfstringargs[i])
-				for j, split := range splits {
-					if split != _EMPTY_ {
-						b.WriteString(split)
-					}
-					if j < len(splits)-1 && splits[j+1] != _EMPTY_ && !(j == 0 && split == _EMPTY_) {
-						b.WriteString(tsep)
-					}
-				}
-			}
-		}
-
-		if i < li {
-			b.WriteByte(btsep)
-		}
-	}
-
-	// We may have more source tokens available. This happens with ">".
-	if tr.dtoks[len(tr.dtoks)-1] == ">" {
-		for sli, i := len(tokens)-1, len(tr.stoks)-1; i < len(tokens); i++ {
-			b.WriteString(tokens[i])
-			if i < sli {
-				b.WriteByte(btsep)
-			}
-		}
-	}
-	return b.String(), nil
-}
-
-// Reverse a transform.
-func (tr *transform) reverse() *transform {
-	if len(tr.dtokmftokindexesargs) == 0 {
-		rtr, _ := newTransform(tr.dest, tr.src)
-		return rtr
-	}
-	// If we are here we need to dynamically get the correct reverse
-	// of this transform.
-	nsrc, phs := transformUntokenize(tr.dest)
-	var nda []string
-	for _, token := range tr.stoks {
-		if token == "*" {
-			if len(phs) == 0 {
-				// TODO(dlc) - Should not happen
-				return nil
-			}
-			nda = append(nda, phs[0])
-			phs = phs[1:]
-		} else {
-			nda = append(nda, token)
-		}
-	}
-	ndest := strings.Join(nda, tsep)
-	rtr, _ := newTransform(nsrc, ndest)
-	return rtr
-}
diff --git a/server/client.go b/server/client.go
index 798a89e2..0c146114 100644
--- a/server/client.go
+++ b/server/client.go
@@ -2754,10 +2754,8 @@ func (c *client) addShadowSub(sub *subscription, ime *ime) (*subscription, error
 		if ime.overlapSubj != _EMPTY_ {
 			s = ime.overlapSubj
 		}
-		subj, err := im.rtr.transformSubject(s)
-		if err != nil {
-			return nil, err
-		}
+		subj := im.rtr.TransformSubject(s)
+
 		nsub.subject = []byte(subj)
 	} else if !im.usePub || (im.usePub && ime.overlapSubj != _EMPTY_) || !ime.dyn {
 		if ime.overlapSubj != _EMPTY_ {
@@ -3035,7 +3033,7 @@ func (c *client) msgHeaderForRouteOrLeaf(subj, reply []byte, rt *routeTarget, ac
 		// Remap subject if its a shadow subscription, treat like a normal client.
 		if rt.sub.im != nil {
 			if rt.sub.im.tr != nil {
-				to, _ := rt.sub.im.tr.transformSubject(string(subj))
+				to := rt.sub.im.tr.TransformSubject(string(subj))
 				subj = []byte(to)
 			} else if !rt.sub.im.usePub {
 				subj = []byte(rt.sub.im.to)
@@ -3991,7 +3989,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 
 	if si.tr != nil {
 		// FIXME(dlc) - This could be slow, may want to look at adding cache to bare transforms?
-		to, _ = si.tr.transformSubject(subject)
+		to = si.tr.TransformSubject(subject)
 	} else if si.usePub {
 		to = subject
 	}
@@ -4280,7 +4278,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 				continue
 			}
 			if sub.im.tr != nil {
-				to, _ := sub.im.tr.transformSubject(string(subject))
+				to := sub.im.tr.TransformSubject(string(subject))
 				dsubj = append(_dsubj[:0], to...)
 			} else if sub.im.usePub {
 				dsubj = append(_dsubj[:0], subj...)
@@ -4427,7 +4425,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 					continue
 				}
 				if sub.im.tr != nil {
-					to, _ := sub.im.tr.transformSubject(string(subject))
+					to := sub.im.tr.TransformSubject(string(subject))
 					dsubj = append(_dsubj[:0], to...)
 				} else if sub.im.usePub {
 					dsubj = append(_dsubj[:0], subj...)
diff --git a/server/consumer.go b/server/consumer.go
index b5d5adc2..cd811f27 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -456,17 +456,8 @@ func checkConsumerCfg(
 		}
 	}
 
-	// As best we can make sure the filtered subject is valid.
-	if config.FilterSubject != _EMPTY_ {
-		subjects := copyStrings(cfg.Subjects)
-		// explicitly skip validFilteredSubject when recovering
-		hasExt := isRecovering
-		if !isRecovering {
-			subjects, hasExt = gatherSourceMirrorSubjects(subjects, cfg, acc)
-		}
-		if !hasExt && !validFilteredSubject(config.FilterSubject, subjects) {
-			return NewJSConsumerFilterNotSubsetError()
-		}
+	if config.FilterSubject != _EMPTY_ && !IsValidSubject(config.FilterSubject) {
+		return NewJSStreamInvalidConfigError(ErrBadSubject)
 	}
 
 	// Helper function to formulate similar errors.
@@ -4262,26 +4253,6 @@ func deliveryFormsCycle(cfg *StreamConfig, deliverySubject string) bool {
 	return false
 }
 
-// Check that the filtered subject is valid given a set of stream subjects.
-func validFilteredSubject(filteredSubject string, subjects []string) bool {
-	if !IsValidSubject(filteredSubject) {
-		return false
-	}
-	hasWC := subjectHasWildcard(filteredSubject)
-
-	for _, subject := range subjects {
-		if subjectIsSubsetMatch(filteredSubject, subject) {
-			return true
-		}
-		// If we have a wildcard as the filtered subject check to see if we are
-		// a wider scope but do match a subject.
-		if hasWC && subjectIsSubsetMatch(subject, filteredSubject) {
-			return true
-		}
-	}
-	return false
-}
-
 // switchToEphemeral is called on startup when recovering ephemerals.
 func (o *consumer) switchToEphemeral() {
 	o.mu.Lock()
diff --git a/server/stream.go b/server/stream.go
index 7c95cc1d..1f3aeb1f 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -58,6 +58,9 @@ type StreamConfig struct {
 	Mirror       *StreamSource   `json:"mirror,omitempty"`
 	Sources      []*StreamSource `json:"sources,omitempty"`
 
+	// Allow applying a subject transform to incoming messages before doing anything else
+	SubjectTransform *SubjectTransformConfig `json:"subject_transform,omitempty"`
+
 	// Allow republish of the message after being sequenced and stored.
 	RePublish *RePublish `json:"republish,omitempty"`
 
@@ -82,6 +85,12 @@ type StreamConfig struct {
 	AllowRollup bool `json:"allow_rollup_hdrs"`
 }
 
+// SubjectTransformConfig is for applying a subject transform (to matching messages) before doing anything else when a new message is received
+type SubjectTransformConfig struct {
+	Source      string `json:"src,omitempty"`
+	Destination string `json:"dest"`
+}
+
 // RePublish is for republishing messages once committed to a stream.
 type RePublish struct {
 	Source      string `json:"src,omitempty"`
@@ -154,20 +163,23 @@ type PeerInfo struct {
 
 // StreamSourceInfo shows information about an upstream stream source.
 type StreamSourceInfo struct {
-	Name     string          `json:"name"`
-	External *ExternalStream `json:"external,omitempty"`
-	Lag      uint64          `json:"lag"`
-	Active   time.Duration   `json:"active"`
-	Error    *ApiError       `json:"error,omitempty"`
+	Name                 string          `json:"name"`
+	External             *ExternalStream `json:"external,omitempty"`
+	Lag                  uint64          `json:"lag"`
+	Active               time.Duration   `json:"active"`
+	Error                *ApiError       `json:"error,omitempty"`
+	FilterSubject        string          `json:"filter_subject,omitempty"`
+	SubjectTransformDest string          `json:"subject_transform_dest,omitempty"`
 }
 
 // StreamSource dictates how streams can source from other streams.
 type StreamSource struct {
-	Name          string          `json:"name"`
-	OptStartSeq   uint64          `json:"opt_start_seq,omitempty"`
-	OptStartTime  *time.Time      `json:"opt_start_time,omitempty"`
-	FilterSubject string          `json:"filter_subject,omitempty"`
-	External      *ExternalStream `json:"external,omitempty"`
+	Name                 string          `json:"name"`
+	OptStartSeq          uint64          `json:"opt_start_seq,omitempty"`
+	OptStartTime         *time.Time      `json:"opt_start_time,omitempty"`
+	FilterSubject        string          `json:"filter_subject,omitempty"`
+	SubjectTransformDest string          `json:"subject_transform_dest,omitempty"`
+	External             *ExternalStream `json:"external,omitempty"`
 
 	// Internal
 	iname string // For indexing when stream names are the same for multiple sources.
@@ -220,8 +232,11 @@ type stream struct {
 	// Indicates we have direct consumers.
 	directs int
 
+	// For input subject transform
+	itr *subjectTransform
+
 	// For republishing.
-	tr *transform
+	tr *subjectTransform
 
 	// For processing consumers without main stream lock.
 	clsMu sync.RWMutex
@@ -271,6 +286,8 @@ type sourceInfo struct {
 	qch   chan struct{}
 	sip   bool // setup in progress
 	wg    sync.WaitGroup
+	sf    string // subject filter
+	tr    *subjectTransform
 }
 
 // For mirrors and direct get
@@ -404,10 +421,17 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		}
 	}
 
-	// Setup our internal indexed names here for sources.
+	// Setup our internal indexed names here for sources and check if the transform (if any) is valid.
 	if len(cfg.Sources) > 0 {
-		for _, ssi := range cfg.Sources {
-			ssi.setIndexName()
+		for i, ssi := range cfg.Sources {
+			ssi.setIndexName(i)
+			// check the transform, if any, is valid
+			if ssi.SubjectTransformDest != _EMPTY_ {
+				if _, err = newSubjectTransform(ssi.FilterSubject, ssi.SubjectTransformDest); err != nil {
+					jsa.mu.Unlock()
+					return nil, fmt.Errorf("subject transform for the source not valid %w", err)
+				}
+			}
 		}
 	}
 
@@ -454,16 +478,22 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		mset.ackq = s.newIPQueue(qpfx + "acks") // of uint64
 	}
 
+	// Check for input subject transform
+	if cfg.SubjectTransform != nil {
+		tr, err := newSubjectTransform(cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination)
+		if err != nil {
+			jsa.mu.Unlock()
+			return nil, fmt.Errorf("stream input subject transform not valid %w", err)
+		}
+		mset.itr = tr
+	}
+
 	// Check for RePublish.
 	if cfg.RePublish != nil {
-		// Empty same as all.
-		if cfg.RePublish.Source == _EMPTY_ {
-			cfg.RePublish.Source = fwcs
-		}
-		tr, err := newTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
+		tr, err := newSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
 		if err != nil {
 			jsa.mu.Unlock()
-			return nil, fmt.Errorf("stream configuration for republish not valid")
+			return nil, fmt.Errorf("stream republish transform not valid %w", err)
 		}
 		// Assign our transform for republishing.
 		mset.tr = tr
@@ -563,13 +593,14 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 	return mset, nil
 }
 
-// Sets the index name. Usually just the stream name but when the stream is external we will
+// Sets the index name. Usually just the stream name and the source's index (so you can source from the same stream more than once)
+// but when the stream is external we will
 // use additional information in case the stream names are the same.
-func (ssi *StreamSource) setIndexName() {
+func (ssi *StreamSource) setIndexName(i int) {
 	if ssi.External != nil {
-		ssi.iname = ssi.Name + ":" + getHash(ssi.External.ApiPrefix)
+		ssi.iname = strconv.Itoa(i) + ":" + ssi.Name + ":" + getHash(ssi.External.ApiPrefix)
 	} else {
-		ssi.iname = ssi.Name
+		ssi.iname = strconv.Itoa(i) + ":" + ssi.Name
 	}
 }
 
@@ -1044,28 +1075,10 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		return exists, cfg.MaxMsgSize, cfg.Subjects
 	}
 
-	hasFilterSubjectOverlap := func(filter string, streamSubs []string) bool {
-		if filter == _EMPTY_ || len(streamSubs) == 0 {
-			return true
-		}
-		for _, sub := range streamSubs {
-			if SubjectsCollide(sub, filter) {
-				return true
-			}
-		}
-		return false
-	}
-
 	var streamSubs []string
 	var deliveryPrefixes []string
 	var apiPrefixes []string
 
-	// Some errors we want to suppress on recovery.
-	var isRecovering bool
-	if js := s.getJetStream(); js != nil {
-		isRecovering = js.isMetaRecovering()
-	}
-
 	// Do some pre-checking for mirror config to avoid cycles in clustered mode.
 	if cfg.Mirror != nil {
 		if len(cfg.Subjects) > 0 {
@@ -1084,11 +1097,6 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 			if cfg.MaxMsgSize > 0 && maxMsgSize > 0 && cfg.MaxMsgSize < maxMsgSize {
 				return StreamConfig{}, NewJSMirrorMaxMessageSizeTooBigError()
 			}
-			if !isRecovering && !hasFilterSubjectOverlap(cfg.Mirror.FilterSubject, subs) {
-				return StreamConfig{}, NewJSStreamInvalidConfigError(
-					fmt.Errorf("mirror '%s' filter subject '%s' does not overlap with any origin stream subject",
-						cfg.Mirror.Name, cfg.Mirror.FilterSubject))
-			}
 		}
 		if cfg.Mirror.External != nil {
 			if cfg.Mirror.External.DeliverPrefix != _EMPTY_ {
@@ -1124,11 +1132,6 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 				if cfg.MaxMsgSize > 0 && maxMsgSize > 0 && cfg.MaxMsgSize < maxMsgSize {
 					return StreamConfig{}, NewJSSourceMaxMessageSizeTooBigError()
 				}
-				if !isRecovering && !hasFilterSubjectOverlap(src.FilterSubject, streamSubs) {
-					return StreamConfig{}, NewJSStreamInvalidConfigError(
-						fmt.Errorf("source '%s' filter subject '%s' does not overlap with any origin stream subject",
-							src.Name, src.FilterSubject))
-				}
 			}
 			if src.External == nil {
 				continue
@@ -1277,7 +1280,7 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		if formsCycle {
 			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish destination forms a cycle"))
 		}
-		if _, err := newTransform(cfg.RePublish.Source, cfg.RePublish.Destination); err != nil {
+		if _, err := newSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination); err != nil {
 			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish not valid"))
 		}
 	}
@@ -1497,14 +1500,22 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 			for _, s := range ocfg.Sources {
 				current[s.iname] = s.FilterSubject
 			}
-			for _, s := range cfg.Sources {
-				s.setIndexName()
+			for i, s := range cfg.Sources {
+				s.setIndexName(i)
 				if oFilter, ok := current[s.iname]; !ok {
 					if mset.sources == nil {
 						mset.sources = make(map[string]*sourceInfo)
 					}
 					mset.cfg.Sources = append(mset.cfg.Sources, s)
-					si := &sourceInfo{name: s.Name, iname: s.iname}
+					si := &sourceInfo{name: s.Name, iname: s.iname, sf: s.FilterSubject}
+					// Check for transform.
+					if s.SubjectTransformDest != _EMPTY_ {
+						var err error
+						if si.tr, err = newSubjectTransform(s.FilterSubject, s.SubjectTransformDest); err != nil {
+							return err
+						}
+					}
+
 					mset.sources[s.iname] = si
 					mset.setStartingSequenceForSource(s.iname)
 					mset.setSourceConsumer(s.iname, si.sseq+1, time.Time{})
@@ -1557,7 +1568,7 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 		if cfg.RePublish.Source == _EMPTY_ {
 			cfg.RePublish.Source = fwcs
 		}
-		tr, err := newTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
+		tr, err := newSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
 		if err != nil {
 			jsa.mu.Unlock()
 			return fmt.Errorf("stream configuration for republish not valid")
@@ -1707,138 +1718,16 @@ func (mset *stream) sourcesInfo() (sis []*StreamSourceInfo) {
 	return sis
 }
 
-func gatherSourceMirrorSubjects(subjects []string, cfg *StreamConfig, acc *Account) ([]string, bool) {
-	var hasExt bool
-	var seen map[string]bool
-
-	if cfg.Mirror != nil {
-		subjs, localHasExt := acc.streamSourceSubjects(cfg.Mirror, make(map[string]bool))
-		if len(subjs) > 0 {
-			subjects = append(subjects, subjs...)
-		}
-		if localHasExt {
-			hasExt = true
-		}
-	} else if len(cfg.Sources) > 0 {
-		seen = make(map[string]bool)
-		for _, si := range cfg.Sources {
-			subjs, localHasExt := acc.streamSourceSubjects(si, seen)
-			if len(subjs) > 0 {
-				subjects = append(subjects, subjs...)
-			}
-			if localHasExt {
-				hasExt = true
-			}
-		}
-	}
-
-	return subjects, hasExt
-}
-
-// Return the subjects for a stream source.
-func (a *Account) streamSourceSubjects(ss *StreamSource, seen map[string]bool) (subjects []string, hasExt bool) {
-	if ss != nil && ss.External != nil {
-		return nil, true
-	}
-
-	s, js, _ := a.getJetStreamFromAccount()
-
-	if !s.JetStreamIsClustered() {
-		return a.streamSourceSubjectsNotClustered(ss.Name, seen)
-	} else {
-		return js.streamSourceSubjectsClustered(a.Name, ss.Name, seen)
-	}
-}
-
-func (js *jetStream) streamSourceSubjectsClustered(accountName, streamName string, seen map[string]bool) (subjects []string, hasExt bool) {
-	if seen[streamName] {
-		return nil, false
-	}
-
-	// We are clustered here so need to work through stream assignments.
-	sa := js.streamAssignment(accountName, streamName)
-	if sa == nil {
-		return nil, false
-	}
-	seen[streamName] = true
-
-	js.mu.RLock()
-	cfg := sa.Config
-	if len(cfg.Subjects) > 0 {
-		subjects = append(subjects, cfg.Subjects...)
-	}
-
-	// Check if we need to keep going.
-	var sources []*StreamSource
-	if cfg.Mirror != nil {
-		sources = append(sources, cfg.Mirror)
-	} else if len(cfg.Sources) > 0 {
-		sources = append(sources, cfg.Sources...)
-	}
-	js.mu.RUnlock()
-
-	if len(sources) > 0 {
-		var subjs []string
-		if acc, err := js.srv.lookupAccount(accountName); err == nil {
-			for _, ss := range sources {
-				subjs, hasExt = acc.streamSourceSubjects(ss, seen)
-				if len(subjs) > 0 {
-					subjects = append(subjects, subjs...)
-				}
-				if hasExt {
-					break
-				}
-			}
-		}
-	}
-
-	return subjects, hasExt
-}
-
-func (a *Account) streamSourceSubjectsNotClustered(streamName string, seen map[string]bool) (subjects []string, hasExt bool) {
-	if seen[streamName] {
-		return nil, false
-	}
-
-	mset, err := a.lookupStream(streamName)
-	if err != nil {
-		return nil, false
-	}
-	seen[streamName] = true
-
-	cfg := mset.config()
-	if len(cfg.Subjects) > 0 {
-		subjects = append(subjects, cfg.Subjects...)
-	}
-
-	var subjs []string
-	if cfg.Mirror != nil {
-		subjs, hasExt = a.streamSourceSubjects(cfg.Mirror, seen)
-		if len(subjs) > 0 {
-			subjects = append(subjects, subjs...)
-		}
-	} else if len(cfg.Sources) > 0 {
-		for _, si := range cfg.Sources {
-			subjs, hasExt = a.streamSourceSubjects(si, seen)
-			if len(subjs) > 0 {
-				subjects = append(subjects, subjs...)
-			}
-			if hasExt {
-				break
-			}
-		}
-	}
-
-	return subjects, hasExt
-}
-
 // Lock should be held
 func (mset *stream) sourceInfo(si *sourceInfo) *StreamSourceInfo {
 	if si == nil {
 		return nil
 	}
 
-	ssi := &StreamSourceInfo{Name: si.name, Lag: si.lag, Error: si.err}
+	var ssi = StreamSourceInfo{Name: si.name, Lag: si.lag, Error: si.err, FilterSubject: si.sf}
+	if si.tr != nil {
+		ssi.SubjectTransformDest = si.tr.dest
+	}
 	// If we have not heard from the source, set Active to -1.
 	if si.last.IsZero() {
 		ssi.Active = -1
@@ -1858,7 +1747,7 @@ func (mset *stream) sourceInfo(si *sourceInfo) *StreamSourceInfo {
 			DeliverPrefix: ext.DeliverPrefix,
 		}
 	}
-	return ssi
+	return &ssi
 }
 
 // Return our source info for our mirror.
@@ -2833,6 +2722,11 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, m *inMsg) bool {
 	// Hold onto the origin reply which has all the metadata.
 	hdr = genHeader(hdr, JSStreamSource, si.genSourceHeader(m.rply))
 
+	// Do the subject transform for the source if there's one
+	if si.tr != nil {
+		m.subj = si.tr.TransformSubject(m.subj)
+	}
+
 	var err error
 	// If we are clustered we need to propose this message to the underlying raft group.
 	if node != nil {
@@ -2869,7 +2763,10 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, m *inMsg) bool {
 // Generate a new style source header.
 func (si *sourceInfo) genSourceHeader(reply string) string {
 	var b strings.Builder
-	b.WriteString(si.iname)
+	// strip the source index number from the iname
+	name := si.iname[strings.IndexRune(si.iname, ':')+1:]
+
+	b.WriteString(name)
 	b.WriteByte(' ')
 	// Grab sequence as text here from reply subject.
 	var tsa [expectedNumReplyTokens]string
@@ -2964,11 +2861,16 @@ func (mset *stream) startingSequenceForSources() {
 	// Always reset here.
 	mset.sources = make(map[string]*sourceInfo)
 
-	for _, ssi := range mset.cfg.Sources {
+	for i, ssi := range mset.cfg.Sources {
 		if ssi.iname == _EMPTY_ {
-			ssi.setIndexName()
+			ssi.setIndexName(i)
 		}
-		si := &sourceInfo{name: ssi.Name, iname: ssi.iname}
+		si := &sourceInfo{name: ssi.Name, iname: ssi.iname, sf: ssi.FilterSubject}
+		// Check for transform.
+		if ssi.SubjectTransformDest != _EMPTY_ {
+			si.tr, _ = newSubjectTransform(ssi.FilterSubject, ssi.SubjectTransformDest) // can not return an error because validated in AddStream
+		}
+
 		mset.sources[ssi.iname] = si
 	}
 
@@ -3681,6 +3583,15 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 		return nil
 	}
 
+	// Apply the input subject transform if any
+	if mset.itr != nil {
+		ts, err := mset.itr.Match(subject)
+		if err == nil {
+			// no filtering: if the subject doesn't map the source of the transform, don't change it
+			subject = ts
+		}
+	}
+
 	var accName string
 	if mset.acc != nil {
 		accName = mset.acc.Name
diff --git a/server/subject_transform.go b/server/subject_transform.go
new file mode 100644
index 00000000..2dd553c4
--- /dev/null
+++ b/server/subject_transform.go
@@ -0,0 +1,518 @@
+// 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
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"fmt"
+	"hash/fnv"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+var commaSeparatorRegEx = regexp.MustCompile(`,\s*`)
+var partitionMappingFunctionRegEx = regexp.MustCompile(`{{\s*[pP]artition\s*\((.*)\)\s*}}`)
+var wildcardMappingFunctionRegEx = regexp.MustCompile(`{{\s*[wW]ildcard\s*\((.*)\)\s*}}`)
+var splitFromLeftMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit[fF]rom[lL]eft\s*\((.*)\)\s*}}`)
+var splitFromRightMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit[fF]rom[rR]ight\s*\((.*)\)\s*}}`)
+var sliceFromLeftMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[lL]eft\s*\((.*)\)\s*}}`)
+var sliceFromRightMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[rR]ight\s*\((.*)\)\s*}}`)
+var splitMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]plit\s*\((.*)\)\s*}}`)
+
+// Enum for the subject mapping subjectTransform function types
+const (
+	NoTransform int16 = iota
+	BadTransform
+	Partition
+	Wildcard
+	SplitFromLeft
+	SplitFromRight
+	SliceFromLeft
+	SliceFromRight
+	Split
+)
+
+// Transforms for arbitrarily mapping subjects from one to another for maps, tees and filters.
+// These can also be used for proper mapping on wildcard exports/imports.
+// These will be grouped and caching and locking are assumed to be in the upper layers.
+type subjectTransform struct {
+	src, dest            string
+	dtoks                []string // destination tokens
+	stoks                []string // source tokens
+	dtokmftypes          []int16  // destination token mapping function types
+	dtokmftokindexesargs [][]int  // destination token mapping function array of source token index arguments
+	dtokmfintargs        []int32  // destination token mapping function int32 arguments
+	dtokmfstringargs     []string // destination token mapping function string arguments
+}
+
+// SubjectTransform transforms subjects using mappings
+//
+// This API is not part of the public API and not subject to SemVer protections
+type SubjectTransform interface {
+	// TODO(dlc) - We could add in client here to allow for things like foo -> foo.$ACCOUNT
+	Match(string) (string, error)
+	TransformSubject(subject string) string
+	TransformTokenizedSubject(tokens []string) string
+}
+
+func newSubjectTransform(src, dest string) (*subjectTransform, error) {
+	// No source given is equivalent to the source being ">"
+	if src == _EMPTY_ {
+		src = fwcs
+	}
+
+	// Both entries need to be valid subjects.
+	sv, stokens, npwcs, hasFwc := subjectInfo(src)
+	dv, dtokens, dnpwcs, dHasFwc := subjectInfo(dest)
+
+	// Make sure both are valid, match fwc if present and there are no pwcs in the dest subject.
+	if !sv || !dv || dnpwcs > 0 || hasFwc != dHasFwc {
+		return nil, ErrBadSubject
+	}
+
+	var dtokMappingFunctionTypes []int16
+	var dtokMappingFunctionTokenIndexes [][]int
+	var dtokMappingFunctionIntArgs []int32
+	var dtokMappingFunctionStringArgs []string
+
+	// If the src has partial wildcards then the dest needs to have the token place markers.
+	if npwcs > 0 || hasFwc {
+		// We need to count to make sure that the dest has token holders for the pwcs.
+		sti := make(map[int]int)
+		for i, token := range stokens {
+			if len(token) == 1 && token[0] == pwc {
+				sti[len(sti)+1] = i
+			}
+		}
+
+		nphs := 0
+		for _, token := range dtokens {
+			tranformType, transformArgWildcardIndexes, transfomArgInt, transformArgString, err := indexPlaceHolders(token)
+			if err != nil {
+				return nil, err
+			}
+
+			if tranformType == NoTransform {
+				dtokMappingFunctionTypes = append(dtokMappingFunctionTypes, NoTransform)
+				dtokMappingFunctionTokenIndexes = append(dtokMappingFunctionTokenIndexes, []int{-1})
+				dtokMappingFunctionIntArgs = append(dtokMappingFunctionIntArgs, -1)
+				dtokMappingFunctionStringArgs = append(dtokMappingFunctionStringArgs, _EMPTY_)
+			} else {
+				nphs++
+				// Now build up our runtime mapping from dest to source tokens.
+				var stis []int
+				for _, wildcardIndex := range transformArgWildcardIndexes {
+					if wildcardIndex > npwcs {
+						return nil, &mappingDestinationErr{fmt.Sprintf("%s: [%d]", token, wildcardIndex), ErrorMappingDestinationFunctionWildcardIndexOutOfRange}
+					}
+					stis = append(stis, sti[wildcardIndex])
+				}
+				dtokMappingFunctionTypes = append(dtokMappingFunctionTypes, tranformType)
+				dtokMappingFunctionTokenIndexes = append(dtokMappingFunctionTokenIndexes, stis)
+				dtokMappingFunctionIntArgs = append(dtokMappingFunctionIntArgs, transfomArgInt)
+				dtokMappingFunctionStringArgs = append(dtokMappingFunctionStringArgs, transformArgString)
+
+			}
+		}
+		if nphs < npwcs {
+			// not all wildcards are being used in the destination
+			return nil, &mappingDestinationErr{dest, ErrMappingDestinationNotUsingAllWildcards}
+		}
+	}
+
+	return &subjectTransform{
+		src:                  src,
+		dest:                 dest,
+		dtoks:                dtokens,
+		stoks:                stokens,
+		dtokmftypes:          dtokMappingFunctionTypes,
+		dtokmftokindexesargs: dtokMappingFunctionTokenIndexes,
+		dtokmfintargs:        dtokMappingFunctionIntArgs,
+		dtokmfstringargs:     dtokMappingFunctionStringArgs,
+	}, nil
+}
+
+func getMappingFunctionArgs(functionRegEx *regexp.Regexp, token string) []string {
+	commandStrings := functionRegEx.FindStringSubmatch(token)
+	if len(commandStrings) > 1 {
+		return commaSeparatorRegEx.Split(commandStrings[1], -1)
+	}
+	return nil
+}
+
+// Helper for mapping functions that take a wildcard index and an integer as arguments
+func transformIndexIntArgsHelper(token string, args []string, transformType int16) (int16, []int, int32, string, error) {
+	if len(args) < 2 {
+		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+	}
+	if len(args) > 2 {
+		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
+	}
+	i, err := strconv.Atoi(strings.Trim(args[0], " "))
+	if err != nil {
+		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+	}
+	mappingFunctionIntArg, err := strconv.Atoi(strings.Trim(args[1], " "))
+	if err != nil {
+		return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+	}
+
+	return transformType, []int{i}, int32(mappingFunctionIntArg), _EMPTY_, nil
+}
+
+// Helper to ingest and index the subjectTransform destination token (e.g. $x or {{}}) in the token
+// returns a transformation type, and three function arguments: an array of source subject token indexes,
+// and a single number (e.g. number of partitions, or a slice size), and a string (e.g.a split delimiter)
+func indexPlaceHolders(token string) (int16, []int, int32, string, error) {
+	length := len(token)
+	if length > 1 {
+		// old $1, $2, etc... mapping format still supported to maintain backwards compatibility
+		if token[0] == '$' { // simple non-partition mapping
+			tp, err := strconv.Atoi(token[1:])
+			if err != nil {
+				// other things rely on tokens starting with $ so not an error just leave it as is
+				return NoTransform, []int{-1}, -1, _EMPTY_, nil
+			}
+			return Wildcard, []int{tp}, -1, _EMPTY_, nil
+		}
+
+		// New 'mustache' style mapping
+		if length > 4 && token[0] == '{' && token[1] == '{' && token[length-2] == '}' && token[length-1] == '}' {
+			// wildcard(wildcard token index) (equivalent to $)
+			args := getMappingFunctionArgs(wildcardMappingFunctionRegEx, token)
+			if args != nil {
+				if len(args) == 1 && args[0] == _EMPTY_ {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+				}
+				if len(args) == 1 {
+					tokenIndex, err := strconv.Atoi(strings.Trim(args[0], " "))
+					if err != nil {
+						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+					}
+					return Wildcard, []int{tokenIndex}, -1, _EMPTY_, nil
+				} else {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
+				}
+			}
+
+			// partition(number of partitions, token1, token2, ...)
+			args = getMappingFunctionArgs(partitionMappingFunctionRegEx, token)
+			if args != nil {
+				if len(args) < 2 {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+				}
+				if len(args) >= 2 {
+					mappingFunctionIntArg, err := strconv.Atoi(strings.Trim(args[0], " "))
+					if err != nil {
+						return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+					}
+					var numPositions = len(args[1:])
+					tokenIndexes := make([]int, numPositions)
+					for ti, t := range args[1:] {
+						i, err := strconv.Atoi(strings.Trim(t, " "))
+						if err != nil {
+							return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+						}
+						tokenIndexes[ti] = i
+					}
+
+					return Partition, tokenIndexes, int32(mappingFunctionIntArg), _EMPTY_, nil
+				}
+			}
+
+			// SplitFromLeft(token, position)
+			args = getMappingFunctionArgs(splitFromLeftMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, SplitFromLeft)
+			}
+
+			// SplitFromRight(token, position)
+			args = getMappingFunctionArgs(splitFromRightMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, SplitFromRight)
+			}
+
+			// SliceFromLeft(token, position)
+			args = getMappingFunctionArgs(sliceFromLeftMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, SliceFromLeft)
+			}
+
+			// SliceFromRight(token, position)
+			args = getMappingFunctionArgs(sliceFromRightMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, SliceFromRight)
+			}
+
+			// split(token, deliminator)
+			args = getMappingFunctionArgs(splitMappingFunctionRegEx, token)
+			if args != nil {
+				if len(args) < 2 {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionNotEnoughArguments}
+				}
+				if len(args) > 2 {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionTooManyArguments}
+				}
+				i, err := strconv.Atoi(strings.Trim(args[0], " "))
+				if err != nil {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrorMappingDestinationFunctionInvalidArgument}
+				}
+				if strings.Contains(args[1], " ") || strings.Contains(args[1], tsep) {
+					return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token: token, err: ErrorMappingDestinationFunctionInvalidArgument}
+				}
+
+				return Split, []int{i}, -1, args[1], nil
+			}
+
+			return BadTransform, []int{}, -1, _EMPTY_, &mappingDestinationErr{token, ErrUnknownMappingDestinationFunction}
+		}
+	}
+	return NoTransform, []int{-1}, -1, _EMPTY_, nil
+}
+
+// Helper function to tokenize subjects with partial wildcards into formal transform destinations.
+// e.g. "foo.*.*" -> "foo.$1.$2"
+func transformTokenize(subject string) string {
+	// We need to make the appropriate markers for the wildcards etc.
+	i := 1
+	var nda []string
+	for _, token := range strings.Split(subject, tsep) {
+		if token == pwcs {
+			nda = append(nda, fmt.Sprintf("$%d", i))
+			i++
+		} else {
+			nda = append(nda, token)
+		}
+	}
+	return strings.Join(nda, tsep)
+}
+
+// Helper function to go from transform destination to a subject with partial wildcards and ordered list of placeholders
+// E.g.:
+//
+//	"bar" -> "bar", []
+//	"foo.$2.$1" -> "foo.*.*", ["$2","$1"]
+func transformUntokenize(subject string) (string, []string) {
+	var phs []string
+	var nda []string
+
+	for _, token := range strings.Split(subject, tsep) {
+		if len(token) > 1 && token[0] == '$' && token[1] >= '1' && token[1] <= '9' {
+			phs = append(phs, token)
+			nda = append(nda, pwcs)
+		} else {
+			nda = append(nda, token)
+		}
+	}
+	return strings.Join(nda, tsep), phs
+}
+
+func tokenizeSubject(subject string) []string {
+	// Tokenize the subject.
+	tsa := [32]string{}
+	tts := tsa[:0]
+	start := 0
+	for i := 0; i < len(subject); i++ {
+		if subject[i] == btsep {
+			tts = append(tts, subject[start:i])
+			start = i + 1
+		}
+	}
+	tts = append(tts, subject[start:])
+	return tts
+}
+
+// Match will take a literal published subject that is associated with a client and will match and subjectTransform
+// the subject if possible.
+//
+// This API is not part of the public API and not subject to SemVer protections
+func (tr *subjectTransform) Match(subject string) (string, error) {
+	// Special case: matches any and no no-op subjectTransform. May not be legal config for some features
+	// but specific validations made at subjectTransform create time
+	if (tr.src == fwcs || tr.src == _EMPTY_) && (tr.dest == fwcs || tr.dest == _EMPTY_) {
+		return subject, nil
+	}
+
+	tts := tokenizeSubject(subject)
+	if !isValidLiteralSubject(tts) {
+		return _EMPTY_, ErrBadSubject
+	}
+
+	if (tr.src == _EMPTY_ || tr.src == fwcs) || isSubsetMatch(tts, tr.src) {
+		return tr.TransformTokenizedSubject(tts), nil
+	}
+	return _EMPTY_, ErrNoTransforms
+}
+
+// TransformSubject transforms a subject
+//
+// This API is not part of the public API and not subject to SemVer protection
+func (tr *subjectTransform) TransformSubject(subject string) string {
+	return tr.TransformTokenizedSubject(tokenizeSubject(subject))
+}
+
+func (tr *subjectTransform) getHashPartition(key []byte, numBuckets int) string {
+	h := fnv.New32a()
+	_, _ = h.Write(key)
+
+	return strconv.Itoa(int(h.Sum32() % uint32(numBuckets)))
+}
+
+// Do a subjectTransform on the subject to the dest subject.
+func (tr *subjectTransform) TransformTokenizedSubject(tokens []string) string {
+	if len(tr.dtokmftypes) == 0 {
+		return tr.dest
+	}
+
+	var b strings.Builder
+
+	// We need to walk destination tokens and create the mapped subject pulling tokens or mapping functions
+	li := len(tr.dtokmftypes) - 1
+	for i, mfType := range tr.dtokmftypes {
+		if mfType == NoTransform {
+			// Break if fwc
+			if len(tr.dtoks[i]) == 1 && tr.dtoks[i][0] == fwc {
+				break
+			}
+			b.WriteString(tr.dtoks[i])
+		} else {
+			switch mfType {
+			case Partition:
+				var (
+					_buffer       [64]byte
+					keyForHashing = _buffer[:0]
+				)
+				for _, sourceToken := range tr.dtokmftokindexesargs[i] {
+					keyForHashing = append(keyForHashing, []byte(tokens[sourceToken])...)
+				}
+				b.WriteString(tr.getHashPartition(keyForHashing, int(tr.dtokmfintargs[i])))
+			case Wildcard: // simple substitution
+				b.WriteString(tokens[tr.dtokmftokindexesargs[i][0]])
+			case SplitFromLeft:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				position := int(tr.dtokmfintargs[i])
+				if position > 0 && position < sourceTokenLen {
+					b.WriteString(sourceToken[:position])
+					b.WriteString(tsep)
+					b.WriteString(sourceToken[position:])
+				} else { // too small to split at the requested position: don't split
+					b.WriteString(sourceToken)
+				}
+			case SplitFromRight:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				position := int(tr.dtokmfintargs[i])
+				if position > 0 && position < sourceTokenLen {
+					b.WriteString(sourceToken[:sourceTokenLen-position])
+					b.WriteString(tsep)
+					b.WriteString(sourceToken[sourceTokenLen-position:])
+				} else { // too small to split at the requested position: don't split
+					b.WriteString(sourceToken)
+				}
+			case SliceFromLeft:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				sliceSize := int(tr.dtokmfintargs[i])
+				if sliceSize > 0 && sliceSize < sourceTokenLen {
+					for i := 0; i+sliceSize <= sourceTokenLen; i += sliceSize {
+						if i != 0 {
+							b.WriteString(tsep)
+						}
+						b.WriteString(sourceToken[i : i+sliceSize])
+						if i+sliceSize != sourceTokenLen && i+sliceSize+sliceSize > sourceTokenLen {
+							b.WriteString(tsep)
+							b.WriteString(sourceToken[i+sliceSize:])
+							break
+						}
+					}
+				} else { // too small to slice at the requested size: don't slice
+					b.WriteString(sourceToken)
+				}
+			case SliceFromRight:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				sliceSize := int(tr.dtokmfintargs[i])
+				if sliceSize > 0 && sliceSize < sourceTokenLen {
+					remainder := sourceTokenLen % sliceSize
+					if remainder > 0 {
+						b.WriteString(sourceToken[:remainder])
+						b.WriteString(tsep)
+					}
+					for i := remainder; i+sliceSize <= sourceTokenLen; i += sliceSize {
+						b.WriteString(sourceToken[i : i+sliceSize])
+						if i+sliceSize < sourceTokenLen {
+							b.WriteString(tsep)
+						}
+					}
+				} else { // too small to slice at the requested size: don't slice
+					b.WriteString(sourceToken)
+				}
+			case Split:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				splits := strings.Split(sourceToken, tr.dtokmfstringargs[i])
+				for j, split := range splits {
+					if split != _EMPTY_ {
+						b.WriteString(split)
+					}
+					if j < len(splits)-1 && splits[j+1] != _EMPTY_ && !(j == 0 && split == _EMPTY_) {
+						b.WriteString(tsep)
+					}
+				}
+			}
+		}
+
+		if i < li {
+			b.WriteByte(btsep)
+		}
+	}
+
+	// We may have more source tokens available. This happens with ">".
+	if tr.dtoks[len(tr.dtoks)-1] == fwcs {
+		for sli, i := len(tokens)-1, len(tr.stoks)-1; i < len(tokens); i++ {
+			b.WriteString(tokens[i])
+			if i < sli {
+				b.WriteByte(btsep)
+			}
+		}
+	}
+	return b.String()
+}
+
+// Reverse a subjectTransform.
+func (tr *subjectTransform) reverse() *subjectTransform {
+	if len(tr.dtokmftokindexesargs) == 0 {
+		rtr, _ := newSubjectTransform(tr.dest, tr.src)
+		return rtr
+	}
+	// If we are here we need to dynamically get the correct reverse
+	// of this subjectTransform.
+	nsrc, phs := transformUntokenize(tr.dest)
+	var nda []string
+	for _, token := range tr.stoks {
+		if token == pwcs {
+			if len(phs) == 0 {
+				// TODO(dlc) - Should not happen
+				return nil
+			}
+			nda = append(nda, phs[0])
+			phs = phs[1:]
+		} else {
+			nda = append(nda, token)
+		}
+	}
+	ndest := strings.Join(nda, tsep)
+	rtr, _ := newSubjectTransform(nsrc, ndest)
+	return rtr
+}


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index 0142344b..4d3f7836 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -16,10 +16,8 @@ package server
 import (
 	"encoding/base64"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"net/http"
-	"reflect"
 	"strconv"
 	"strings"
 	"sync"
@@ -48,60 +46,6 @@ func simpleAccountServer(t *testing.T) (*Server, *Account, *Account) {
 	return s, f, b
 }
 
-func TestPlaceHolderIndex(t *testing.T) {
-	testString := "$1"
-	transformType, indexes, nbPartitions, _, err := indexPlaceHolders(testString)
-	var position int32
-
-	if err != nil || transformType != Wildcard || len(indexes) != 1 || indexes[0] != 1 || nbPartitions != -1 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-
-	testString = "{{partition(10,1,2,3)}}"
-
-	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
-
-	if err != nil || transformType != Partition || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-
-	testString = "{{ Partition (10,1,2,3) }}"
-
-	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
-
-	if err != nil || transformType != Partition || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-
-	testString = "{{wildcard(2)}}"
-	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
-
-	if err != nil || transformType != Wildcard || len(indexes) != 1 || indexes[0] != 2 || nbPartitions != -1 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-
-	testString = "{{SplitFromLeft(2,1)}}"
-	transformType, indexes, position, _, err = indexPlaceHolders(testString)
-
-	if err != nil || transformType != SplitFromLeft || len(indexes) != 1 || indexes[0] != 2 || position != 1 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-
-	testString = "{{SplitFromRight(3,2)}}"
-	transformType, indexes, position, _, err = indexPlaceHolders(testString)
-
-	if err != nil || transformType != SplitFromRight || len(indexes) != 1 || indexes[0] != 3 || position != 2 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-
-	testString = "{{SliceFromLeft(2,2)}}"
-	transformType, indexes, sliceSize, _, err := indexPlaceHolders(testString)
-
-	if err != nil || transformType != SliceFromLeft || len(indexes) != 1 || indexes[0] != 2 || sliceSize != 2 {
-		t.Fatalf("Error parsing %s", testString)
-	}
-}
-
 func TestRegisterDuplicateAccounts(t *testing.T) {
 	s, _, _ := simpleAccountServer(t)
 	if _, err := s.RegisterAccount("$foo"); err == nil {
@@ -3264,78 +3208,6 @@ func TestSamplingHeader(t *testing.T) {
 	test(false, http.Header{"traceparent": []string{"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00"}})
 }
 
-func TestSubjectTransforms(t *testing.T) {
-	shouldErr := func(src, dest string) {
-		t.Helper()
-		if _, err := newTransform(src, dest); err != ErrBadSubject && !errors.Is(err, ErrInvalidMappingDestination) {
-			t.Fatalf("Did not get an error for src=%q and dest=%q", src, dest)
-		}
-	}
-
-	shouldErr("foo.*.*", "bar.$2") // Must place all pwcs.
-
-	// Must be valid subjects.
-	shouldErr("foo", "")
-	shouldErr("foo..", "bar")
-
-	// Wildcards are allowed in src, but must be matched by token placements on the other side.
-	// e.g. foo.* -> bar.$1.
-	// Need to have as many pwcs as placements on other side.
-	shouldErr("foo.*", "bar.*")
-	shouldErr("foo.*", "bar.$2")                   // Bad pwc token identifier
-	shouldErr("foo.*", "bar.$1.>")                 // fwcs have to match.
-	shouldErr("foo.>", "bar.baz")                  // fwcs have to match.
-	shouldErr("foo.*.*", "bar.$2")                 // Must place all pwcs.
-	shouldErr("foo.*", "foo.$foo")                 // invalid $ value
-	shouldErr("foo.*", "foo.{{wildcard(2)}}")      // Mapping function being passed an out of range wildcard index
-	shouldErr("foo.*", "foo.{{unimplemented(1)}}") // Mapping trying to use an unknown mapping function
-	shouldErr("foo.*", "foo.{{partition(10)}}")    // Not enough arguments passed to the mapping function
-	shouldErr("foo.*", "foo.{{wildcard(foo)}}")    // Invalid argument passed to the mapping function
-	shouldErr("foo.*", "foo.{{wildcard()}}")       // Not enough arguments passed to the mapping function
-	shouldErr("foo.*", "foo.{{wildcard(1,2)}}")    // Too many arguments passed to the mapping function
-	shouldErr("foo.*", "foo.{{ wildcard5) }}")     // Bad mapping function
-	shouldErr("foo.*", "foo.{{splitLeft(2,2}}")    // arg out of range
-
-	shouldBeOK := func(src, dest string) *transform {
-		t.Helper()
-		tr, err := newTransform(src, dest)
-		if err != nil {
-			t.Fatalf("Got an error %v for src=%q and dest=%q", err, src, dest)
-		}
-		return tr
-	}
-
-	shouldBeOK("foo", "bar")
-	shouldBeOK("foo.*.bar.*.baz", "req.$2.$1")
-	shouldBeOK("baz.>", "mybaz.>")
-	shouldBeOK("*", "{{splitfromleft(1,1)}}")
-
-	shouldMatch := func(src, dest, sample, expected string) {
-		t.Helper()
-		tr := shouldBeOK(src, dest)
-		s, err := tr.Match(sample)
-		if err != nil {
-			t.Fatalf("Got an error %v when expecting a match for %q to %q", err, sample, expected)
-		}
-		if s != expected {
-			t.Fatalf("Dest does not match what was expected. Got %q, expected %q", s, expected)
-		}
-	}
-
-	shouldMatch("foo", "bar", "foo", "bar")
-	shouldMatch("foo.*.bar.*.baz", "req.$2.$1", "foo.A.bar.B.baz", "req.B.A")
-	shouldMatch("baz.>", "my.pre.>", "baz.1.2.3", "my.pre.1.2.3")
-	shouldMatch("baz.>", "foo.bar.>", "baz.1.2.3", "foo.bar.1.2.3")
-	shouldMatch("*", "foo.bar.$1", "foo", "foo.bar.foo")
-	shouldMatch("*", "{{splitfromleft(1,3)}}", "12345", "123.45")
-	shouldMatch("*", "{{SplitFromRight(1,3)}}", "12345", "12.345")
-	shouldMatch("*", "{{SliceFromLeft(1,3)}}", "1234567890", "123.456.789.0")
-	shouldMatch("*", "{{SliceFromRight(1,3)}}", "1234567890", "1.234.567.890")
-	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
-}
-
 func TestAccountSystemPermsWithGlobalAccess(t *testing.T) {
 	conf := createConfFile(t, []byte(`
 		listen: 127.0.0.1:-1
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index ad718dc7..e93116ce 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -5544,39 +5544,6 @@ func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
 	})
 }
 
-func TestJetStreamClusterSourceFilterSubjectUpdateFail(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "MSR", 3)
-	defer c.shutdown()
-
-	// Client for API requests.
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	// Origin
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:     "TEST",
-		Subjects: []string{"foo"},
-		Replicas: 2,
-	})
-	require_NoError(t, err)
-
-	_, err = js.AddStream(&nats.StreamConfig{
-		Name:     "S",
-		Sources:  []*nats.StreamSource{{Name: "TEST", FilterSubject: "notthere"}},
-		Replicas: 2,
-	})
-	require_Error(t, err)
-	require_Equal(t, err.Error(), "nats: source 'TEST' filter subject 'notthere' does not overlap with any origin stream subject")
-
-	_, err = js.AddStream(&nats.StreamConfig{
-		Name:     "M",
-		Mirror:   &nats.StreamSource{Name: "TEST", FilterSubject: "notthere"},
-		Replicas: 2,
-	})
-	require_Error(t, err)
-	require_Equal(t, err.Error(), "nats: mirror 'TEST' filter subject 'notthere' does not overlap with any origin stream subject")
-}
-
 func TestJetStreamClusterMirrorAndSourcesFilteredConsumers(t *testing.T) {
 	c := createJetStreamClusterWithTemplate(t, jsClusterMirrorSourceImportsTempl, "MS5", 5)
 	defer c.shutdown()
@@ -5625,9 +5592,8 @@ func TestJetStreamClusterMirrorAndSourcesFilteredConsumers(t *testing.T) {
 	createConsumer("M", "foo")
 	createConsumer("M", "bar")
 	createConsumer("M", "baz.foo")
-	expectFail("M", "baz")
-	expectFail("M", "baz.1.2")
-	expectFail("M", "apple")
+	expectFail("M", ".")
+	expectFail("M", ">.foo")
 
 	// Make sure wider scoped subjects work as well.
 	createConsumer("M", "*")
@@ -5652,9 +5618,6 @@ func TestJetStreamClusterMirrorAndSourcesFilteredConsumers(t *testing.T) {
 
 	createConsumer("S", "foo.1")
 	createConsumer("S", "bar.1")
-	expectFail("S", "baz")
-	expectFail("S", "baz.1")
-	expectFail("S", "apple")
 
 	// Now cross account stuff.
 	nc2, js2 := jsClientConnect(t, s, nats.UserInfo("rip", "pass"))
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 844a944f..3463a32b 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -530,15 +530,11 @@ func TestJetStreamConsumerWithStartTime(t *testing.T) {
 				OptStartTime:  &startTime,
 				AckPolicy:     AckExplicit,
 			})
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			msg, err := nc.Request(o.requestNextMsgSubject(), nil, time.Second)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
 			sseq, dseq, _, _, _ := replyInfo(msg.Reply)
 			if dseq != 1 {
 				t.Fatalf("Expected delivered seq of 1, got %d", dseq)
@@ -584,13 +580,9 @@ func TestJetStreamConsumerWithMultipleStartOptions(t *testing.T) {
 				},
 			}
 			req, err := json.Marshal(obsReq)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
 			_, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, subj), req, time.Second)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
 			nc.Close()
 			s.Shutdown()
 		})
@@ -635,9 +627,7 @@ func TestJetStreamConsumerMaxDeliveries(t *testing.T) {
 				AckWait:        ackWait,
 				MaxDeliver:     maxDeliver,
 			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			// Wait for redeliveries to pile up.
@@ -697,9 +687,7 @@ func TestJetStreamPullConsumerDelayedFirstPullWithReplayOriginal(t *testing.T) {
 				AckPolicy:    AckExplicit,
 				ReplayPolicy: ReplayOriginal,
 			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			// Force delay here which triggers the bug.
@@ -747,9 +735,7 @@ func TestJetStreamConsumerAckFloorFill(t *testing.T) {
 				DeliverSubject: sub.Subject,
 				AckPolicy:      AckExplicit,
 			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			var first *nats.Msg
@@ -805,9 +791,7 @@ func TestJetStreamNoPanicOnRaceBetweenShutdownAndConsumerDelete(t *testing.T) {
 					Durable:   fmt.Sprintf("d%d", i),
 					AckPolicy: AckExplicit,
 				})
-				if err != nil {
-					t.Fatalf("Expected no error, got %v", err)
-				}
+				require_NoError(t, err)
 				defer o.delete()
 				cons = append(cons, o)
 			}
@@ -870,9 +854,7 @@ func TestJetStreamAddStreamMaxMsgSize(t *testing.T) {
 
 			tooBig := []byte("1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ")
 			resp, err := nc.Request("foo", tooBig, time.Second)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
 			if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error.Description != "message size exceeds maximum allowed" {
 				t.Fatalf("Expected to get an error for maximum message size, got %q", pa.Error)
 			}
@@ -913,9 +895,7 @@ func TestJetStreamAddStreamBadSubjects(t *testing.T) {
 	expectAPIErr := func(cfg StreamConfig) {
 		t.Helper()
 		req, err := json.Marshal(cfg)
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
 		resp, _ := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
 		var scResp JSApiStreamCreateResponse
 		if err := json.Unmarshal(resp.Data, &scResp); err != nil {
@@ -950,9 +930,7 @@ func TestJetStreamMaxConsumers(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 	si, err := js.StreamInfo("MAXC")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if si.Config.MaxConsumers != 1 {
 		t.Fatalf("Expected max of 1, got %d", si.Config.MaxConsumers)
 	}
@@ -1762,9 +1740,7 @@ func TestJetStreamWorkQueueRequest(t *testing.T) {
 			jreq, _ = json.Marshal(req)
 
 			resp, err := nc.Request(getSubj, jreq, 100*time.Millisecond)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
 			if status := resp.Header.Get("Status"); !strings.HasPrefix(status, "404") {
 				t.Fatalf("Expected status code of 404")
 			}
@@ -1869,9 +1845,7 @@ func TestJetStreamSubjectFiltering(t *testing.T) {
 			// Now let's check the messages
 			for i := 1; i <= toSend; i++ {
 				m, err := sub.NextMsg(time.Second)
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
+				require_NoError(t, err)
 				// JetStream will have the subject match the stream subject, not delivery subject.
 				// We want these to only be subjB.
 				if m.Subject != subjB {
@@ -1941,9 +1915,7 @@ func TestJetStreamWorkQueueSubjectFiltering(t *testing.T) {
 			getNext := func(seqno int) {
 				t.Helper()
 				nextMsg, err := nc.Request(o.requestNextMsgSubject(), nil, time.Second)
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
+				require_NoError(t, err)
 				if nextMsg.Subject != subjA {
 					t.Fatalf("Expected subject of %q, got %q", subjA, nextMsg.Subject)
 				}
@@ -2264,9 +2236,7 @@ func TestJetStreamWorkQueueRetentionStream(t *testing.T) {
 
 			// We will create a non-partitioned consumer. This should succeed.
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			// Now if we create another this should fail, only can have one non-partitioned.
@@ -2287,16 +2257,12 @@ func TestJetStreamWorkQueueRetentionStream(t *testing.T) {
 				return &ConsumerConfig{Durable: dname, FilterSubject: pname, AckPolicy: AckExplicit}
 			}
 			o, err = mset.addConsumer(pConfig("MY_WORK_QUEUE.A"))
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			// Now creating another with separate partition should work.
 			o2, err := mset.addConsumer(pConfig("MY_WORK_QUEUE.B"))
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o2.delete()
 
 			// Anything that would overlap should fail though.
@@ -2308,9 +2274,7 @@ func TestJetStreamWorkQueueRetentionStream(t *testing.T) {
 			}
 
 			o3, err := mset.addConsumer(pConfig("MY_WORK_QUEUE.C"))
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 
 			o.delete()
 			o2.delete()
@@ -2485,9 +2449,7 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			}
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			expectPending := func(ep int) {
@@ -2554,9 +2516,7 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			// Now do filtered consumers.
 			o.delete()
 			o, err = mset.addConsumer(&ConsumerConfig{Durable: "PBO-FILTERED", AckPolicy: AckExplicit, FilterSubject: "foo.22"})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			for i := 0; i < toSend; i++ {
@@ -2637,9 +2597,7 @@ func TestJetStreamAckReplyStreamPendingWithAcks(t *testing.T) {
 				FilterSubject:  "foo",
 				DeliverSubject: dsubj,
 			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			if info := o.info(); int(info.NumPending) != toSend {
@@ -2701,9 +2659,7 @@ func TestJetStreamWorkQueueAckWaitRedelivery(t *testing.T) {
 			ackWait := 100 * time.Millisecond
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit, AckWait: ackWait})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			sub, _ := nc.SubscribeSync(nats.NewInbox())
@@ -2799,9 +2755,7 @@ func TestJetStreamWorkQueueNakRedelivery(t *testing.T) {
 			}
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			getMsg := func(sseq, dseq int) *nats.Msg {
@@ -2876,9 +2830,7 @@ func TestJetStreamWorkQueueWorkingIndicator(t *testing.T) {
 			ackWait := 100 * time.Millisecond
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit, AckWait: ackWait})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			getMsg := func(sseq, dseq int) *nats.Msg {
@@ -2959,9 +2911,7 @@ func TestJetStreamWorkQueueTerminateDelivery(t *testing.T) {
 			ackWait := 25 * time.Millisecond
 
 			o, err := mset.addConsumer(&ConsumerConfig{Durable: "PBO", AckPolicy: AckExplicit, AckWait: ackWait})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
 			defer o.delete()
 
 			getMsg := func(sseq, dseq int) *nats.Msg {
@@ -3341,9 +3291,7 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	m.Data = []byte("HELLO")
 	m.Header.Set(JSExpectedStream, mname)
 	resp, err := nc.RequestMsg(m, 100*time.Millisecond)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error != nil {
 		t.Fatalf("Expected a valid JetStreamPubAck, got %q", resp.Data)
 	}
@@ -3351,9 +3299,7 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	// Now test that we get an error back when expecting a different stream.
 	m.Header.Set(JSExpectedStream, "ORDERS")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
 		t.Fatalf("Expected an error, got %q", resp.Data)
 	}
@@ -3362,9 +3308,7 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	m.Header.Set(JSExpectedStream, mname)
 	m.Header.Set(JSExpectedLastSeq, "10")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
 		t.Fatalf("Expected an error, got %q", resp.Data)
 	}
@@ -3372,9 +3316,7 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	// Or if we expect that there are no messages by setting "0" as the expected last seq
 	m.Header.Set(JSExpectedLastSeq, "0")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
 		t.Fatalf("Expected an error, got %q", resp.Data)
 	}
@@ -3391,9 +3333,7 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	m.Header.Set(JSMsgId, "ZZZ")
 	m.Header.Set(JSExpectedLastMsgId, "BBB")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error == nil {
 		t.Fatalf("Expected an error, got %q", resp.Data)
 	}
@@ -3416,9 +3356,7 @@ func TestJetStreamPublishExpect(t *testing.T) {
 	m.Header.Set(JSExpectedLastMsgId, "AAA")
 	m.Header.Set(JSMsgId, "BBB")
 	resp, err = nc.RequestMsg(m, 100*time.Millisecond)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if pa := getPubAckResponse(resp.Data); pa == nil || pa.Error != nil {
 		t.Fatalf("Expected a valid JetStreamPubAck, got %q", resp.Data)
 	}
@@ -3469,9 +3407,7 @@ func TestJetStreamPullConsumerRemoveInterest(t *testing.T) {
 	sendStreamMsg(t, nc, mname, "Hello World!")
 
 	msg, err := nc.Request(rqn, nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	_, dseq, dc, _, _ := replyInfo(msg.Reply)
 	if dseq != 1 {
 		t.Fatalf("Expected consumer sequence of 1, got %d", dseq)
@@ -3495,9 +3431,7 @@ func TestJetStreamPullConsumerRemoveInterest(t *testing.T) {
 	sendStreamMsg(t, nc, mname, "Hello World!")
 
 	msg, err = nc.Request(rqn, nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	_, dseq, dc, _, _ = replyInfo(msg.Reply)
 	if dseq != 2 {
 		t.Fatalf("Expected consumer sequence of 2, got %d", dseq)
@@ -3550,9 +3484,7 @@ func TestJetStreamConsumerRateLimit(t *testing.T) {
 		DeliverSubject: "to",
 		RateLimit:      rateLimit,
 		AckPolicy:      AckNone})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	defer o.delete()
 
 	var received int
@@ -3898,9 +3830,7 @@ func TestJetStreamRedeliveryAfterServerRestart(t *testing.T) {
 		AckPolicy:      AckExplicit,
 		AckWait:        100 * time.Millisecond,
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	defer o.delete()
 
 	checkFor(t, 250*time.Millisecond, 10*time.Millisecond, func() error {
@@ -4016,9 +3946,7 @@ func TestJetStreamSnapshots(t *testing.T) {
 	r.Reset(snapshot)
 
 	mset, err = acc.RestoreStream(&info.cfg, r)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	// Now compare to make sure they are equal.
 	if nusage := acc.JetStreamUsage(); !reflect.DeepEqual(nusage, pusage) {
 		t.Fatalf("Usage does not match after restore: %+v vs %+v", nusage, pusage)
@@ -4050,9 +3978,7 @@ func TestJetStreamSnapshots(t *testing.T) {
 	acc = s2.GlobalAccount()
 	r.Reset(snapshot)
 	mset, err = acc.RestoreStream(&info.cfg, r)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	o := mset.lookupConsumer("WQ-1")
 	if o == nil {
@@ -4126,9 +4052,7 @@ func TestJetStreamSnapshotsAPI(t *testing.T) {
 	}
 
 	o, err := mset.addConsumer(workerModeConfig("WQ"))
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	// Now grab some messages.
 	toReceive := rand.Intn(toSend) + 1
 	for r := 0; r < toReceive; r++ {
@@ -5412,9 +5336,7 @@ func TestJetStreamRedeliverAndLateAck(t *testing.T) {
 
 	nextSubj := o.requestNextMsgSubject()
 	msg, err := nc.Request(nextSubj, nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	// Wait for past ackwait time
 	time.Sleep(150 * time.Millisecond)
@@ -7271,9 +7193,7 @@ func TestJetStreamPushConsumerFlowControl(t *testing.T) {
 	}
 
 	sub, err := nc.SubscribeSync(nats.NewInbox())
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	defer sub.Unsubscribe()
 
 	obsReq := CreateConsumerRequest{
@@ -7286,13 +7206,9 @@ func TestJetStreamPushConsumerFlowControl(t *testing.T) {
 		},
 	}
 	req, err := json.Marshal(obsReq)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, err := nc.Request(fmt.Sprintf(JSApiDurableCreateT, "TEST", "dlc"), req, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7391,9 +7307,7 @@ func TestJetStreamPushConsumerIdleHeartbeats(t *testing.T) {
 	}
 
 	sub, err := nc.SubscribeSync(nats.NewInbox())
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	defer sub.Unsubscribe()
 
 	// Test errors first
@@ -7405,13 +7319,9 @@ func TestJetStreamPushConsumerIdleHeartbeats(t *testing.T) {
 		},
 	}
 	req, err := json.Marshal(obsReq)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, err := nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "TEST"), req, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7422,13 +7332,9 @@ func TestJetStreamPushConsumerIdleHeartbeats(t *testing.T) {
 	// Set acceptable heartbeat.
 	obsReq.Config.Heartbeat = 100 * time.Millisecond
 	req, err = json.Marshal(obsReq)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "TEST"), req, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	ccResp.Error = nil
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7462,9 +7368,7 @@ func TestJetStreamPushConsumerIdleHeartbeatsWithFilterSubject(t *testing.T) {
 
 	hbC := make(chan *nats.Msg, 8)
 	sub, err := nc.ChanSubscribe(nats.NewInbox(), hbC)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	defer sub.Unsubscribe()
 
 	obsReq := CreateConsumerRequest{
@@ -7477,13 +7381,9 @@ func TestJetStreamPushConsumerIdleHeartbeatsWithFilterSubject(t *testing.T) {
 	}
 
 	req, err := json.Marshal(obsReq)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, err := nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "TEST"), req, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7522,9 +7422,7 @@ func TestJetStreamPushConsumerIdleHeartbeatsWithNoInterest(t *testing.T) {
 	dsubj := "d.22"
 	hbC := make(chan *nats.Msg, 8)
 	sub, err := nc.ChanSubscribe("d.>", hbC)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	defer sub.Unsubscribe()
 
 	obsReq := CreateConsumerRequest{
@@ -7536,13 +7434,9 @@ func TestJetStreamPushConsumerIdleHeartbeatsWithNoInterest(t *testing.T) {
 	}
 
 	req, err := json.Marshal(obsReq)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, err := nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "TEST"), req, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7580,9 +7474,7 @@ func TestJetStreamInfoAPIWithHeaders(t *testing.T) {
 	m.Data = []byte("HELLO-JS!")
 
 	resp, err := nc.RequestMsg(m, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	var info JSApiAccountInfoResponse
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
@@ -7617,9 +7509,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		MaxMsgs:  100,
 	}
 	req, err := json.Marshal(msetCfg)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, _ = nc.Request(fmt.Sprintf(JSApiStreamCreateT, msetCfg.Name), req, time.Second)
 	var scResp JSApiStreamCreateResponse
 	if err := json.Unmarshal(resp.Data, &scResp); err != nil {
@@ -7644,9 +7534,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	msetCfg.Subjects = []string{"foo", "bar", "baz"}
 	msetCfg.MaxBytes = 2222222
 	req, err = json.Marshal(msetCfg)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, _ = nc.Request(fmt.Sprintf(JSApiStreamUpdateT, msetCfg.Name), req, time.Second)
 	scResp.Error, scResp.StreamInfo = nil, nil
 	if err := json.Unmarshal(resp.Data, &scResp); err != nil {
@@ -7663,9 +7551,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		Subjects: []string{"foo"},
 	}
 	req, err = json.Marshal(cfg)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, _ = nc.Request(fmt.Sprintf(JSApiStreamUpdateT, cfg.Name), req, time.Second)
 	scResp.Error, scResp.StreamInfo = nil, nil
 	if err := json.Unmarshal(resp.Data, &scResp); err != nil {
@@ -7677,9 +7563,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Now lookup info again and see that we can see the new stream.
 	resp, err = nc.Request(JSApiAccountInfo, nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if err = json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -7689,9 +7573,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Make sure list names works.
 	resp, err = nc.Request(JSApiStreams, nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var namesResponse JSApiStreamNamesResponse
 	if err = json.Unmarshal(resp.Data, &namesResponse); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7744,9 +7626,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	}
 
 	resp, err = nc.Request(fmt.Sprintf(JSApiStreamInfoT, msetCfg.Name), nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var msi StreamInfo
 	if err = json.Unmarshal(resp.Data, &msi); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7760,9 +7640,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Looking up one that is not there should yield an error.
 	resp, err = nc.Request(fmt.Sprintf(JSApiStreamInfoT, "BOB"), nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var bResp JSApiStreamInfoResponse
 	if err = json.Unmarshal(resp.Data, &bResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7776,13 +7654,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		Config: ConsumerConfig{DeliverSubject: delivery},
 	}
 	req, err = json.Marshal(obsReq)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, msetCfg.Name), req, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var ccResp JSApiConsumerCreateResponse
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7808,9 +7682,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Check that we get an error if the stream name in the subject does not match the config.
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, "BOB"), req, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	ccResp.Error, ccResp.ConsumerInfo = nil, nil
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7820,9 +7692,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 
 	// Get the list of all of the consumers for our stream.
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumersT, msetCfg.Name), nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var clResponse JSApiConsumerNamesResponse
 	if err = json.Unmarshal(resp.Data, &clResponse); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7833,9 +7703,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	// Now let's get info about our consumer.
 	cName := clResponse.Consumers[0]
 	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerInfoT, msetCfg.Name, cName), nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var oinfo ConsumerInfo
 	if err = json.Unmarshal(resp.Data, &oinfo); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7875,15 +7743,11 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		Config: ConsumerConfig{Durable: "myd", DeliverSubject: delivery},
 	}
 	req, err = json.Marshal(obsReq)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, msetCfg.Name), req, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	ccResp.Error, ccResp.ConsumerInfo = nil, nil
-	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
+	require_NoError(t, err)
+	resp, err = nc.Request(fmt.Sprintf(JSApiConsumerCreateT, msetCfg.Name), req, time.Second)
+	require_NoError(t, err)
+	ccResp.Error, ccResp.ConsumerInfo = nil, nil
+	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 	checkNatsError(t, ccResp.Error, JSConsumerEphemeralWithDurableNameErr)
@@ -7904,13 +7768,9 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		Config: ConsumerConfig{DeliverSubject: delivery},
 	}
 	req2, err := json.Marshal(obsReq2)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, err = nc.Request(fmt.Sprintf(JSApiDurableCreateT, msetCfg.Name, obsReq.Config.Durable), req2, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	ccResp.Error, ccResp.ConsumerInfo = nil, nil
 	if err = json.Unmarshal(resp.Data, &ccResp); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -7920,9 +7780,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	// Now delete a msg.
 	dreq := JSApiMsgDeleteRequest{Seq: 2}
 	dreqj, err := json.Marshal(dreq)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	resp, _ = nc.Request(fmt.Sprintf(JSApiMsgDeleteT, msetCfg.Name), dreqj, time.Second)
 	var delMsgResp JSApiMsgDeleteResponse
 	if err = json.Unmarshal(resp.Data, &delMsgResp); err != nil {
@@ -7958,9 +7816,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	// Now grab stats again.
 	// This will get the current information about usage and limits for this account.
 	resp, err = nc.Request(JSApiAccountInfo, nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -7983,9 +7839,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 		MaxStreams: 4,
 	}
 	req, err = json.Marshal(template)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	// Check that the name in config has to match the name in the subject
 	resp, _ = nc.Request(fmt.Sprintf(JSApiTemplateCreateT, "BOB"), req, time.Second)
@@ -8073,9 +7927,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	sendStreamMsg(t, nc, "kv.22", "derek")
 	// Last do info
 	resp, err = nc.Request(fmt.Sprintf(JSApiTemplateInfoT, "kv"), nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var ti StreamTemplateInfo
 	if err = json.Unmarshal(resp.Data, &ti); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -8911,9 +8763,7 @@ func TestJetStreamTemplateFileStoreRecovery(t *testing.T) {
 		t.Fatalf("Expected 100 auto-created streams, got %d", nms)
 	}
 	tmpl, err := acc.lookupStreamTemplate(template.Name)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	// Make sure t.delete() survives restart.
 	tmpl.delete()
 
@@ -9083,9 +8933,7 @@ func TestJetStreamMultipleAccountsBasics(t *testing.T) {
 	defer ncb.Close()
 
 	resp, err := ncb.Request(JSApiAccountInfo, nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var info JSApiAccountInfoResponse
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -9174,9 +9022,7 @@ func TestJetStreamMultipleAccountsBasics(t *testing.T) {
 	// Now check that limits have been updated.
 	// Account B
 	resp, err = ncb.Request(JSApiAccountInfo, nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -9196,9 +9042,7 @@ func TestJetStreamMultipleAccountsBasics(t *testing.T) {
 
 	// Account C
 	resp, err = ncc.Request(JSApiAccountInfo, nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -9268,9 +9112,7 @@ func TestJetStreamStoreDirectoryFix(t *testing.T) {
 	}
 	// Push based.
 	sub, err := js.SubscribeSync("TEST", nats.Durable("dlc"))
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	defer sub.Unsubscribe()
 
 	// Now shutdown the server.
@@ -9324,20 +9166,14 @@ func TestJetStreamPushConsumersPullError(t *testing.T) {
 	}
 	// Push based.
 	sub, err := js.SubscribeSync("TEST")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	defer sub.Unsubscribe()
 	ci, err := sub.ConsumerInfo()
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	// Now do a pull. Make sure we get an error.
 	m, err := nc.Request(fmt.Sprintf(JSApiRequestNextT, "TEST", ci.Name), nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if m.Header.Get("Status") != "409" {
 		t.Fatalf("Expected a 409 status code, got %q", m.Header.Get("Status"))
 	}
@@ -9534,9 +9370,7 @@ func TestJetStreamPubWithSyncPerf(t *testing.T) {
 	defer nc.Close()
 
 	_, err := js.AddStream(&nats.StreamConfig{Name: "foo"})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	toSend := 1_000_000
 	payload := []byte("Hello World")
@@ -10244,9 +10078,8 @@ func TestJetStreamConsumerMaxAckPending(t *testing.T) {
 				AckPolicy:      AckExplicit,
 				MaxAckPending:  maxAckPending,
 			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
+
 			defer o.delete()
 
 			sub, _ := nc.SubscribeSync(o.info().Config.DeliverSubject)
@@ -10288,9 +10121,8 @@ func TestJetStreamConsumerMaxAckPending(t *testing.T) {
 				AckPolicy:      AckExplicit,
 				MaxAckPending:  maxAckPending,
 			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
+
 			defer o.delete()
 
 			sub, _ = nc.SubscribeSync(o.info().Config.DeliverSubject)
@@ -10356,9 +10188,8 @@ func TestJetStreamPullConsumerMaxAckPending(t *testing.T) {
 				AckPolicy:     AckExplicit,
 				MaxAckPending: maxAckPending,
 			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
+
 			defer o.delete()
 
 			getSubj := o.requestNextMsgSubject()
@@ -10454,9 +10285,8 @@ func TestJetStreamPullConsumerMaxAckPendingRedeliveries(t *testing.T) {
 				AckWait:       ackWait,
 				MaxAckPending: maxAckPending,
 			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
+			require_NoError(t, err)
+
 			defer o.delete()
 
 			getSubj := o.requestNextMsgSubject()
@@ -10680,9 +10510,7 @@ func TestJetStreamAccountImportBasics(t *testing.T) {
 
 	// We mapped the next message request, "$JS.API.CONSUMER.MSG.NEXT.ORDERS.d" -> "nxt.msg"
 	m, err := nc.Request("nxt.msg", nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if string(m.Data) != "ORDERS-1" {
 		t.Fatalf("Expected to receive %q, got %q", "ORDERS-1", m.Data)
 	}
@@ -10784,9 +10612,7 @@ func TestJetStreamAccountImportAll(t *testing.T) {
 
 	// This will get the current information about usage and limits for this account.
 	resp, err := nc.Request(mapSubj(JSApiAccountInfo), nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var info JSApiAccountInfoResponse
 	if err := json.Unmarshal(resp.Data, &info); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -10796,9 +10622,7 @@ func TestJetStreamAccountImportAll(t *testing.T) {
 	}
 	// Lookup streams.
 	resp, err = nc.Request(mapSubj(JSApiStreams), nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var namesResponse JSApiStreamNamesResponse
 	if err = json.Unmarshal(resp.Data, &namesResponse); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -10858,9 +10682,7 @@ func TestJetStreamServerReload(t *testing.T) {
 		t.Fatalf("Unexpected error looking up account: %v", err)
 	}
 	mset, err := acc.addStream(&StreamConfig{Name: "22"})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	toSend := 10
 	for i := 0; i < toSend; i++ {
@@ -10930,9 +10752,7 @@ func TestJetStreamConfigReloadWithGlobalAccount(t *testing.T) {
 		}
 	}
 	si, err := js.StreamInfo("foo")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	if si.State.Msgs != uint64(toSend) {
 		t.Fatalf("Expected %d msgs after restart, got %d", toSend, si.State.Msgs)
 	}
@@ -11268,9 +11088,7 @@ func TestJetStreamFilteredConsumersWithWiderFilter(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo", "bar", "baz", "N.*"},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	// Add in some messages.
 	js.Publish("foo", []byte("OK"))
@@ -11282,9 +11100,8 @@ func TestJetStreamFilteredConsumersWithWiderFilter(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js.StreamInfo("TEST")
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
+
 		if si.State.Msgs != 15 {
 			return fmt.Errorf("Expected 15 msgs, got state: %+v", si.State)
 		}
@@ -11293,9 +11110,8 @@ func TestJetStreamFilteredConsumersWithWiderFilter(t *testing.T) {
 
 	checkWider := func(subj string, numExpected int) {
 		sub, err := js.SubscribeSync(subj)
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
+
 		defer sub.Unsubscribe()
 		checkSubsPending(t, sub, numExpected)
 	}
@@ -11320,17 +11136,14 @@ func TestJetStreamMirrorAndSourcesFilteredConsumers(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo", "bar", "baz.*"},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
+
 	// Create Mirror now.
 	_, err = js.AddStream(&nats.StreamConfig{
 		Name:   "M",
 		Mirror: &nats.StreamSource{Name: "TEST"},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	dsubj := nats.NewInbox()
 	nc.SubscribeSync(dsubj)
@@ -11339,24 +11152,13 @@ func TestJetStreamMirrorAndSourcesFilteredConsumers(t *testing.T) {
 	createConsumer := func(sn, fs string) {
 		t.Helper()
 		_, err = js.AddConsumer(sn, &nats.ConsumerConfig{DeliverSubject: dsubj, FilterSubject: fs})
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-	}
-	expectFail := func(sn, fs string) {
-		t.Helper()
-		_, err = js.AddConsumer(sn, &nats.ConsumerConfig{DeliverSubject: dsubj, FilterSubject: fs})
-		if err == nil {
-			t.Fatalf("Expected error but got none")
-		}
+		require_NoError(t, err)
+
 	}
 
 	createConsumer("M", "foo")
 	createConsumer("M", "bar")
 	createConsumer("M", "baz.foo")
-	expectFail("M", "baz")
-	expectFail("M", "baz.1.2")
-	expectFail("M", "apple")
 
 	// Now do some sources.
 	if _, err := js.AddStream(&nats.StreamConfig{Name: "O1", Subjects: []string{"foo.*"}}); err != nil {
@@ -11371,15 +11173,10 @@ func TestJetStreamMirrorAndSourcesFilteredConsumers(t *testing.T) {
 		Name:    "S",
 		Sources: []*nats.StreamSource{{Name: "O1"}, {Name: "O2"}},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	createConsumer("S", "foo.1")
 	createConsumer("S", "bar.1")
-	expectFail("S", "baz")
-	expectFail("S", "baz.1")
-	expectFail("S", "apple")
 
 	// Chaining
 	// Create Mirror now.
@@ -11387,16 +11184,11 @@ func TestJetStreamMirrorAndSourcesFilteredConsumers(t *testing.T) {
 		Name:   "M2",
 		Mirror: &nats.StreamSource{Name: "M"},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	createConsumer("M2", "foo")
 	createConsumer("M2", "bar")
 	createConsumer("M2", "baz.foo")
-	expectFail("M2", "baz")
-	expectFail("M2", "baz.1.2")
-	expectFail("M2", "apple")
 }
 
 func TestJetStreamMirrorBasics(t *testing.T) {
@@ -11456,14 +11248,12 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 
 	// Faster timeout since we loop below checking for condition.
 	js2, err := nc.JetStream(nats.MaxWait(250 * time.Millisecond))
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
+
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("M1")
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
+
 		if si.State.Msgs != 100 {
 			return fmt.Errorf("Expected 100 msgs, got state: %+v", si.State)
 		}
@@ -11492,9 +11282,8 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("M2")
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
+
 		if si.State.Msgs != 50 {
 			return fmt.Errorf("Expected 50 msgs, got state: %+v", si.State)
 		}
@@ -11520,9 +11309,8 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("M3")
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
+
 		if si.State.Msgs != 101 {
 			return fmt.Errorf("Expected 101 msgs, got state: %+v", si.State)
 		}
@@ -11542,9 +11330,8 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("M4")
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
+
 		if si.State.Msgs != 150 {
 			return fmt.Errorf("Expected 150 msgs, got state: %+v", si.State)
 		}
@@ -11586,13 +11373,11 @@ func TestJetStreamSourceBasics(t *testing.T) {
 	createStream := func(cfg *StreamConfig) {
 		t.Helper()
 		req, err := json.Marshal(cfg)
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
+
 		rm, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
+
 		var resp JSApiStreamCreateResponse
 		if err := json.Unmarshal(rm.Data, &resp); err != nil {
 			t.Fatalf("Unexpected error: %v", err)
@@ -11623,7 +11408,7 @@ func TestJetStreamSourceBasics(t *testing.T) {
 		Name:    "MS",
 		Storage: FileStorage,
 		Sources: []*StreamSource{
-			{Name: "foo"},
+			{Name: "foo", SubjectTransformDest: "foo2.>"},
 			{Name: "bar"},
 			{Name: "baz"},
 		},
@@ -11633,20 +11418,25 @@ func TestJetStreamSourceBasics(t *testing.T) {
 
 	// Faster timeout since we loop below checking for condition.
 	js2, err := nc.JetStream(nats.MaxWait(250 * time.Millisecond))
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
+
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("MS")
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
+
 		if si.State.Msgs != 50 {
 			return fmt.Errorf("Expected 50 msgs, got state: %+v", si.State)
 		}
 		return nil
 	})
 
+	m, err := js.GetMsg("MS", 1)
+	require_NoError(t, err)
+
+	if m.Subject != "foo2.foo" {
+		t.Fatalf("Expected message subject foo2.foo, got %s", m.Subject)
+	}
+
 	// Test Source Updates
 	ncfg := &nats.StreamConfig{
 		Name: "MS",
@@ -11679,19 +11469,15 @@ func TestJetStreamSourceBasics(t *testing.T) {
 	createStream(cfg)
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("FMS")
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
 		if si.State.Msgs != 25 {
 			return fmt.Errorf("Expected 25 msgs, got state: %+v", si.State)
 		}
 		return nil
 	})
 	// Double check first starting.
-	m, err := js.GetMsg("FMS", 1)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	m, err = js.GetMsg("FMS", 1)
+	require_NoError(t, err)
 	if shdr := m.Header.Get(JSStreamSource); shdr == _EMPTY_ {
 		t.Fatalf("Expected a header, got none")
 	} else if _, sseq := streamAndSeq(shdr); sseq != 26 {
@@ -11709,9 +11495,7 @@ func TestJetStreamSourceBasics(t *testing.T) {
 	createStream(cfg)
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		si, err := js2.StreamInfo("FMS2")
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
+		require_NoError(t, err)
 		if si.State.Msgs != 20 {
 			return fmt.Errorf("Expected 20 msgs, got state: %+v", si.State)
 		}
@@ -11729,6 +11513,46 @@ func TestJetStreamSourceBasics(t *testing.T) {
 	}
 }
 
+func TestJetStreamInputTransform(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	createStream := func(cfg *StreamConfig) {
+		t.Helper()
+		req, err := json.Marshal(cfg)
+		require_NoError(t, err)
+
+		rm, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
+		require_NoError(t, err)
+
+		var resp JSApiStreamCreateResponse
+		if err := json.Unmarshal(rm.Data, &resp); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if resp.Error != nil {
+			t.Fatalf("Unexpected error: %+v", resp.Error)
+		}
+	}
+
+	createStream(&StreamConfig{Name: "T1", Subjects: []string{"foo"}, SubjectTransform: &SubjectTransformConfig{Source: ">", Destination: "transformed.>"}, Storage: MemoryStorage})
+
+	// publish a message
+	if _, err := js.Publish("foo", []byte("OK")); err != nil {
+		t.Fatalf("Unexpected publish error: %v", err)
+	}
+
+	m, err := js.GetMsg("T1", 1)
+	require_NoError(t, err)
+
+	if m.Subject != "transformed.foo" {
+		t.Fatalf("Expected message subject transformed.foo, got %s", m.Subject)
+	}
+}
+
 func TestJetStreamOperatorAccounts(t *testing.T) {
 	s, _ := RunServerWithConfig("./configs/js-op.conf")
 	if config := s.JetStreamConfig(); config != nil {
@@ -11759,9 +11583,7 @@ func TestJetStreamOperatorAccounts(t *testing.T) {
 	defer s.Shutdown()
 
 	jsz, err := s.Jsz(nil)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	if jsz.Streams != 1 {
 		t.Fatalf("Expected jsz to report our stream on restart")
@@ -11859,9 +11681,7 @@ func TestJetStreamDomainInPubAck(t *testing.T) {
 
 	// Check by hand for now til it makes its way into Go client.
 	am, err := nc.Request("foo", nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	var pa PubAck
 	json.Unmarshal(am.Data, &pa)
 	if pa.Domain != "HUB" {
@@ -11882,9 +11702,7 @@ func TestJetStreamDirectConsumersBeingReported(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo"},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	_, err = js.AddStream(&nats.StreamConfig{
 		Name: "S",
@@ -11892,9 +11710,7 @@ func TestJetStreamDirectConsumersBeingReported(t *testing.T) {
 			Name: "TEST",
 		}},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	if _, err = js.Publish("foo", nil); err != nil {
 		t.Fatalf("Unexpected publish error: %v", err)
@@ -11911,9 +11727,7 @@ func TestJetStreamDirectConsumersBeingReported(t *testing.T) {
 	})
 
 	si, err := js.StreamInfo("TEST")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	// Direct consumers should not be reported
 	if si.State.Consumers != 0 {
@@ -11952,9 +11766,7 @@ func TestJetStreamTemplatedErrorsBug(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo"},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	_, err = js.PullSubscribe("foo", "")
 	if err != nil && strings.Contains(err.Error(), "{err}") {
@@ -12070,9 +11882,8 @@ func TestJetStreamServerEncryption(t *testing.T) {
 				checkFor(filepath.Join(sdir, "obs", "dlc", JetStreamMetaFile), "TEST", "dlc", "foo", "bar", "baz", "max_msgs", "ack_policy")
 				// Load and see if we can parse the consumer state.
 				state, err := os.ReadFile(filepath.Join(sdir, "obs", "dlc", "o.dat"))
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
+				require_NoError(t, err)
+
 				if _, err := decodeConsumerState(state); err == nil {
 					t.Fatalf("Expected decoding consumer state to fail")
 				}
@@ -12092,9 +11903,8 @@ func TestJetStreamServerEncryption(t *testing.T) {
 			defer nc.Close()
 
 			si2, err := js.StreamInfo("TEST")
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
+
 			if !reflect.DeepEqual(si, si2) {
 				t.Fatalf("Stream infos did not match\n%+v\nvs\n%+v", si, si2)
 			}
@@ -12125,9 +11935,7 @@ func TestJetStreamServerEncryption(t *testing.T) {
 			// Now test snapshots etc.
 			acc := s.GlobalAccount()
 			mset, err := acc.lookupStream("TEST")
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
 			scfg := mset.config()
 			sr, err := mset.snapshot(5*time.Second, false, true)
 			if err != nil {
@@ -12145,9 +11953,7 @@ func TestJetStreamServerEncryption(t *testing.T) {
 			nacc := ns.GlobalAccount()
 			r := bytes.NewReader(snapshot)
 			mset, err = nacc.RestoreStream(&scfg, r)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
 			ss := mset.store.State()
 			if ss.Msgs != si.State.Msgs || ss.FirstSeq != si.State.FirstSeq || ss.LastSeq != si.State.LastSeq {
 				t.Fatalf("Stream states do not match: %+v vs %+v", ss, si.State)
@@ -12161,9 +11967,7 @@ func TestJetStreamServerEncryption(t *testing.T) {
 			acc = s.GlobalAccount()
 			r.Reset(snapshot)
 			mset, err = acc.RestoreStream(&scfg, r)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
+			require_NoError(t, err)
 			ss = mset.store.State()
 			if ss.Msgs != si.State.Msgs || ss.FirstSeq != si.State.FirstSeq || ss.LastSeq != si.State.LastSeq {
 				t.Fatalf("Stream states do not match: %+v vs %+v", ss, si.State)
@@ -12188,9 +11992,7 @@ func TestJetStreamConsumerBadNumPending(t *testing.T) {
 		Name:     "ORDERS",
 		Subjects: []string{"orders.*"},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	newOrders := func(n int) {
 		// Queue up new orders.
@@ -12207,15 +12009,13 @@ func TestJetStreamConsumerBadNumPending(t *testing.T) {
 	}
 
 	op, err := js.Subscribe("orders.created", process)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
+
 	defer op.Unsubscribe()
 
 	mon, err := js.SubscribeSync("orders.*")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
+
 	defer mon.Unsubscribe()
 
 	waitForMsgs := func(n uint64) {
@@ -12450,14 +12250,10 @@ func TestJetStreamConsumerCleanupWithRetentionPolicy(t *testing.T) {
 		Subjects:  []string{"orders.*"},
 		Retention: nats.InterestPolicy,
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	sub, err := js.SubscribeSync("orders.*")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	payload := []byte("Hello World")
 	for i := 0; i < 10; i++ {
@@ -12482,9 +12278,8 @@ func TestJetStreamConsumerCleanupWithRetentionPolicy(t *testing.T) {
 
 	acc := s.GlobalAccount()
 	mset, err := acc.lookupStream("ORDERS")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
+
 	o := mset.lookupConsumer(ci.Name)
 	if o == nil {
 		t.Fatalf("Error looking up consumer %q", ci.Name)
@@ -12515,16 +12310,13 @@ func TestJetStreamPurgeEffectsConsumerDelivery(t *testing.T) {
 		Name:     "TEST",
 		Subjects: []string{"foo.*"},
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	js.Publish("foo.a", []byte("show once"))
 
 	sub, err := js.SubscribeSync("foo.*", nats.AckWait(250*time.Millisecond), nats.DeliverAll(), nats.AckExplicit())
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
+
 	defer sub.Unsubscribe()
 
 	checkSubsPending(t, sub, 1)
@@ -12566,14 +12358,11 @@ func TestJetStreamExpireCausesDeadlock(t *testing.T) {
 		MaxMsgs:   10,
 		Retention: nats.InterestPolicy,
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	sub, err := js.SubscribeSync("foo.bar")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
+
 	defer sub.Unsubscribe()
 
 	// Publish from two connections to get the write lock request wedged in between
@@ -12674,9 +12463,8 @@ func TestJetStreamDefaultMaxMsgsPer(t *testing.T) {
 		Storage:  nats.MemoryStorage,
 		MaxMsgs:  10,
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
+
 	if si.Config.MaxMsgsPerSubject != -1 {
 		t.Fatalf("Expected default of -1, got %d", si.Config.MaxMsgsPerSubject)
 	}
@@ -12696,9 +12484,7 @@ func TestJetStreamBadConsumerCreateErr(t *testing.T) {
 		Subjects: []string{"foo.*"},
 		Storage:  nats.MemoryStorage,
 	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	// When adding a consumer with both deliver subject and max wait (push vs pull),
 	// we got the wrong err about deliver subject having a wildcard.
diff --git a/server/subject_transform_test.go b/server/subject_transform_test.go
new file mode 100644
index 00000000..8ac9b6ca
--- /dev/null
+++ b/server/subject_transform_test.go
@@ -0,0 +1,148 @@
+// 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
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"errors"
+	"reflect"
+	"testing"
+)
+
+func TestPlaceHolderIndex(t *testing.T) {
+	testString := "$1"
+	transformType, indexes, nbPartitions, _, err := indexPlaceHolders(testString)
+	var position int32
+
+	if err != nil || transformType != Wildcard || len(indexes) != 1 || indexes[0] != 1 || nbPartitions != -1 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	testString = "{{partition(10,1,2,3)}}"
+
+	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != Partition || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	testString = "{{ Partition (10,1,2,3) }}"
+
+	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != Partition || !reflect.DeepEqual(indexes, []int{1, 2, 3}) || nbPartitions != 10 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	testString = "{{wildcard(2)}}"
+	transformType, indexes, nbPartitions, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != Wildcard || len(indexes) != 1 || indexes[0] != 2 || nbPartitions != -1 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	testString = "{{SplitFromLeft(2,1)}}"
+	transformType, indexes, position, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != SplitFromLeft || len(indexes) != 1 || indexes[0] != 2 || position != 1 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	testString = "{{SplitFromRight(3,2)}}"
+	transformType, indexes, position, _, err = indexPlaceHolders(testString)
+
+	if err != nil || transformType != SplitFromRight || len(indexes) != 1 || indexes[0] != 3 || position != 2 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+
+	testString = "{{SliceFromLeft(2,2)}}"
+	transformType, indexes, sliceSize, _, err := indexPlaceHolders(testString)
+
+	if err != nil || transformType != SliceFromLeft || len(indexes) != 1 || indexes[0] != 2 || sliceSize != 2 {
+		t.Fatalf("Error parsing %s", testString)
+	}
+}
+
+func TestSubjectTransforms(t *testing.T) {
+	shouldErr := func(src, dest string) {
+		t.Helper()
+		if _, err := newSubjectTransform(src, dest); err != ErrBadSubject && !errors.Is(err, ErrInvalidMappingDestination) {
+			t.Fatalf("Did not get an error for src=%q and dest=%q", src, dest)
+		}
+	}
+
+	shouldErr("foo.*.*", "bar.$2") // Must place all pwcs.
+
+	// Must be valid subjects.
+	shouldErr("foo", "")
+	shouldErr("foo..", "bar")
+
+	// Wildcards are allowed in src, but must be matched by token placements on the other side.
+	// e.g. foo.* -> bar.$1.
+	// Need to have as many pwcs as placements on other side.
+	shouldErr("foo.*", "bar.*")
+	shouldErr("foo.*", "bar.$2")                   // Bad pwc token identifier
+	shouldErr("foo.*", "bar.$1.>")                 // fwcs have to match.
+	shouldErr("foo.>", "bar.baz")                  // fwcs have to match.
+	shouldErr("foo.*.*", "bar.$2")                 // Must place all pwcs.
+	shouldErr("foo.*", "foo.$foo")                 // invalid $ value
+	shouldErr("foo.*", "foo.{{wildcard(2)}}")      // Mapping function being passed an out of range wildcard index
+	shouldErr("foo.*", "foo.{{unimplemented(1)}}") // Mapping trying to use an unknown mapping function
+	shouldErr("foo.*", "foo.{{partition(10)}}")    // Not enough arguments passed to the mapping function
+	shouldErr("foo.*", "foo.{{wildcard(foo)}}")    // Invalid argument passed to the mapping function
+	shouldErr("foo.*", "foo.{{wildcard()}}")       // Not enough arguments passed to the mapping function
+	shouldErr("foo.*", "foo.{{wildcard(1,2)}}")    // Too many arguments passed to the mapping function
+	shouldErr("foo.*", "foo.{{ wildcard5) }}")     // Bad mapping function
+	shouldErr("foo.*", "foo.{{splitLeft(2,2}}")    // arg out of range
+
+	shouldBeOK := func(src, dest string) *subjectTransform {
+		t.Helper()
+		tr, err := newSubjectTransform(src, dest)
+		if err != nil {
+			t.Fatalf("Got an error %v for src=%q and dest=%q", err, src, dest)
+		}
+		return tr
+	}
+
+	shouldBeOK("foo", "bar")
+	shouldBeOK("foo.*.bar.*.baz", "req.$2.$1")
+	shouldBeOK("baz.>", "mybaz.>")
+	shouldBeOK("*", "{{splitfromleft(1,1)}}")
+	shouldBeOK("", "prefix.>")
+
+	shouldMatch := func(src, dest, sample, expected string) {
+		t.Helper()
+		tr := shouldBeOK(src, dest)
+		s, err := tr.Match(sample)
+		if err != nil {
+			t.Fatalf("Got an error %v when expecting a match for %q to %q", err, sample, expected)
+		}
+		if s != expected {
+			t.Fatalf("Dest does not match what was expected. Got %q, expected %q", s, expected)
+		}
+	}
+
+	shouldMatch("", "prefix.>", "foo", "prefix.foo")
+	shouldMatch("foo", "bar", "foo", "bar")
+	shouldMatch("foo.*.bar.*.baz", "req.$2.$1", "foo.A.bar.B.baz", "req.B.A")
+	shouldMatch("baz.>", "my.pre.>", "baz.1.2.3", "my.pre.1.2.3")
+	shouldMatch("baz.>", "foo.bar.>", "baz.1.2.3", "foo.bar.1.2.3")
+	shouldMatch("*", "foo.bar.$1", "foo", "foo.bar.foo")
+	shouldMatch("*", "{{splitfromleft(1,3)}}", "12345", "123.45")
+	shouldMatch("*", "{{SplitFromRight(1,3)}}", "12345", "12.345")
+	shouldMatch("*", "{{SliceFromLeft(1,3)}}", "1234567890", "123.456.789.0")
+	shouldMatch("*", "{{SliceFromRight(1,3)}}", "1234567890", "1.234.567.890")
+	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
+}