Repository: nats-io/nats.go
Pull request #1204 (2023-02-06 13:42:45)

Add option to skip host lookup
Sometimes we want NATS to avoid resolving the IP address of a host. For example, when using a proxy that only lets a specific address connect and blocks everything else.
This is why I added a new option in this PR. It lets you skip the hostname lookup.


Patch:
diff --git a/.travis.yml b/.travis.yml
index 5381fb3..7268558 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,11 +1,11 @@
 language: go
 go:
-- 1.19.x
-- 1.18.x
+- "1.20.x"
+- "1.19.x"
 go_import_path: github.com/nats-io/nats.go
 install:
 - go get -t ./...
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.19 ]]; then
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then
     go install github.com/mattn/goveralls@latest;
     go install github.com/wadey/gocovmerge@latest;
     go install honnef.co/go/tools/cmd/staticcheck@latest;
@@ -14,12 +14,12 @@ install:
 before_script:
 - $(exit $(go fmt ./... | wc -l))
 - go vet -modfile=go_test.mod ./...
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.19 ]]; then
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then
     find . -type f -name "*.go" | xargs misspell -error -locale US;
     GOFLAGS="-mod=mod -modfile=go_test.mod" staticcheck ./...;
   fi
 script:
 - go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.19 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi
 after_success:
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.19 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
diff --git a/go.mod b/go.mod
index a89cd9a..f4bc2c0 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/nats-io/nats.go
 
-go 1.18
+go 1.19
 
 require (
 	github.com/nats-io/nkeys v0.3.0
diff --git a/js.go b/js.go
index a295ff1..c1be762 100644
--- a/js.go
+++ b/js.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2022 The NATS Authors
+// Copyright 2020-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
@@ -1989,6 +1989,11 @@ func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
 		}
 
 		if cinfo.Error != nil {
+			if cinfo.Error.ErrorCode == JSErrCodeInsufficientResourcesErr {
+				// retry for insufficient resources, as it may mean that client is connected to a running
+				// server in cluster while the server hosting R1 JetStream resources is restarting
+				return
+			}
 			pushErr(cinfo.Error)
 			return
 		}
@@ -2673,16 +2678,6 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 		return nil, err
 	}
 
-	// Use the deadline of the context to base the expire times.
-	deadline, _ := ctx.Deadline()
-	ttl = time.Until(deadline)
-	checkCtxErr := func(err error) error {
-		if o.ctx == nil && err == context.DeadlineExceeded {
-			return ErrTimeout
-		}
-		return err
-	}
-
 	var (
 		msgs = make([]*Msg, 0, batch)
 		msg  *Msg
@@ -2716,7 +2711,7 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 		sendReq := func() error {
 			// The current deadline for the context will be used
 			// to set the expires TTL for a fetch request.
-			deadline, _ = ctx.Deadline()
+			deadline, _ := ctx.Deadline()
 			ttl = time.Until(deadline)
 
 			// Check if context has already been canceled or expired.
@@ -2766,11 +2761,235 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 	}
 	// If there is at least a message added to msgs, then need to return OK and no error
 	if err != nil && len(msgs) == 0 {
-		return nil, checkCtxErr(err)
+		return nil, o.checkCtxErr(err)
 	}
 	return msgs, nil
 }
 
+// MessageBatch provides methods to retrieve messages consumed using [Subscribe.FetchBatch].
+type MessageBatch interface {
+	// Messages returns a channel on which messages will be published.
+	Messages() <-chan *Msg
+
+	// Error returns an error encountered when fetching messages.
+	Error() error
+
+	// Done signals end of execution.
+	Done() <-chan struct{}
+}
+
+type messageBatch struct {
+	msgs chan *Msg
+	err  error
+	done chan struct{}
+}
+
+func (mb *messageBatch) Messages() <-chan *Msg {
+	return mb.msgs
+}
+
+func (mb *messageBatch) Error() error {
+	return mb.err
+}
+
+func (mb *messageBatch) Done() <-chan struct{} {
+	return mb.done
+}
+
+// FetchBatch pulls a batch of messages from a stream for a pull consumer.
+// Unlike [Subscription.Fetch], it is non blocking and returns [MessageBatch],
+// allowing to retrieve incoming messages from a channel.
+// The returned channel is always closed after all messages for a batch have been
+// delivered by the server - it is safe to iterate over it using range.
+//
+// To avoid using default JetStream timeout as fetch expiry time, use [nats.MaxWait]
+// or [nats.Context] (with deadline set).
+//
+// This method will not return error in case of pull request expiry (even if there are no messages).
+// Any other error encountered when receiving messages will cause FetchBatch to stop receiving new messages.
+func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, error) {
+	if sub == nil {
+		return nil, ErrBadSubscription
+	}
+	if batch < 1 {
+		return nil, ErrInvalidArg
+	}
+
+	var o pullOpts
+	for _, opt := range opts {
+		if err := opt.configurePull(&o); err != nil {
+			return nil, err
+		}
+	}
+	if o.ctx != nil && o.ttl != 0 {
+		return nil, ErrContextAndTimeout
+	}
+	sub.mu.Lock()
+	jsi := sub.jsi
+	// Reject if this is not a pull subscription. Note that sub.typ is SyncSubscription,
+	// so check for jsi.pull boolean instead.
+	if jsi == nil || !jsi.pull {
+		sub.mu.Unlock()
+		return nil, ErrTypeSubscription
+	}
+
+	nc := sub.conn
+	nms := sub.jsi.nms
+	rply := sub.jsi.deliver
+	js := sub.jsi.js
+	pmc := len(sub.mch) > 0
+
+	// All fetch requests have an expiration, in case of no explicit expiration
+	// then the default timeout of the JetStream context is used.
+	ttl := o.ttl
+	if ttl == 0 {
+		ttl = js.opts.wait
+	}
+	sub.mu.Unlock()
+
+	// Use the given context or setup a default one for the span
+	// of the pull batch request.
+	var (
+		ctx           = o.ctx
+		cancel        context.CancelFunc
+		cancelContext = true
+	)
+	if ctx == nil {
+		ctx, cancel = context.WithTimeout(context.Background(), ttl)
+	} else if _, hasDeadline := ctx.Deadline(); !hasDeadline {
+		// Prevent from passing the background context which will just block
+		// and cannot be canceled either.
+		if octx, ok := ctx.(ContextOpt); ok && octx.Context == context.Background() {
+			return nil, ErrNoDeadlineContext
+		}
+
+		// If the context did not have a deadline, then create a new child context
+		// that will use the default timeout from the JS context.
+		ctx, cancel = context.WithTimeout(ctx, ttl)
+	}
+	defer func() {
+		// only cancel the context here if we are sure the fetching goroutine has not been started yet
+		if cancel != nil && cancelContext {
+			cancel()
+		}
+	}()
+
+	// Check if context not done already before making the request.
+	select {
+	case <-ctx.Done():
+		if o.ctx != nil { // Timeout or Cancel triggered by context object option
+			return nil, ctx.Err()
+		} else { // Timeout triggered by timeout option
+			return nil, ErrTimeout
+		}
+	default:
+	}
+
+	result := &messageBatch{
+		msgs: make(chan *Msg, batch),
+		done: make(chan struct{}, 1),
+	}
+	var msg *Msg
+	for pmc && len(result.msgs) < batch {
+		// Check next msg with booleans that say that this is an internal call
+		// for a pull subscribe (so don't reject it) and don't wait if there
+		// are no messages.
+		msg, err := sub.nextMsgWithContext(ctx, true, false)
+		if err != nil {
+			if err == errNoMessages {
+				err = nil
+			}
+			result.err = err
+			break
+		}
+		// Check msg but just to determine if this is a user message
+		// or status message, however, we don't care about values of status
+		// messages at this point in the Fetch() call, so checkMsg can't
+		// return an error.
+		if usrMsg, _ := checkMsg(msg, false, false); usrMsg {
+			result.msgs <- msg
+		}
+	}
+	if len(result.msgs) == batch || result.err != nil {
+		close(result.msgs)
+		result.done <- struct{}{}
+		return result, nil
+	}
+
+	deadline, _ := ctx.Deadline()
+	ttl = time.Until(deadline)
+
+	// Make our request expiration a bit shorter than the current timeout.
+	expires := ttl
+	if ttl >= 20*time.Millisecond {
+		expires = ttl - 10*time.Millisecond
+	}
+
+	requestBatch := batch - len(result.msgs)
+	req := nextRequest{
+		Expires:  expires,
+		Batch:    requestBatch,
+		MaxBytes: o.maxBytes,
+	}
+	reqJSON, err := json.Marshal(req)
+	if err != nil {
+		close(result.msgs)
+		result.done <- struct{}{}
+		result.err = err
+		return result, nil
+	}
+	if err := nc.PublishRequest(nms, rply, reqJSON); err != nil {
+		if len(result.msgs) == 0 {
+			return nil, err
+		}
+		close(result.msgs)
+		result.done <- struct{}{}
+		result.err = err
+		return result, nil
+	}
+	cancelContext = false
+	go func() {
+		if cancel != nil {
+			defer cancel()
+		}
+		var requestMsgs int
+		for requestMsgs < requestBatch {
+			// Ask for next message and wait if there are no messages
+			msg, err = sub.nextMsgWithContext(ctx, true, true)
+			if err != nil {
+				break
+			}
+			var usrMsg bool
+
+			usrMsg, err = checkMsg(msg, true, false)
+			if err != nil {
+				if err == ErrTimeout {
+					err = nil
+				}
+				break
+			}
+			if usrMsg {
+				result.msgs <- msg
+				requestMsgs++
+			}
+		}
+		if err != nil {
+			result.err = o.checkCtxErr(err)
+		}
+		close(result.msgs)
+		result.done <- struct{}{}
+	}()
+	return result, nil
+}
+
+// checkCtxErr is used to determine whether ErrTimeout should be returned in case of context timeout
+func (o *pullOpts) checkCtxErr(err error) error {
+	if o.ctx == nil && err == context.DeadlineExceeded {
+		return ErrTimeout
+	}
+	return err
+}
+
 func (js *js) getConsumerInfo(stream, consumer string) (*ConsumerInfo, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), js.opts.wait)
 	defer cancel()
diff --git a/jserrors.go b/jserrors.go
index dd81cae..d7959ca 100644
--- a/jserrors.go
+++ b/jserrors.go
@@ -80,10 +80,10 @@ var (
 	// ErrNotJSMessage is returned when attempting to get metadata from non JetStream message .
 	ErrNotJSMessage JetStreamError = &jsError{message: "not a jetstream message"}
 
-	// ErrInvalidStreamName is returned when the provided stream name is invalid (contains '.').
+	// ErrInvalidStreamName is returned when the provided stream name is invalid (contains '.' or ' ').
 	ErrInvalidStreamName JetStreamError = &jsError{message: "invalid stream name"}
 
-	// ErrInvalidConsumerName is returned when the provided consumer name is invalid (contains '.').
+	// ErrInvalidConsumerName is returned when the provided consumer name is invalid (contains '.' or ' ').
 	ErrInvalidConsumerName JetStreamError = &jsError{message: "invalid consumer name"}
 
 	// ErrNoMatchingStream is returned when stream lookup by subject is unsuccessful.
@@ -115,6 +115,7 @@ type ErrorCode uint16
 const (
 	JSErrCodeJetStreamNotEnabledForAccount ErrorCode = 10039
 	JSErrCodeJetStreamNotEnabled           ErrorCode = 10076
+	JSErrCodeInsufficientResourcesErr      ErrorCode = 10023
 
 	JSErrCodeStreamNotFound  ErrorCode = 10059
 	JSErrCodeStreamNameInUse ErrorCode = 10058
diff --git a/jsm.go b/jsm.go
index c02e757..fbf85f5 100644
--- a/jsm.go
+++ b/jsm.go
@@ -127,9 +127,6 @@ type StreamConfig struct {
 	DenyPurge            bool            `json:"deny_purge,omitempty"`
 	AllowRollup          bool            `json:"allow_rollup_hdrs,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"`
 
@@ -139,12 +136,6 @@ type StreamConfig struct {
 	MirrorDirect bool `json:"mirror_direct"`
 }
 
-// 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. The original
 // subject cis remapped from the subject pattern to the destination pattern.
 type RePublish struct {
@@ -161,13 +152,12 @@ type Placement struct {
 
 // 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"`
-	SubjectTransformDest string          `json:"subject_transform_dest,omitempty"`
-	External             *ExternalStream `json:"external,omitempty"`
-	Domain               string          `json:"-"`
+	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"`
+	Domain        string          `json:"-"`
 }
 
 // ExternalStream allows you to qualify access to a stream source in another
@@ -421,19 +411,20 @@ func checkStreamName(stream string) error {
 	if stream == _EMPTY_ {
 		return ErrStreamNameRequired
 	}
-	if strings.Contains(stream, ".") {
+	if strings.ContainsAny(stream, ". ") {
 		return ErrInvalidStreamName
 	}
 	return nil
 }
 
-// Check that the durable name exists and is valid, that is, that it does not contain any "."
+// Check that the consumer name is not empty and is valid (does not contain "." and " ").
+// Additional consumer name validation is done in nats-server.
 // Returns ErrConsumerNameRequired if consumer name is empty, ErrInvalidConsumerName is invalid, otherwise nil
 func checkConsumerName(consumer string) error {
 	if consumer == _EMPTY_ {
 		return ErrConsumerNameRequired
 	}
-	if strings.Contains(consumer, ".") {
+	if strings.ContainsAny(consumer, ". ") {
 		return ErrInvalidConsumerName
 	}
 	return nil
@@ -897,13 +888,11 @@ type StreamAlternate struct {
 
 // StreamSourceInfo shows information about an upstream stream source.
 type StreamSourceInfo struct {
-	Name                 string          `json:"name"`
-	Lag                  uint64          `json:"lag"`
-	Active               time.Duration   `json:"active"`
-	External             *ExternalStream `json:"external"`
-	Error                *APIError       `json:"error"`
-	FilterSubject        string          `json:"filter_subject,omitempty"`
-	SubjectTransformDest string          `json:"subject_transform_dest,omitempty"`
+	Name     string          `json:"name"`
+	Lag      uint64          `json:"lag"`
+	Active   time.Duration   `json:"active"`
+	External *ExternalStream `json:"external"`
+	Error    *APIError       `json:"error"`
 }
 
 // StreamState is information about the given stream.
diff --git a/kv.go b/kv.go
index 1e61beb..5f416bc 100644
--- a/kv.go
+++ b/kv.go
@@ -414,22 +414,14 @@ func (js *js) CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error) {
 		scfg.Mirror = m
 		scfg.MirrorDirect = true
 	} else if len(cfg.Sources) > 0 {
+		// For now we do not allow direct subjects for sources. If that is desired a user could use stream API directly.
 		for _, ss := range cfg.Sources {
-			var sourceBucketName string
-			if strings.HasPrefix(ss.Name, kvBucketNamePre) {
-				sourceBucketName = ss.Name[len(kvBucketNamePre):]
-			} else {
-				sourceBucketName = ss.Name
+			if !strings.HasPrefix(ss.Name, kvBucketNamePre) {
+				ss = ss.copy()
 				ss.Name = fmt.Sprintf(kvBucketNameTmpl, ss.Name)
 			}
-
-			if ss.External == nil || sourceBucketName != cfg.Bucket {
-				ss.FilterSubject = fmt.Sprintf(kvSubjectsTmpl, sourceBucketName)
-				ss.SubjectTransformDest = fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)
-			}
 			scfg.Sources = append(scfg.Sources, ss)
 		}
-		scfg.Subjects = []string{fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)}
 	} else {
 		scfg.Subjects = []string{fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)}
 	}
@@ -446,11 +438,15 @@ func (js *js) CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error) {
 		// and we are now moving to a v2.7.2+. If that is the case
 		// and the only difference is the discard policy, then update
 		// the stream.
+		// The same logic applies for KVs created pre 2.9.x and
+		// the AllowDirect setting.
 		if err == ErrStreamNameAlreadyInUse {
 			if si, _ = js.StreamInfo(scfg.Name); si != nil {
 				// To compare, make the server's stream info discard
 				// policy same than ours.
 				si.Config.Discard = scfg.Discard
+				// Also need to set allow direct for v2.9.x+
+				si.Config.AllowDirect = scfg.AllowDirect
 				if reflect.DeepEqual(&si.Config, scfg) {
 					si, err = js.UpdateStream(scfg)
 				}
diff --git a/micro/request.go b/micro/request.go
index 2d1b056..ce24985 100644
--- a/micro/request.go
+++ b/micro/request.go
@@ -14,6 +14,7 @@
 package micro
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -82,6 +83,14 @@ func (fn HandlerFunc) Handle(req Request) {
 	fn(req)
 }
 
+// ContextHandler is a helper function used to utilize [context.Context]
+// in request handlers.
+func ContextHandler(ctx context.Context, handler func(context.Context, Request)) Handler {
+	return HandlerFunc(func(req Request) {
+		handler(ctx, req)
+	})
+}
+
 // Respond sends the response for the request.
 // Additional headers can be passed using [WithHeaders] option.
 func (r *request) Respond(response []byte, opts ...RespondOpt) error {
diff --git a/nats.go b/nats.go
index 5418f31..84d5299 100644
--- a/nats.go
+++ b/nats.go
@@ -1,4 +1,4 @@
-// Copyright 2012-2022 The NATS Authors
+// Copyright 2012-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
@@ -140,10 +140,6 @@ var (
 	ErrConnectionNotTLS       = errors.New("nats: connection is not tls")
 )
 
-func init() {
-	rand.Seed(time.Now().UnixNano())
-}
-
 // GetDefaultOptions returns default configuration options for the client.
 func GetDefaultOptions() Options {
 	return Options{
@@ -475,6 +471,9 @@ type Options struct {
 	// IgnoreAuthErrorAbort - if set to true, client opts out of the default connect behavior of aborting
 	// subsequent reconnect attempts if server returns the same auth error twice (regardless of reconnect policy).
 	IgnoreAuthErrorAbort bool
+
+	// SkipHostLookup skips the DNS lookup for the server hostname.
+	SkipHostLookup bool
 }
 
 const (
@@ -1268,6 +1267,14 @@ func IgnoreAuthErrorAbort() Option {
 	}
 }
 
+// SkipHostLookup is an Option to skip the host lookup when connecting to a server.
+func SkipHostLookup() Option {
+	return func(o *Options) error {
+		o.SkipHostLookup = true
+		return nil
+	}
+}
+
 // Handler processing
 
 // SetDisconnectHandler will set the disconnect event handler.
@@ -1901,7 +1908,7 @@ func (nc *Conn) createConn() (err error) {
 	hosts := []string{}
 	u := nc.current.url
 
-	if net.ParseIP(u.Hostname()) == nil {
+	if !nc.Opts.SkipHostLookup && net.ParseIP(u.Hostname()) == nil {
 		addrs, _ := net.LookupHost(u.Hostname())
 		for _, addr := range addrs {
 			hosts = append(hosts, net.JoinHostPort(addr, u.Port()))
diff --git a/rand.go b/rand.go
new file mode 100644
index 0000000..0cdee0a
--- /dev/null
+++ b/rand.go
@@ -0,0 +1,29 @@
+// 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.
+
+//go:build !go1.20
+// +build !go1.20
+
+// A Go client for the NATS messaging system (https://nats.io).
+package nats
+
+import (
+	"math/rand"
+	"time"
+)
+
+func init() {
+	// This is not needed since Go 1.20 because now rand.Seed always happens
+	// by default (uses runtime.fastrand64 instead as source).
+	rand.Seed(time.Now().UnixNano())
+}


Test patch:
diff --git a/go_test.mod b/go_test.mod
index 0b544ba..b847036 100644
--- a/go_test.mod
+++ b/go_test.mod
@@ -1,11 +1,11 @@
 module github.com/nats-io/nats.go
 
-go 1.18
+go 1.19
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/nats-io/nats-server/v2 v2.9.12-0.20230129183410-a505c88d65e8
-	github.com/nats-io/nkeys v0.3.1-0.20221215194120-47c7408e7546
+	github.com/nats-io/nats-server/v2 v2.9.6
+	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 	google.golang.org/protobuf v1.23.0
 )
@@ -13,8 +13,7 @@ require (
 require (
 	github.com/klauspost/compress v1.15.11 // indirect
 	github.com/minio/highwayhash v1.0.2 // indirect
-	github.com/nats-io/jwt/v2 v2.3.1-0.20221227170542-bdf40fa3627b // indirect
-	go.uber.org/automaxprocs v1.5.1 // indirect
+	github.com/nats-io/jwt/v2 v2.3.0 // indirect
 	golang.org/x/crypto v0.5.0 // indirect
 	golang.org/x/sys v0.4.0 // indirect
 	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af // indirect
diff --git a/go_test.sum b/go_test.sum
index 161953a..1662e18 100644
--- a/go_test.sum
+++ b/go_test.sum
@@ -13,25 +13,27 @@ github.com/klauspost/compress v1.15.11 h1:Lcadnb3RKGin4FYM/orgq0qde+nc15E5Cbqg4B
 github.com/klauspost/compress v1.15.11/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
 github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
-github.com/nats-io/jwt/v2 v2.3.1-0.20221227170542-bdf40fa3627b h1:exHeHbghpBp1JvdYq7muaKFvJgLD93UDcmoIbFu/9PA=
-github.com/nats-io/jwt/v2 v2.3.1-0.20221227170542-bdf40fa3627b/go.mod h1:DYujvzCMZzUuqB3i1Pnpf1YtkuTwhdI84Aah9wRXkK0=
-github.com/nats-io/nats-server/v2 v2.9.12-0.20230127214148-11ffd695750a h1:iZE9RvG9JCC0TployL6UsxJ3dbwMKBrJakYfvPMYDRc=
-github.com/nats-io/nats-server/v2 v2.9.12-0.20230127214148-11ffd695750a/go.mod h1:ibVHvIWZwqnarh51bnfR3zZWtlL3SjG9X49ocsfFUm4=
-github.com/nats-io/nats-server/v2 v2.9.12-0.20230129183410-a505c88d65e8 h1:MnuDEDsBCO0yU2MVA7BMbiob17nDN1TklcqlxJi6py8=
-github.com/nats-io/nats-server/v2 v2.9.12-0.20230129183410-a505c88d65e8/go.mod h1:ibVHvIWZwqnarh51bnfR3zZWtlL3SjG9X49ocsfFUm4=
-github.com/nats-io/nkeys v0.3.1-0.20221215194120-47c7408e7546 h1:7ZylVLLiDSFqxJTSibNsO2RjVSXj3QWnDc+zKara2HE=
-github.com/nats-io/nkeys v0.3.1-0.20221215194120-47c7408e7546/go.mod h1:JOEZlxMfMnmaLwr+mpmP+RGIYSxLNBFsZykCGaI2PvA=
+github.com/nats-io/jwt/v2 v2.3.0 h1:z2mA1a7tIf5ShggOFlR1oBPgd6hGqcDYsISxZByUzdI=
+github.com/nats-io/jwt/v2 v2.3.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
+github.com/nats-io/nats-server/v2 v2.9.6 h1:RTtK+rv/4CcliOuqGsy58g7MuWkBaWmF5TUNwuUo9Uw=
+github.com/nats-io/nats-server/v2 v2.9.6/go.mod h1:AB6hAnGZDlYfqb7CTAm66ZKMZy9DpfierY1/PbpvI2g=
+github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
+github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-go.uber.org/automaxprocs v1.5.1 h1:e1YG66Lrk73dn4qhg8WFSvhF0JuFQF0ERIp4rpuV8Qk=
-go.uber.org/automaxprocs v1.5.1/go.mod h1:BF4eumQw0P9GtnuxxovUd06vwm1o18oMzFtK66vU6XU=
+golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.5.0 h1:U/0M97KRkSFvyD/3FSmdP5W5swImpNgle/EHFhOsQPE=
 golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.4.0 h1:Zr2JFtRQNX3BCZ8YtxRE9hNJYC8J6I1MVbMg6owUp18=
 golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20220922220347-f3bd1da661af h1:Yx9k8YCG3dvF87UAn2tu2HQLf2dt/eR1bXxpLMWeH+Y=
 golang.org/x/time v0.0.0-20220922220347-f3bd1da661af/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
diff --git a/js_test.go b/js_test.go
index 82d6a7e..3915e91 100644
--- a/js_test.go
+++ b/js_test.go
@@ -1,4 +1,4 @@
-// Copyright 2012-2022 The NATS Authors
+// Copyright 2012-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
@@ -1268,51 +1268,3 @@ func TestStreamNameBySubject(t *testing.T) {
 		}
 	}
 }
-
-func TestJetStreamTransform(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer shutdownJSServerAndRemoveStorage(t, s)
-
-	nc, js := jsClient(t, s)
-	defer nc.Close()
-
-	_, err := js.AddStream(&StreamConfig{
-		Name:             "ORIGIN",
-		Subjects:         []string{"test"},
-		SubjectTransform: &SubjectTransformConfig{Source: ">", Destination: "transformed.>"},
-		Storage:          MemoryStorage,
-	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-
-	err = nc.Publish("test", []byte("1"))
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-
-	_, err = js.AddStream(&StreamConfig{
-		Subjects: []string{},
-		Name:     "SOURCING",
-		Sources:  []*StreamSource{{Name: "ORIGIN", SubjectTransformDest: "fromtest.>"}},
-		Storage:  MemoryStorage,
-	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-
-	// Create a sync subscription with an in-memory ephemeral consumer.
-	sub, err := js.SubscribeSync("fromtest.>", ConsumerMemoryStorage(), BindStream("SOURCING"))
-	if err != nil {
-		t.Fatalf("Error on subscribe: %v", err)
-	}
-
-	m, err := sub.NextMsg(time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	if m.Subject != "fromtest.transformed.test" {
-		t.Fatalf("the subject of the message doesn't match the expected fromtest.transformed.test: %s", m.Subject)
-	}
-
-}
diff --git a/kv_test.go b/kv_test.go
index 3e0d484..0eb96cb 100644
--- a/kv_test.go
+++ b/kv_test.go
@@ -251,66 +251,3 @@ func TestKeyValueCreate(t *testing.T) {
 		t.Fatalf("Unexpected error code, got: %v", kerr.APIError().ErrorCode)
 	}
 }
-
-func TestKeyValueSourcing(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer shutdownJSServerAndRemoveStorage(t, s)
-
-	nc, js := jsClient(t, s)
-	defer nc.Close()
-
-	kvA, err := js.CreateKeyValue(&KeyValueConfig{Bucket: "A"})
-	if err != nil {
-		t.Fatalf("Error creating kv: %v", err)
-	}
-
-	_, err = kvA.Create("keyA", []byte("1"))
-	if err != nil {
-		t.Fatalf("Error creating key: %v", err)
-	}
-
-	if _, err := kvA.Get("keyA"); err != nil {
-		t.Fatalf("Got error getting keyA from A: %v", err)
-	}
-
-	kvB, err := js.CreateKeyValue(&KeyValueConfig{Bucket: "B"})
-	if err != nil {
-		t.Fatalf("Error creating kv: %v", err)
-	}
-
-	_, err = kvB.Create("keyB", []byte("1"))
-	if err != nil {
-		t.Fatalf("Error creating key: %v", err)
-	}
-
-	kvC, err := js.CreateKeyValue(&KeyValueConfig{Bucket: "C", Sources: []*StreamSource{{Name: "A"}, {Name: "B"}}})
-	if err != nil {
-		t.Fatalf("Error creating kv: %v", err)
-	}
-
-	// Wait half a second to make sure it has time to populate the stream from it's sources
-	i := 0
-	for {
-		status, err := kvC.Status()
-		if err != nil {
-			t.Fatalf("Error getting bucket status: %v", err)
-		}
-		if status.Values() == 2 {
-			break
-		} else {
-			i++
-			if i > 3 {
-				t.Fatalf("Error sourcing bucket does not contain the expected number of values")
-			}
-		}
-		time.Sleep(20 * time.Millisecond)
-	}
-
-	if _, err := kvC.Get("keyA"); err != nil {
-		t.Fatalf("Got error getting keyA from C: %v", err)
-	}
-
-	if _, err := kvC.Get("keyB"); err != nil {
-		t.Fatalf("Got error getting keyB from C: %v", err)
-	}
-}
diff --git a/micro/example_test.go b/micro/example_test.go
index 9ead68f..a8e48fd 100644
--- a/micro/example_test.go
+++ b/micro/example_test.go
@@ -14,6 +14,7 @@
 package micro_test
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"reflect"
@@ -313,6 +314,37 @@ func ExampleService_Reset() {
 	}
 }
 
+func ExampleContextHandler() {
+	nc, err := nats.Connect("127.0.0.1:4222")
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer nc.Close()
+
+	handler := func(ctx context.Context, req micro.Request) {
+		select {
+		case <-ctx.Done():
+			req.Error("400", "context canceled", nil)
+		default:
+			req.Respond([]byte("ok"))
+		}
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	config := micro.Config{
+		Name:    "EchoService",
+		Version: "0.1.0",
+		Endpoint: &micro.EndpointConfig{
+			Subject: "echo",
+			Handler: micro.ContextHandler(ctx, handler),
+		},
+	}
+
+	srv, _ := micro.AddService(nc, config)
+	defer srv.Stop()
+}
+
 func ExampleControlSubject() {
 
 	// subject used to get PING from all services
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index 2eb64f9..ac5af67 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -15,6 +15,7 @@ package micro_test
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -984,6 +985,69 @@ func TestMonitoringHandlers(t *testing.T) {
 	}
 }
 
+func TestContextHandler(t *testing.T) {
+	s := RunServerOnPort(-1)
+	defer s.Shutdown()
+
+	nc, err := nats.Connect(s.ClientURL())
+	if err != nil {
+		t.Fatalf("Expected to connect to server, got %v", err)
+	}
+	defer nc.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	type key string
+	ctx = context.WithValue(ctx, key("key"), []byte("val"))
+
+	handler := func(ctx context.Context, req micro.Request) {
+		select {
+		case <-ctx.Done():
+			req.Error("400", "context canceled", nil)
+		default:
+			v := ctx.Value(key("key"))
+			req.Respond(v.([]byte))
+		}
+	}
+	config := micro.Config{
+		Name:    "test_service",
+		Version: "0.1.0",
+		APIURL:  "http://someapi.com/v1",
+		Endpoint: &micro.EndpointConfig{
+			Subject: "test.func",
+			Handler: micro.ContextHandler(ctx, handler),
+			Schema: &micro.Schema{
+				Request:  "request_schema",
+				Response: "response_schema",
+			},
+		},
+	}
+
+	srv, err := micro.AddService(nc, config)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer srv.Stop()
+
+	resp, err := nc.Request("test.func", nil, 1*time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	if string(resp.Data) != "val" {
+		t.Fatalf("Invalid response; want: %q; got: %q", "val", string(resp.Data))
+	}
+	cancel()
+	resp, err = nc.Request("test.func", nil, 1*time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	if resp.Header.Get(micro.ErrorCodeHeader) != "400" {
+		t.Fatalf("Expected error response after canceling context; got: %q", string(resp.Data))
+	}
+
+}
+
 func TestServiceNilSchema(t *testing.T) {
 	s := RunServerOnPort(-1)
 	defer s.Shutdown()
@@ -1316,6 +1380,7 @@ func TestRequestRespond(t *testing.T) {
 			handler := func(req micro.Request) {
 				if errors.Is(test.withRespondError, micro.ErrRespond) {
 					nc.Close()
+					return
 				}
 				if val := req.Headers().Get("key"); val != "value" {
 					t.Fatalf("Expected headers in the request")
diff --git a/test/js_test.go b/test/js_test.go
index 4c0f8ed..8e4a5f5 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2022 The NATS Authors
+// Copyright 2020-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
@@ -955,6 +955,390 @@ func TestJetStreamSubscribe(t *testing.T) {
 	}
 }
 
+func TestPullSubscribeFetchBatch(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	t.Run("basic fetch", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		for i := 0; i < 5; i++ {
+			if _, err := js.Publish("foo", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+		res, err := sub.FetchBatch(10)
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		go func() {
+			time.Sleep(10 * time.Millisecond)
+			for i := 0; i < 5; i++ {
+				js.Publish("foo", []byte("msg"))
+			}
+		}()
+		msgs := make([]*nats.Msg, 0)
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() != nil {
+			t.Fatalf("Unexpected error: %s", res.Error())
+		}
+		if len(msgs) != 10 {
+			t.Fatalf("Expected %d messages; got: %d", 10, len(msgs))
+		}
+	})
+
+	t.Run("multiple concurrent fetches", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		for i := 0; i < 50; i++ {
+			if _, err := js.Publish("foo", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+		var r1, r2, r3 nats.MessageBatch
+		started := &sync.WaitGroup{}
+		started.Add(3)
+		errs := make(chan error, 3)
+		go func() {
+			var err error
+			r1, err = sub.FetchBatch(10)
+			if err != nil {
+				errs <- err
+			}
+			started.Done()
+		}()
+		go func() {
+			var err error
+			r2, err = sub.FetchBatch(10)
+			if err != nil {
+				errs <- err
+			}
+			started.Done()
+		}()
+		go func() {
+			var err error
+			r3, err = sub.FetchBatch(10)
+			if err != nil {
+				errs <- err
+			}
+			started.Done()
+		}()
+		// wait until batch results are available
+		started.Wait()
+
+		// check if any FetchBatch call returned an error
+		select {
+		case err := <-errs:
+			t.Fatalf("Error initializing fetch: %s", err)
+		default:
+		}
+
+		var msgsReceived int
+		for msgsReceived < 30 {
+			select {
+			case <-r1.Messages():
+				msgsReceived++
+			case <-r2.Messages():
+				msgsReceived++
+			case <-r3.Messages():
+				msgsReceived++
+			case <-time.After(1 * time.Second):
+				t.Fatalf("Timeout waiting for incoming messages")
+			}
+		}
+		select {
+		case <-r1.Done():
+		case <-time.After(1 * time.Second):
+			t.Fatalf("FetchBatch result channel should be closed after receiving all messages on r1")
+		}
+		select {
+		case <-r2.Done():
+		case <-time.After(1 * time.Second):
+			t.Fatalf("FetchBatch result channel should be closed after receiving all messages on r2")
+		}
+		select {
+		case <-r3.Done():
+		case <-time.After(1 * time.Second):
+			t.Fatalf("FetchBatch result channel should be closed after receiving all messages on r3")
+		}
+		if r1.Error() != nil {
+			t.Fatalf("Unexpected error: %s", r1.Error())
+		}
+		if r2.Error() != nil {
+			t.Fatalf("Unexpected error: %s", r2.Error())
+		}
+		if r3.Error() != nil {
+			t.Fatalf("Unexpected error: %s", r3.Error())
+		}
+		if msgsReceived != 30 {
+			t.Fatalf("Expected %d messages; got: %d", 30, msgsReceived)
+		}
+	})
+
+	t.Run("deliver all, then consume", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		for i := 0; i < 5; i++ {
+			if _, err := js.Publish("foo", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+		res, err := sub.FetchBatch(5)
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		time.Sleep(10 * time.Millisecond)
+		msgs := make([]*nats.Msg, 0)
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() != nil {
+			t.Fatalf("Unexpected error: %s", res.Error())
+		}
+		if len(msgs) != 5 {
+			t.Fatalf("Expected %d messages; got: %d", 5, len(msgs))
+		}
+	})
+
+	t.Run("fetch with context", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		for i := 0; i < 5; i++ {
+			if _, err := js.Publish("foo", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+		res, err := sub.FetchBatch(10, nats.Context(ctx))
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		go func() {
+			time.Sleep(10 * time.Millisecond)
+			for i := 0; i < 5; i++ {
+				js.Publish("foo", []byte("msg"))
+			}
+		}()
+		msgs := make([]*nats.Msg, 0)
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() != nil {
+			t.Fatalf("Unexpected error: %s", res.Error())
+		}
+		if len(msgs) != 10 {
+			t.Fatalf("Expected %d messages; got: %d", 10, len(msgs))
+		}
+	})
+
+	t.Run("fetch subset of messages", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		for i := 0; i < 10; i++ {
+			js.Publish("foo", []byte("msg"))
+		}
+		res, err := sub.FetchBatch(5)
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		msgs := make([]*nats.Msg, 0)
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() != nil {
+			t.Fatalf("Unexpected error: %s", res.Error())
+		}
+		if len(msgs) != 5 {
+			t.Fatalf("Expected %d messages; got: %d", 10, len(msgs))
+		}
+	})
+
+	t.Run("context timeout, no error", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		for i := 0; i < 5; i++ {
+			if _, err := js.Publish("foo", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+		res, err := sub.FetchBatch(10, nats.Context(ctx))
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		msgs := make([]*nats.Msg, 0)
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() != nil {
+			t.Fatalf("Unexpected error: %s", res.Error())
+		}
+		if len(msgs) != 5 {
+			t.Fatalf("Expected %d messages; got: %d", 5, len(msgs))
+		}
+	})
+
+	t.Run("request expired", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		for i := 0; i < 5; i++ {
+			if _, err := js.Publish("foo", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+		res, err := sub.FetchBatch(10, nats.MaxWait(50*time.Millisecond))
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		msgs := make([]*nats.Msg, 0)
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() != nil {
+			t.Fatalf("Unexpected error: %s", res.Error())
+		}
+		if len(msgs) != 5 {
+			t.Fatalf("Expected %d messages; got: %d", 5, len(msgs))
+		}
+	})
+
+	t.Run("cancel context during fetch", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		for i := 0; i < 5; i++ {
+			if _, err := js.Publish("foo", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		res, err := sub.FetchBatch(10, nats.Context(ctx))
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		go func() {
+			time.Sleep(10 * time.Millisecond)
+			cancel()
+		}()
+		msgs := make([]*nats.Msg, 0)
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() == nil || !errors.Is(res.Error(), context.Canceled) {
+			t.Fatalf("Expected error: %s; got: %s", nats.ErrConsumerDeleted, res.Error())
+		}
+		if len(msgs) != 5 {
+			t.Fatalf("Expected %d messages; got: %d", 5, len(msgs))
+		}
+	})
+
+	t.Run("remove durable consumer during fetch", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "cons")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		for i := 0; i < 5; i++ {
+			if _, err := js.Publish("foo", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+		}
+		res, err := sub.FetchBatch(10)
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		go func() {
+			time.Sleep(10 * time.Millisecond)
+			js.DeleteConsumer("TEST", "cons")
+		}()
+		msgs := make([]*nats.Msg, 0)
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+		if res.Error() == nil || !errors.Is(res.Error(), nats.ErrConsumerDeleted) {
+			t.Fatalf("Expected error: %s; got: %s", nats.ErrConsumerDeleted, err)
+		}
+		if len(msgs) != 5 {
+			t.Fatalf("Expected %d messages; got: %d", 5, len(msgs))
+		}
+	})
+
+	t.Run("validation errors", func(t *testing.T) {
+		defer js.PurgeStream("TEST")
+		sub, err := js.PullSubscribe("foo", "")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		// negative batch size
+		_, err = sub.FetchBatch(-1)
+		if !errors.Is(err, nats.ErrInvalidArg) {
+			t.Errorf("Expected error: %s; got: %s", nats.ErrInvalidArg, err)
+		}
+
+		syncSub, err := js.SubscribeSync("foo")
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+		// invalid subscription type
+		_, err = syncSub.FetchBatch(10)
+		if !errors.Is(err, nats.ErrTypeSubscription) {
+			t.Errorf("Expected error: %s; got: %s", nats.ErrTypeSubscription, err)
+		}
+
+		// both context and max wait set
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+		_, err = sub.FetchBatch(10, nats.Context(ctx), nats.MaxWait(2*time.Second))
+		if !errors.Is(err, nats.ErrContextAndTimeout) {
+			t.Errorf("Expected error: %s; got: %s", nats.ErrContextAndTimeout, err)
+		}
+
+		// passing context.Background() to fetch
+		_, err = sub.FetchBatch(10, nats.Context(context.Background()))
+		if !errors.Is(err, nats.ErrNoDeadlineContext) {
+			t.Errorf("Expected error: %s; got: %s", nats.ErrNoDeadlineContext, err)
+		}
+	})
+}
+
 func TestPullSubscribeConsumerDeleted(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer shutdownJSServerAndRemoveStorage(t, s)
@@ -1397,6 +1781,9 @@ func TestJetStreamManagement(t *testing.T) {
 		if _, err := js.AddStream(&nats.StreamConfig{Name: "bad.stream.name"}); err != nats.ErrInvalidStreamName {
 			t.Fatalf("Expected %v, got %v", nats.ErrInvalidStreamName, err)
 		}
+		if _, err := js.AddStream(&nats.StreamConfig{Name: "bad stream name"}); err != nats.ErrInvalidStreamName {
+			t.Fatalf("Expected %v, got %v", nats.ErrInvalidStreamName, err)
+		}
 	})
 
 	t.Run("bad stream info", func(t *testing.T) {
@@ -1572,6 +1959,9 @@ func TestJetStreamManagement(t *testing.T) {
 			if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "test.durable"}); err != nats.ErrInvalidConsumerName {
 				t.Fatalf("Expected: %v; got: %v", nats.ErrInvalidConsumerName, err)
 			}
+			if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "test durable"}); err != nats.ErrInvalidConsumerName {
+				t.Fatalf("Expected: %v; got: %v", nats.ErrInvalidConsumerName, err)
+			}
 		})
 
 		t.Run("consumer with given name already exists, configs do not match", func(t *testing.T) {
@@ -5362,6 +5752,17 @@ func testJetStreamMirror_Source(t *testing.T, nodes ...*jsServer) {
 		}
 	})
 
+	t.Run("bind to stream with wrong subject fails", func(t *testing.T) {
+		_, err := js.SubscribeSync("secret", nats.BindStream("origin"))
+		if err == nil {
+			t.Fatal("Unexpected success")
+		}
+		apiErr := &nats.APIError{}
+		if !errors.As(err, &apiErr) || apiErr.ErrorCode != 10093 {
+			t.Fatalf("Expected API error 10093; got: %v", err)
+		}
+	})
+
 	t.Run("bind to origin stream", func(t *testing.T) {
 		// This would only avoid the stream names lookup.
 		sub, err := js.SubscribeSync("origin", nats.BindStream("origin"))