Repository: nats-io/nats.go
Pull request #1557 (2024-02-16 13:17:44)

[IMPROVED] Missing section in jetstream readme ToC
Signed-off-by: Piotr Piotrowski <piotr@synadia.com>


Patch:
diff --git a/.travis.yml b/.travis.yml
index 1505f77..9a6b4a8 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,12 +1,12 @@
 language: go
 go:
+- "1.22.x"
 - "1.21.x"
-- "1.20.x"
 go_import_path: github.com/nats-io/nats.go
 install:
 - go get -t ./...
 - curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.22 ]]; then
     go install github.com/mattn/goveralls@latest;
     go install github.com/wadey/gocovmerge@latest;
     go install honnef.co/go/tools/cmd/staticcheck@latest;
@@ -15,22 +15,22 @@ install:
 before_script:
 - $(exit $(go fmt ./... | wc -l))
 - go vet -modfile=go_test.mod ./...
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.22 ]]; then
     find . -type f -name "*.go" | xargs misspell -error -locale US;
     GOFLAGS="-mod=mod -modfile=go_test.mod" staticcheck ./...;
   fi
 - golangci-lint run ./jetstream/...
 script:
 - go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.22 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing; fi
 after_success:
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.22 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
 
 jobs:
   include:
-  - name: "Go: 1.21.x (nats-server@main)"
-    go: "1.21.x"
+  - name: "Go: 1.22.x (nats-server@main)"
+    go: "1.22.x"
     before_script:
     - go get -modfile go_test.mod github.com/nats-io/nats-server/v2@main
   allow_failures:
-  - name: "Go: 1.21.x (nats-server@main)"
+  - name: "Go: 1.22.x (nats-server@main)"
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 9eea61b..cfb7e4e 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -1,6 +1,8 @@
 # Contributing
 
-Thanks for your interest in contributing! This document contains `nats-io/nats.go` specific contributing details. If you are a first-time contributor, please refer to the general [NATS Contributor Guide](https://nats.io/contributing/) to get a comprehensive overview of contributing to the NATS project.
+Thanks for your interest in contributing! This document contains `nats-io/nats.go` specific contributing details. If you
+are a first-time contributor, please refer to the general [NATS Contributor Guide](https://nats.io/contributing/) to get
+a comprehensive overview of contributing to the NATS project.
 
 ## Getting started
 
@@ -10,36 +12,69 @@ There are three general ways you can contribute to this repo:
 - Reporting a bug or regression
 - Contributing changes to the source code
 
-For the first two, refer to the [GitHub Issues](https://github.com/nats-io/nats.go/issues/new/choose) which guides you through the available options along with the needed information to collect.
+For the first two, refer to the [GitHub Issues](https://github.com/nats-io/nats.go/issues/new/choose) which guides you
+through the available options along with the needed information to collect.
 
 ## Contributing changes
 
-_Prior to opening a pull request, it is recommended to open an issue first to ensure the maintainers can review intended changes. Exceptions to this rule include fixing non-functional source such as code comments, documentation or other supporting files._
+_Prior to opening a pull request, it is recommended to open an issue first to ensure the maintainers can review intended
+changes. Exceptions to this rule include fixing non-functional source such as code comments, documentation or other
+supporting files._
 
 Proposing source code changes is done through GitHub's standard pull request workflow.
 
-If your branch is a work-in-progress then please start by creating your pull requests as draft, by clicking the down-arrow next to the `Create pull request` button and instead selecting `Create draft pull request`.
+If your branch is a work-in-progress then please start by creating your pull requests as draft, by clicking the
+down-arrow next to the `Create pull request` button and instead selecting `Create draft pull request`.
 
-This will defer the automatic process of requesting a review from the NATS team and significantly reduces noise until you are ready. Once you are happy, you can click the `Ready for review` button.
+This will defer the automatic process of requesting a review from the NATS team and significantly reduces noise until
+you are ready. Once you are happy, you can click the `Ready for review` button.
 
 ### Guidelines
 
 A good pull request includes:
 
-- A high-level description of the changes, including links to any issues that are related by adding comments like `Resolves #NNN` to your description. See [Linking a Pull Request to an Issue](https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue) for more information.
-- An up-to-date parent commit. Please make sure you are pulling in the latest `main` branch and rebasing your work on top of it, i.e. `git rebase main`.
-- Unit tests where appropriate. Bug fixes will benefit from the addition of regression tests. New features will not be accepted without suitable test coverage!
-- No more commits than necessary. Sometimes having multiple commits is useful for telling a story or isolating changes from one another, but please squash down any unnecessary commits that may just be for clean-up, comments or small changes.
-- No additional external dependencies that aren't absolutely essential. Please do everything you can to avoid pulling in additional libraries/dependencies into `go.mod` as we will be very critical of these.
+- A high-level description of the changes, including links to any issues that are related by adding comments
+  like `Resolves #NNN` to your description.
+  See [Linking a Pull Request to an Issue](https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue)
+  for more information.
+- An up-to-date parent commit. Please make sure you are pulling in the latest `main` branch and rebasing your work on
+  top of it, i.e. `git rebase main`.
+- Unit tests where appropriate. Bug fixes will benefit from the addition of regression tests. New features will not be
+  accepted without suitable test coverage!
+- No more commits than necessary. Sometimes having multiple commits is useful for telling a story or isolating changes
+  from one another, but please squash down any unnecessary commits that may just be for clean-up, comments or small
+  changes.
+- No additional external dependencies that aren't absolutely essential. Please do everything you can to avoid pulling in
+  additional libraries/dependencies into `go.mod` as we will be very critical of these.
 
 ### Sign-off
 
-In order to accept a contribution, you will first need to certify that the contribution is your original work and that you license the work to the project under the [Apache-2.0 license](https://github.com/nats-io/nats.go/blob/main/LICENSE).
+In order to accept a contribution, you will first need to certify that the contribution is your original work and that
+you license the work to the project under
+the [Apache-2.0 license](https://github.com/nats-io/nats.go/blob/main/LICENSE).
 
-This is done by using `Signed-off-by` statements, which should appear in **both** your commit messages and your PR description. Please note that we can only accept sign-offs under a legal name. Nicknames and aliases are not permitted.
+This is done by using `Signed-off-by` statements, which should appear in **both** your commit messages and your PR
+description. Please note that we can only accept sign-offs under a legal name. Nicknames and aliases are not permitted.
 
 To perform a sign-off with `git`, use `git commit -s` (or `--signoff`).
 
 ## Get help
 
-If you have questions about the contribution process, please start a [GitHub discussion](https://github.com/nats-io/nats.go/discussions), join the [NATS Slack](https://slack.nats.io/), or send your question to the [NATS Google Group](https://groups.google.com/forum/#!forum/natsio).
+If you have questions about the contribution process, please start
+a [GitHub discussion](https://github.com/nats-io/nats.go/discussions), join the [NATS Slack](https://slack.nats.io/), or
+send your question to the [NATS Google Group](https://groups.google.com/forum/#!forum/natsio).
+
+## Testing
+
+You should use `go_test.mod` to manage your testing dependencies. Please use the following command to update your
+dependencies and avoid changing the main `go.mod` in a PR:
+
+```shell
+go mod tidy -modfile=go_test.mod
+```
+
+To the tests you can pass `-modfile=go_test.mod` flag to `go test` or instead you can also set `GOFLAGS="-modfile=go_test.mod"` as an environment variable:
+
+```shell
+go test ./... -modfile=go_test.mod
+```
diff --git a/bench/bench.go b/bench/bench.go
index cb72473..4ea4b30 100644
--- a/bench/bench.go
+++ b/bench/bench.go
@@ -36,7 +36,7 @@ type Sample struct {
 	End       time.Time
 }
 
-// SampleGroup for a number of samples, the group is a Sample itself agregating the values the Samples
+// SampleGroup for a number of samples, the group is a Sample itself aggregating the values the Samples
 type SampleGroup struct {
 	Sample
 	Samples []*Sample
@@ -156,7 +156,7 @@ func (s *Sample) Throughput() float64 {
 	return float64(s.MsgBytes) / s.Duration().Seconds()
 }
 
-// Rate of meessages in the job per second
+// Rate of messages in the job per second
 func (s *Sample) Rate() int64 {
 	return int64(float64(s.JobMsgCnt) / s.Duration().Seconds())
 }
diff --git a/jetstream/README.md b/jetstream/README.md
index e2ee01e..3acec99 100644
--- a/jetstream/README.md
+++ b/jetstream/README.md
@@ -29,6 +29,10 @@ This doc covers the basic usage of the `jetstream` package in `nats.go` client.
     - [Basic usage of KV bucket](#basic-usage-of-kv-bucket)
     - [Watching for changes on a bucket](#watching-for-changes-on-a-bucket)
     - [Additional operations on a bucket](#additional-operations-on-a-bucket)
+  - [Object Store](#object-store)
+    - [Basic usage of Object Store](#basic-usage-of-object-store)
+    - [Watching for changes on a store](#watching-for-changes-on-a-store)
+    - [Additional operations on a store](#additional-operations-on-a-store)
   - [Examples](#examples)
 
 ## Overview
diff --git a/jetstream/object.go b/jetstream/object.go
index 271cc22..a0eecff 100644
--- a/jetstream/object.go
+++ b/jetstream/object.go
@@ -918,7 +918,13 @@ func (obs *obs) Get(ctx context.Context, name string, opts ...GetObjectOpt) (Obj
 	}
 
 	chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, info.NUID)
-	_, err = obs.pushJS.Subscribe(chunkSubj, processChunk, nats.OrderedConsumer(), nats.Context(ctx))
+	streamName := fmt.Sprintf(objNameTmpl, obs.name)
+	subscribeOpts := []nats.SubOpt{
+		nats.OrderedConsumer(),
+		nats.Context(ctx),
+		nats.BindStream(streamName),
+	}
+	_, err = obs.pushJS.Subscribe(chunkSubj, processChunk, subscribeOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -1302,7 +1308,8 @@ func (obs *obs) Watch(ctx context.Context, opts ...WatchOpt) (ObjectWatcher, err
 	}
 
 	// Used ordered consumer to deliver results.
-	subOpts := []nats.SubOpt{nats.OrderedConsumer()}
+	streamName := fmt.Sprintf(objNameTmpl, obs.name)
+	subOpts := []nats.SubOpt{nats.OrderedConsumer(), nats.BindStream(streamName)}
 	if !o.includeHistory {
 		subOpts = append(subOpts, nats.DeliverLastPerSubject())
 	}
diff --git a/jserrors.go b/jserrors.go
index ef5d4af..b5c9684 100644
--- a/jserrors.go
+++ b/jserrors.go
@@ -51,7 +51,7 @@ var (
 	// ErrStreamSourceMultipleSubjectTransformsNotSupported is returned when the connected nats-server version does not support setting
 	// the stream sources. If this error is returned when executing AddStream(), the stream with invalid
 	// configuration was already created in the server.
-	ErrStreamSourceMultipleSubjectTransformsNotSupported JetStreamError = &jsError{message: "stream sourceing with multiple subject transforms not supported by nats-server"}
+	ErrStreamSourceMultipleSubjectTransformsNotSupported JetStreamError = &jsError{message: "stream sourcing with multiple subject transforms not supported by nats-server"}
 
 	// ErrConsumerNotFound is an error returned when consumer with given name does not exist.
 	ErrConsumerNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerNotFound, Description: "consumer not found", Code: 404}}
diff --git a/nats.go b/nats.go
index 721624c..bdec2cf 100644
--- a/nats.go
+++ b/nats.go
@@ -1,4 +1,4 @@
-// Copyright 2012-2023 The NATS Authors
+// Copyright 2012-2024 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
@@ -90,55 +90,56 @@ const (
 
 // Errors
 var (
-	ErrConnectionClosed       = errors.New("nats: connection closed")
-	ErrConnectionDraining     = errors.New("nats: connection draining")
-	ErrDrainTimeout           = errors.New("nats: draining connection timed out")
-	ErrConnectionReconnecting = errors.New("nats: connection reconnecting")
-	ErrSecureConnRequired     = errors.New("nats: secure connection required")
-	ErrSecureConnWanted       = errors.New("nats: secure connection not available")
-	ErrBadSubscription        = errors.New("nats: invalid subscription")
-	ErrTypeSubscription       = errors.New("nats: invalid subscription type")
-	ErrBadSubject             = errors.New("nats: invalid subject")
-	ErrBadQueueName           = errors.New("nats: invalid queue name")
-	ErrSlowConsumer           = errors.New("nats: slow consumer, messages dropped")
-	ErrTimeout                = errors.New("nats: timeout")
-	ErrBadTimeout             = errors.New("nats: timeout invalid")
-	ErrAuthorization          = errors.New("nats: authorization violation")
-	ErrAuthExpired            = errors.New("nats: authentication expired")
-	ErrAuthRevoked            = errors.New("nats: authentication revoked")
-	ErrAccountAuthExpired     = errors.New("nats: account authentication expired")
-	ErrNoServers              = errors.New("nats: no servers available for connection")
-	ErrJsonParse              = errors.New("nats: connect message, json parse error")
-	ErrChanArg                = errors.New("nats: argument needs to be a channel type")
-	ErrMaxPayload             = errors.New("nats: maximum payload exceeded")
-	ErrMaxMessages            = errors.New("nats: maximum messages delivered")
-	ErrSyncSubRequired        = errors.New("nats: illegal call on an async subscription")
-	ErrMultipleTLSConfigs     = errors.New("nats: multiple tls.Configs not allowed")
-	ErrNoInfoReceived         = errors.New("nats: protocol exception, INFO not received")
-	ErrReconnectBufExceeded   = errors.New("nats: outbound buffer limit exceeded")
-	ErrInvalidConnection      = errors.New("nats: invalid connection")
-	ErrInvalidMsg             = errors.New("nats: invalid message or message nil")
-	ErrInvalidArg             = errors.New("nats: invalid argument")
-	ErrInvalidContext         = errors.New("nats: invalid context")
-	ErrNoDeadlineContext      = errors.New("nats: context requires a deadline")
-	ErrNoEchoNotSupported     = errors.New("nats: no echo option not supported by this server")
-	ErrClientIDNotSupported   = errors.New("nats: client ID not supported by this server")
-	ErrUserButNoSigCB         = errors.New("nats: user callback defined without a signature handler")
-	ErrNkeyButNoSigCB         = errors.New("nats: nkey defined without a signature handler")
-	ErrNoUserCB               = errors.New("nats: user callback not defined")
-	ErrNkeyAndUser            = errors.New("nats: user callback and nkey defined")
-	ErrNkeysNotSupported      = errors.New("nats: nkeys not supported by the server")
-	ErrStaleConnection        = errors.New("nats: " + STALE_CONNECTION)
-	ErrTokenAlreadySet        = errors.New("nats: token and token handler both set")
-	ErrMsgNotBound            = errors.New("nats: message is not bound to subscription/connection")
-	ErrMsgNoReply             = errors.New("nats: message does not have a reply")
-	ErrClientIPNotSupported   = errors.New("nats: client IP not supported by this server")
-	ErrDisconnected           = errors.New("nats: server is disconnected")
-	ErrHeadersNotSupported    = errors.New("nats: headers not supported by this server")
-	ErrBadHeaderMsg           = errors.New("nats: message could not decode headers")
-	ErrNoResponders           = errors.New("nats: no responders available for request")
-	ErrMaxConnectionsExceeded = errors.New("nats: server maximum connections exceeded")
-	ErrConnectionNotTLS       = errors.New("nats: connection is not tls")
+	ErrConnectionClosed            = errors.New("nats: connection closed")
+	ErrConnectionDraining          = errors.New("nats: connection draining")
+	ErrDrainTimeout                = errors.New("nats: draining connection timed out")
+	ErrConnectionReconnecting      = errors.New("nats: connection reconnecting")
+	ErrSecureConnRequired          = errors.New("nats: secure connection required")
+	ErrSecureConnWanted            = errors.New("nats: secure connection not available")
+	ErrBadSubscription             = errors.New("nats: invalid subscription")
+	ErrTypeSubscription            = errors.New("nats: invalid subscription type")
+	ErrBadSubject                  = errors.New("nats: invalid subject")
+	ErrBadQueueName                = errors.New("nats: invalid queue name")
+	ErrSlowConsumer                = errors.New("nats: slow consumer, messages dropped")
+	ErrTimeout                     = errors.New("nats: timeout")
+	ErrBadTimeout                  = errors.New("nats: timeout invalid")
+	ErrAuthorization               = errors.New("nats: authorization violation")
+	ErrAuthExpired                 = errors.New("nats: authentication expired")
+	ErrAuthRevoked                 = errors.New("nats: authentication revoked")
+	ErrAccountAuthExpired          = errors.New("nats: account authentication expired")
+	ErrNoServers                   = errors.New("nats: no servers available for connection")
+	ErrJsonParse                   = errors.New("nats: connect message, json parse error")
+	ErrChanArg                     = errors.New("nats: argument needs to be a channel type")
+	ErrMaxPayload                  = errors.New("nats: maximum payload exceeded")
+	ErrMaxMessages                 = errors.New("nats: maximum messages delivered")
+	ErrSyncSubRequired             = errors.New("nats: illegal call on an async subscription")
+	ErrMultipleTLSConfigs          = errors.New("nats: multiple tls.Configs not allowed")
+	ErrClientCertOrRootCAsRequired = errors.New("nats: at least one of certCB or rootCAsCB must be set")
+	ErrNoInfoReceived              = errors.New("nats: protocol exception, INFO not received")
+	ErrReconnectBufExceeded        = errors.New("nats: outbound buffer limit exceeded")
+	ErrInvalidConnection           = errors.New("nats: invalid connection")
+	ErrInvalidMsg                  = errors.New("nats: invalid message or message nil")
+	ErrInvalidArg                  = errors.New("nats: invalid argument")
+	ErrInvalidContext              = errors.New("nats: invalid context")
+	ErrNoDeadlineContext           = errors.New("nats: context requires a deadline")
+	ErrNoEchoNotSupported          = errors.New("nats: no echo option not supported by this server")
+	ErrClientIDNotSupported        = errors.New("nats: client ID not supported by this server")
+	ErrUserButNoSigCB              = errors.New("nats: user callback defined without a signature handler")
+	ErrNkeyButNoSigCB              = errors.New("nats: nkey defined without a signature handler")
+	ErrNoUserCB                    = errors.New("nats: user callback not defined")
+	ErrNkeyAndUser                 = errors.New("nats: user callback and nkey defined")
+	ErrNkeysNotSupported           = errors.New("nats: nkeys not supported by the server")
+	ErrStaleConnection             = errors.New("nats: " + STALE_CONNECTION)
+	ErrTokenAlreadySet             = errors.New("nats: token and token handler both set")
+	ErrMsgNotBound                 = errors.New("nats: message is not bound to subscription/connection")
+	ErrMsgNoReply                  = errors.New("nats: message does not have a reply")
+	ErrClientIPNotSupported        = errors.New("nats: client IP not supported by this server")
+	ErrDisconnected                = errors.New("nats: server is disconnected")
+	ErrHeadersNotSupported         = errors.New("nats: headers not supported by this server")
+	ErrBadHeaderMsg                = errors.New("nats: message could not decode headers")
+	ErrNoResponders                = errors.New("nats: no responders available for request")
+	ErrMaxConnectionsExceeded      = errors.New("nats: server maximum connections exceeded")
+	ErrConnectionNotTLS            = errors.New("nats: connection is not tls")
 )
 
 // GetDefaultOptions returns default configuration options for the client.
@@ -565,7 +566,6 @@ type Conn struct {
 	respSub       string               // The wildcard subject
 	respSubPrefix string               // the wildcard prefix including trailing .
 	respSubLen    int                  // the length of the wildcard prefix excluding trailing .
-	respScanf     string               // The scanf template to extract mux token
 	respMux       *Subscription        // A single response subscription
 	respMap       map[string]chan *Msg // Request map for the response msg channels
 	respRand      *rand.Rand           // Used for generating suffix
@@ -607,14 +607,17 @@ type Subscription struct {
 	// For holding information about a JetStream consumer.
 	jsi *jsSub
 
-	delivered  uint64
-	max        uint64
-	conn       *Conn
-	mcb        MsgHandler
-	mch        chan *Msg
-	closed     bool
-	sc         bool
-	connClosed bool
+	delivered     uint64
+	max           uint64
+	conn          *Conn
+	mcb           MsgHandler
+	mch           chan *Msg
+	closed        bool
+	sc            bool
+	connClosed    bool
+	draining      bool
+	status        SubStatus
+	statListeners map[chan SubStatus][]SubStatus
 
 	// Type of Subscription
 	typ SubscriptionType
@@ -635,6 +638,30 @@ type Subscription struct {
 	dropped     int
 }
 
+// Status represents the state of the connection.
+type SubStatus int
+
+const (
+	SubscriptionActive = SubStatus(iota)
+	SubscriptionDraining
+	SubscriptionClosed
+	SubscriptionSlowConsumer
+)
+
+func (s SubStatus) String() string {
+	switch s {
+	case SubscriptionActive:
+		return "Active"
+	case SubscriptionDraining:
+		return "Draining"
+	case SubscriptionClosed:
+		return "Closed"
+	case SubscriptionSlowConsumer:
+		return "SlowConsumer"
+	}
+	return "unknown status"
+}
+
 // Msg represents a message delivered by NATS. This structure is used
 // by Subscribers and PublishMsg().
 //
@@ -864,6 +891,40 @@ func Secure(tls ...*tls.Config) Option {
 	}
 }
 
+// ClientTLSConfig is an Option to set the TLS configuration for secure
+// connections. It can be used to e.g. set TLS config with cert and root CAs
+// from memory. For simple use case of loading cert and CAs from file,
+// ClientCert and RootCAs options are more convenient.
+// If Secure is not already set this will set it as well.
+func ClientTLSConfig(certCB TLSCertHandler, rootCAsCB RootCAsHandler) Option {
+	return func(o *Options) error {
+		o.Secure = true
+
+		if certCB == nil && rootCAsCB == nil {
+			return ErrClientCertOrRootCAsRequired
+		}
+
+		// Smoke test the callbacks to fail early
+		// if they are not valid.
+		if certCB != nil {
+			if _, err := certCB(); err != nil {
+				return err
+			}
+		}
+		if rootCAsCB != nil {
+			if _, err := rootCAsCB(); err != nil {
+				return err
+			}
+		}
+		if o.TLSConfig == nil {
+			o.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
+		}
+		o.TLSCertCB = certCB
+		o.RootCAsCB = rootCAsCB
+		return nil
+	}
+}
+
 // RootCAs is a helper option to provide the RootCAs pool from a list of filenames.
 // If Secure is not already set this will set it as well.
 func RootCAs(file ...string) Option {
@@ -3257,6 +3318,9 @@ func (nc *Conn) processMsg(data []byte) {
 	}
 
 	// Clear any SlowConsumer status.
+	if sub.sc {
+		sub.changeSubStatus(SubscriptionActive)
+	}
 	sub.sc = false
 	sub.mu.Unlock()
 
@@ -3280,8 +3344,9 @@ slowConsumer:
 		sub.pMsgs--
 		sub.pBytes -= len(m.Data)
 	}
-	sub.mu.Unlock()
 	if sc {
+		sub.changeSubStatus(SubscriptionSlowConsumer)
+		sub.mu.Unlock()
 		// Now we need connection's lock and we may end-up in the situation
 		// that we were trying to avoid, except that in this case, the client
 		// is already experiencing client-side slow consumer situation.
@@ -3291,6 +3356,8 @@ slowConsumer:
 			nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, sub, ErrSlowConsumer) })
 		}
 		nc.mu.Unlock()
+	} else {
+		sub.mu.Unlock()
 	}
 }
 
@@ -3903,7 +3970,6 @@ func (nc *Conn) createNewRequestAndSend(subj string, hdr, data []byte) (chan *Ms
 			nc.mu.Unlock()
 			return nil, token, err
 		}
-		nc.respScanf = strings.Replace(nc.respSub, "*", "%s", -1)
 		nc.respMux = s
 	}
 	nc.mu.Unlock()
@@ -4084,16 +4150,14 @@ func (nc *Conn) NewRespInbox() string {
 }
 
 // respToken will return the last token of a literal response inbox
-// which we use for the message channel lookup. This needs to do a
-// scan to protect itself against the server changing the subject.
+// which we use for the message channel lookup. This needs to verify the subject
+// prefix matches to protect itself against the server changing the subject.
 // Lock should be held.
 func (nc *Conn) respToken(respInbox string) string {
-	var token string
-	n, err := fmt.Sscanf(respInbox, nc.respScanf, &token)
-	if err != nil || n != 1 {
-		return ""
+	if token, found := strings.CutPrefix(respInbox, nc.respSubPrefix); found {
+		return token
 	}
-	return token
+	return ""
 }
 
 // Subscribe will express interest in the given subject. The subject
@@ -4263,6 +4327,7 @@ func (nc *Conn) subscribeLocked(subj, queue string, cb MsgHandler, ch chan *Msg,
 		nc.kickFlusher()
 	}
 
+	sub.changeSubStatus(SubscriptionActive)
 	return sub, nil
 }
 
@@ -4306,6 +4371,7 @@ func (nc *Conn) removeSub(s *Subscription) {
 	}
 	// Mark as invalid
 	s.closed = true
+	s.changeSubStatus(SubscriptionClosed)
 	if s.pCond != nil {
 		s.pCond.Broadcast()
 	}
@@ -4375,6 +4441,91 @@ func (s *Subscription) Drain() error {
 	return conn.unsubscribe(s, 0, true)
 }
 
+// IsDraining returns a boolean indicating whether the subscription
+// is being drained.
+// This will return false if the subscription has already been closed.
+func (s *Subscription) IsDraining() bool {
+	if s == nil {
+		return false
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.draining
+}
+
+// StatusChanged returns a channel on which given list of subscription status
+// changes will be sent. If no status is provided, all status changes will be sent.
+// Available statuses are SubscriptionActive, SubscriptionDraining, SubscriptionClosed,
+// and SubscriptionSlowConsumer.
+// The returned channel will be closed when the subscription is closed.
+func (s *Subscription) StatusChanged(statuses ...SubStatus) <-chan SubStatus {
+	if len(statuses) == 0 {
+		statuses = []SubStatus{SubscriptionActive, SubscriptionDraining, SubscriptionClosed, SubscriptionSlowConsumer}
+	}
+	ch := make(chan SubStatus, 10)
+	for _, status := range statuses {
+		s.registerStatusChangeListener(status, ch)
+		// initial status
+		if status == s.status {
+			ch <- status
+		}
+	}
+	return ch
+}
+
+// registerStatusChangeListener registers a channel waiting for a specific status change event.
+// Status change events are non-blocking - if no receiver is waiting for the status change,
+// it will not be sent on the channel. Closed channels are ignored.
+func (s *Subscription) registerStatusChangeListener(status SubStatus, ch chan SubStatus) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.statListeners == nil {
+		s.statListeners = make(map[chan SubStatus][]SubStatus)
+	}
+	if _, ok := s.statListeners[ch]; !ok {
+		s.statListeners[ch] = make([]SubStatus, 0)
+	}
+	s.statListeners[ch] = append(s.statListeners[ch], status)
+}
+
+// sendStatusEvent sends subscription status event to all channels.
+// If there is no listener, sendStatusEvent
+// will not block. Lock should be held entering.
+func (s *Subscription) sendStatusEvent(status SubStatus) {
+	for ch, statuses := range s.statListeners {
+		if !containsStatus(statuses, status) {
+			continue
+		}
+		// only send event if someone's listening
+		select {
+		case ch <- status:
+		default:
+		}
+		if status == SubscriptionClosed {
+			close(ch)
+		}
+	}
+}
+
+func containsStatus(statuses []SubStatus, status SubStatus) bool {
+	for _, s := range statuses {
+		if s == status {
+			return true
+		}
+	}
+	return false
+}
+
+// changeSubStatus changes subscription status and sends events
+// to all listeners. Lock should be held entering.
+func (s *Subscription) changeSubStatus(status SubStatus) {
+	if s == nil {
+		return
+	}
+	s.sendStatusEvent(status)
+	s.status = status
+}
+
 // Unsubscribe will remove interest in the given subject.
 //
 // For a JetStream subscription, if the library has created the JetStream
@@ -4413,6 +4564,11 @@ func (s *Subscription) Unsubscribe() error {
 // checkDrained will watch for a subscription to be fully drained
 // and then remove it.
 func (nc *Conn) checkDrained(sub *Subscription) {
+	defer func() {
+		sub.mu.Lock()
+		defer sub.mu.Unlock()
+		sub.draining = false
+	}()
 	if nc == nil || sub == nil {
 		return
 	}
@@ -4522,6 +4678,10 @@ func (nc *Conn) unsubscribe(sub *Subscription, max int, drainMode bool) error {
 	}
 
 	if drainMode {
+		s.mu.Lock()
+		s.draining = true
+		sub.changeSubStatus(SubscriptionDraining)
+		s.mu.Unlock()
 		go nc.checkDrained(sub)
 	}
 
@@ -4624,6 +4784,7 @@ func (s *Subscription) validateNextMsgState(pullSubInternal bool) error {
 		return ErrSyncSubRequired
 	}
 	if s.sc {
+		s.changeSubStatus(SubscriptionActive)
 		s.sc = false
 		return ErrSlowConsumer
 	}
diff --git a/object.go b/object.go
index f3f5d97..75ceaa8 100644
--- a/object.go
+++ b/object.go
@@ -155,7 +155,7 @@ type ObjectStoreConfig struct {
 	Metadata map[string]string `json:"metadata,omitempty"`
 	// Enable underlying stream compression.
 	// NOTE: Compression is supported for nats-server 2.10.0+
-	Compression bool
+	Compression bool `json:"compression,omitempty"`
 }
 
 type ObjectStoreStatus interface {
@@ -694,7 +694,12 @@ func (obs *obs) Get(name string, opts ...GetObjectOpt) (ObjectResult, error) {
 	}
 
 	chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, info.NUID)
-	_, err = obs.js.Subscribe(chunkSubj, processChunk, OrderedConsumer())
+	streamName := fmt.Sprintf(objNameTmpl, obs.name)
+	subscribeOpts := []SubOpt{
+		OrderedConsumer(),
+		BindStream(streamName),
+	}
+	_, err = obs.js.Subscribe(chunkSubj, processChunk, subscribeOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -1110,7 +1115,8 @@ func (obs *obs) Watch(opts ...WatchOpt) (ObjectWatcher, error) {
 	}
 
 	// Used ordered consumer to deliver results.
-	subOpts := []SubOpt{OrderedConsumer()}
+	streamName := fmt.Sprintf(objNameTmpl, obs.name)
+	subOpts := []SubOpt{OrderedConsumer(), BindStream(streamName)}
 	if !o.includeHistory {
 		subOpts = append(subOpts, DeliverLastPerSubject())
 	}


Test patch:
diff --git a/bench/benchlib_test.go b/bench/benchlib_test.go
index d3b515c..020bae4 100644
--- a/bench/benchlib_test.go
+++ b/bench/benchlib_test.go
@@ -116,7 +116,7 @@ func TestGroupThoughput(t *testing.T) {
 	sg.AddSample(millionMessagesSecondSample(2))
 	sg.AddSample(millionMessagesSecondSample(3))
 	if sg.Throughput() != 2*Million*MsgSize {
-		t.Fatalf("Expected througput at %d million bytes/sec", 2*MsgSize)
+		t.Fatalf("Expected throughput at %d million bytes/sec", 2*MsgSize)
 	}
 }
 
diff --git a/go_test.mod b/go_test.mod
index 4e747bf..24cbb52 100644
--- a/go_test.mod
+++ b/go_test.mod
@@ -4,11 +4,11 @@ go 1.19
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/klauspost/compress v1.17.4
-	github.com/nats-io/nats-server/v2 v2.10.9
+	github.com/klauspost/compress v1.17.6
+	github.com/nats-io/nats-server/v2 v2.10.11
 	github.com/nats-io/nkeys v0.4.7
 	github.com/nats-io/nuid v1.0.1
-	go.uber.org/goleak v1.2.1
+	go.uber.org/goleak v1.3.0
 	golang.org/x/text v0.14.0
 	google.golang.org/protobuf v1.23.0
 )
@@ -16,7 +16,7 @@ require (
 require (
 	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/nats-io/jwt/v2 v2.5.3 // indirect
-	golang.org/x/crypto v0.18.0 // indirect
-	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/crypto v0.19.0 // indirect
+	golang.org/x/sys v0.17.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 )
diff --git a/go_test.sum b/go_test.sum
index e33016d..e6fe158 100644
--- a/go_test.sum
+++ b/go_test.sum
@@ -10,27 +10,27 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4=
-github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
+github.com/klauspost/compress v1.17.6 h1:60eq2E/jlfwQXtvZEeBUYADs+BwKBWURIY+Gj2eRGjI=
+github.com/klauspost/compress v1.17.6/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
 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.5.3 h1:/9SWvzc6hTfamcgXJ3uYRpgj+QuY2aLNqRiqrKcrpEo=
 github.com/nats-io/jwt/v2 v2.5.3/go.mod h1:iysuPemFcc7p4IoYots3IuELSI4EDe9Y0bQMe+I3Bf4=
-github.com/nats-io/nats-server/v2 v2.10.9 h1:VEW43Zz+p+9lARtiPM9ctd6ckun+92ZT2T17HWtwiFI=
-github.com/nats-io/nats-server/v2 v2.10.9/go.mod h1:oorGiV9j3BOLLO3ejQe+U7pfAGyPo+ppD7rpgNF6KTQ=
+github.com/nats-io/nats-server/v2 v2.10.11 h1:yKUiLVincZISpo3A4YljJQ+HfLltGAgoNNJl99KL8I0=
+github.com/nats-io/nats-server/v2 v2.10.11/go.mod h1:dXtOqVWzbMTEj+tUyC/itXjJhW37xh0tUBrTAlqAfx8=
 github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
 github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
-go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
-go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
-golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
-golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+golang.org/x/crypto v0.19.0 h1:ENy+Az/9Y1vSrlrvBSyna3PITt4tiZLf7sgCjZBX7Wo=
+golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
-golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
+golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go
index f820535..d852316 100644
--- a/jetstream/test/kv_test.go
+++ b/jetstream/test/kv_test.go
@@ -1213,6 +1213,35 @@ func TestListKeyValueStores(t *testing.T) {
 }
 
 func TestKeyValueMirrorCrossDomains(t *testing.T) {
+	keyExists := func(t *testing.T, kv jetstream.KeyValue, key string, expected string) jetstream.KeyValueEntry {
+		var e jetstream.KeyValueEntry
+		var err error
+		checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
+			e, err = kv.Get(context.Background(), key)
+			if err != nil {
+				return err
+			}
+			if string(e.Value()) != expected {
+				return fmt.Errorf("Expected value to be %q, got %q", expected, e.Value())
+			}
+			return nil
+		})
+
+		return e
+	}
+
+	keyDeleted := func(t *testing.T, kv jetstream.KeyValue, key string) {
+		checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
+			_, err := kv.Get(context.Background(), key)
+			if err == nil {
+				return fmt.Errorf("Expected key to be gone")
+			}
+			if !errors.Is(err, jetstream.ErrKeyNotFound) {
+				return err
+			}
+			return nil
+		})
+	}
 	conf := createConfFile(t, []byte(`
 		server_name: HUB
 		listen: 127.0.0.1:-1
@@ -1303,21 +1332,16 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 
 	_, err = mkv.PutString(ctx, "v", "vv")
 	expectOk(t, err)
-	e, err := mkv.Get(ctx, "v")
-	expectOk(t, err)
+
+	e := keyExists(t, kv, "v", "vv")
 	if e.Operation() != jetstream.KeyValuePut {
 		t.Fatalf("Got wrong value: %q vs %q", e.Operation(), nats.KeyValuePut)
 	}
 	err = mkv.Delete(ctx, "v")
 	expectOk(t, err)
-	_, err = mkv.Get(ctx, "v")
-	expectErr(t, err, jetstream.ErrKeyNotFound)
+	keyDeleted(t, kv, "v")
 
-	e, err = mkv.Get(ctx, "name")
-	expectOk(t, err)
-	if string(e.Value()) != "rip" {
-		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "rip")
-	}
+	keyExists(t, kv, "name", "rip")
 
 	// Also make sure we can create a watcher on the mirror KV.
 	watcher, err := mkv.WatchAll(ctx)
@@ -1334,31 +1358,21 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 	_, err = rkv.PutString(ctx, "name", "ivan")
 	expectOk(t, err)
 
-	e, err = rkv.Get(ctx, "name")
-	expectOk(t, err)
-	if string(e.Value()) != "ivan" {
-		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "ivan")
-	}
+	keyExists(t, mkv, "name", "ivan")
 	_, err = rkv.PutString(ctx, "v", "vv")
 	expectOk(t, err)
-	e, err = mkv.Get(ctx, "v")
-	expectOk(t, err)
+	e = keyExists(t, mkv, "v", "vv")
 	if e.Operation() != jetstream.KeyValuePut {
 		t.Fatalf("Got wrong value: %q vs %q", e.Operation(), nats.KeyValuePut)
 	}
 	err = rkv.Delete(ctx, "v")
 	expectOk(t, err)
-	_, err = rkv.Get(ctx, "v")
-	expectErr(t, err, jetstream.ErrKeyNotFound)
+	keyDeleted(t, mkv, "v")
 
 	// Shutdown cluster and test get still work.
 	shutdownJSServerAndRemoveStorage(t, s)
 
-	e, err = mkv.Get(ctx, "name")
-	expectOk(t, err)
-	if string(e.Value()) != "ivan" {
-		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "ivan")
-	}
+	keyExists(t, mkv, "name", "ivan")
 }
 
 func TestKeyValueRePublish(t *testing.T) {
diff --git a/jetstream/test/object_test.go b/jetstream/test/object_test.go
index e315797..8f421c5 100644
--- a/jetstream/test/object_test.go
+++ b/jetstream/test/object_test.go
@@ -974,13 +974,6 @@ func TestObjectList(t *testing.T) {
 			t.Fatalf("Expected %+v but got %+v", expected, omap)
 		}
 	})
-
-	t.Run("context timeout", func(t *testing.T) {
-		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Nanosecond)
-		defer cancel()
-		_, err := root.List(ctx)
-		expectErr(t, err, context.DeadlineExceeded)
-	})
 }
 
 func TestObjectMaxBytes(t *testing.T) {
@@ -1204,3 +1197,72 @@ func TestObjectStoreCompression(t *testing.T) {
 		t.Fatalf("Expected stream to be compressed with S2")
 	}
 }
+
+func TestObjectStoreMirror(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	bucketName := "test-bucket"
+
+	ctx := context.Background()
+	obs, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: bucketName, Description: "testing"})
+	expectOk(t, err)
+
+	mirrorBucketName := "mirror-test-bucket"
+
+	_, err = js.CreateStream(ctx, jetstream.StreamConfig{
+		Name: fmt.Sprintf("OBJ_%s", mirrorBucketName),
+		Mirror: &jetstream.StreamSource{
+			Name: fmt.Sprintf("OBJ_%s", bucketName),
+			SubjectTransforms: []jetstream.SubjectTransformConfig{
+				{
+					Source:      fmt.Sprintf("$O.%s.>", bucketName),
+					Destination: fmt.Sprintf("$O.%s.>", mirrorBucketName),
+				},
+			},
+		},
+		AllowRollup: true, // meta messages are always rollups
+	})
+	if err != nil {
+		t.Fatalf("Error creating object store bucket mirror: %v", err)
+	}
+
+	_, err = obs.PutString(ctx, "A", "abc")
+	expectOk(t, err)
+
+	mirrorObs, err := js.ObjectStore(ctx, mirrorBucketName)
+	expectOk(t, err)
+
+	// Make sure we sync.
+	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
+		mirrorValue, err := mirrorObs.GetString(ctx, "A")
+		if err != nil {
+			return err
+		}
+		if mirrorValue != "abc" {
+			t.Fatalf("Expected mirrored object store value to be the same as original")
+		}
+		return nil
+	})
+
+	watcher, err := mirrorObs.Watch(ctx)
+	if err != nil {
+		t.Fatalf("Error creating watcher: %v", err)
+	}
+	defer watcher.Stop()
+
+	// expect to get one value and nil
+	for {
+		select {
+		case info := <-watcher.Updates():
+			if info == nil {
+				return
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("Expected to receive an update")
+		}
+	}
+}
diff --git a/jetstream/test/publish_test.go b/jetstream/test/publish_test.go
index e49b6a9..a7ba8c3 100644
--- a/jetstream/test/publish_test.go
+++ b/jetstream/test/publish_test.go
@@ -1269,18 +1269,23 @@ func TestPublishMsgAsyncWithPendingMsgs(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
 
-		_, err = js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		_, err = js.CreateStream(ctx, jetstream.StreamConfig{
+			Name:     "foo",
+			Subjects: []string{"FOO.*"},
+			// disable stream acks
+			NoAck: true,
+		})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
 
 		for i := 0; i < 5; i++ {
-			_, err = js.PublishAsync("FOO.1", []byte("msg"), jetstream.WithStallWait(1*time.Nanosecond))
+			_, err = js.PublishAsync("FOO.1", []byte("msg"), jetstream.WithStallWait(10*time.Millisecond))
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
 		}
-		if _, err = js.PublishAsync("FOO.1", []byte("msg"), jetstream.WithStallWait(1*time.Nanosecond)); err == nil || !errors.Is(err, jetstream.ErrTooManyStalledMsgs) {
+		if _, err = js.PublishAsync("FOO.1", []byte("msg"), jetstream.WithStallWait(10*time.Millisecond)); err == nil || !errors.Is(err, jetstream.ErrTooManyStalledMsgs) {
 			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrTooManyStalledMsgs, err)
 		}
 	})
diff --git a/test/conn_test.go b/test/conn_test.go
index e621c90..36d602c 100644
--- a/test/conn_test.go
+++ b/test/conn_test.go
@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"net"
 	"os"
@@ -234,6 +235,103 @@ func TestServerSecureConnections(t *testing.T) {
 	}
 }
 
+func TestClientTLSConfig(t *testing.T) {
+	s, opts := RunServerWithConfig("./configs/tlsverify.conf")
+	defer s.Shutdown()
+
+	endpoint := fmt.Sprintf("%s:%d", opts.Host, opts.Port)
+	secureURL := fmt.Sprintf("nats://%s", endpoint)
+
+	// Make sure this fails
+	nc, err := nats.Connect(secureURL, nats.Secure())
+	if err == nil {
+		nc.Close()
+		t.Fatal("Should have failed (TLS) connection without client certificate")
+	}
+	cert, err := os.ReadFile("./configs/certs/client-cert.pem")
+	if err != nil {
+		t.Fatal("Failed to read client certificate")
+	}
+	key, err := os.ReadFile("./configs/certs/client-key.pem")
+	if err != nil {
+		t.Fatal("Failed to read client key")
+	}
+	rootCAs, err := os.ReadFile("./configs/certs/ca.pem")
+	if err != nil {
+		t.Fatal("Failed to read root CAs")
+	}
+
+	certCB := func() (tls.Certificate, error) {
+		cert, err := tls.X509KeyPair(cert, key)
+		if err != nil {
+			return tls.Certificate{}, fmt.Errorf("nats: error loading client certificate: %w", err)
+		}
+		cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
+		if err != nil {
+			return tls.Certificate{}, fmt.Errorf("nats: error parsing client certificate: %w", err)
+		}
+		return cert, nil
+	}
+
+	caCB := func() (*x509.CertPool, error) {
+		pool := x509.NewCertPool()
+		ok := pool.AppendCertsFromPEM(rootCAs)
+		if !ok {
+			return nil, fmt.Errorf("nats: failed to parse root certificate from")
+		}
+		return pool, nil
+	}
+
+	// Check parameters validity
+	_, err = nats.Connect(secureURL, nats.ClientTLSConfig(nil, nil))
+	if !errors.Is(err, nats.ErrClientCertOrRootCAsRequired) {
+		t.Fatalf("Expected error %q, got %q", nats.ErrClientCertOrRootCAsRequired, err)
+	}
+
+	certErr := &tls.CertificateVerificationError{}
+	// Should fail because of missing CA
+	_, err = nats.Connect(secureURL,
+		nats.ClientCert("./configs/certs/client-cert.pem", "./configs/certs/client-key.pem"))
+	if ok := errors.As(err, &certErr); !ok {
+		t.Fatalf("Expected error %q, got %q", nats.ErrClientCertOrRootCAsRequired, err)
+	}
+
+	// Should fail because of missing certificate
+	_, err = nats.Connect(secureURL,
+		nats.ClientTLSConfig(nil, caCB))
+	if !strings.Contains(err.Error(), "bad certificate") && !strings.Contains(err.Error(), "certificate required") {
+		t.Fatalf("Expected missing certificate error; got: %s", err)
+	}
+
+	nc, err = nats.Connect(secureURL,
+		nats.ClientTLSConfig(certCB, caCB))
+	if err != nil {
+		t.Fatalf("Failed to create (TLS) connection: %v", err)
+	}
+	defer nc.Close()
+
+	omsg := []byte("Hello!")
+	checkRecv := make(chan bool)
+
+	received := 0
+	nc.Subscribe("foo", func(m *nats.Msg) {
+		received++
+		if !bytes.Equal(m.Data, omsg) {
+			t.Fatal("Message received does not match")
+		}
+		checkRecv <- true
+	})
+	err = nc.Publish("foo", omsg)
+	if err != nil {
+		t.Fatalf("Failed to publish on secure (TLS) connection: %v", err)
+	}
+	nc.Flush()
+
+	if err := Wait(checkRecv); err != nil {
+		t.Fatal("Failed to receive message")
+	}
+}
+
 func TestClientCertificate(t *testing.T) {
 	s, opts := RunServerWithConfig("./configs/tlsverify.conf")
 	defer s.Shutdown()
diff --git a/test/context_test.go b/test/context_test.go
index 4557fe3..b9c2f24 100644
--- a/test/context_test.go
+++ b/test/context_test.go
@@ -220,20 +220,28 @@ func testContextRequestWithCancel(t *testing.T, nc *nats.Conn) {
 		cancelCB()
 	})
 
-	nc.Subscribe("slow", func(m *nats.Msg) {
+	sub1, err := nc.Subscribe("slow", func(m *nats.Msg) {
 		// simulates latency into the client so that timeout is hit.
 		time.Sleep(40 * time.Millisecond)
 		nc.Publish(m.Reply, []byte("OK"))
 	})
-	nc.Subscribe("slower", func(m *nats.Msg) {
+	if err != nil {
+		t.Fatalf("Expected to be able to subscribe: %s", err)
+	}
+	defer sub1.Unsubscribe()
+	sub2, err := nc.Subscribe("slower", func(m *nats.Msg) {
 		// we know this request will take longer so extend the timeout
 		expirationTimer.Reset(100 * time.Millisecond)
 
 		// slower reply which would have hit original timeout
-		time.Sleep(90 * time.Millisecond)
+		time.Sleep(70 * time.Millisecond)
 
 		nc.Publish(m.Reply, []byte("Also OK"))
 	})
+	if err != nil {
+		t.Fatalf("Expected to be able to subscribe: %s", err)
+	}
+	defer sub2.Unsubscribe()
 
 	for i := 0; i < 2; i++ {
 		resp, err := nc.RequestWithContext(ctx, "slow", []byte(""))
@@ -263,7 +271,7 @@ func testContextRequestWithCancel(t *testing.T, nc *nats.Conn) {
 	}
 
 	// One more slow request will expire the timer and cause an error...
-	_, err := nc.RequestWithContext(ctx, "slow", []byte(""))
+	_, err = nc.RequestWithContext(ctx, "slow", []byte(""))
 	if err == nil {
 		t.Fatal("Expected request with cancellation context to fail")
 	}
@@ -1089,12 +1097,12 @@ func TestFlushWithContext(t *testing.T) {
 		t.Fatalf("Expected '%v', got '%v'", nats.ErrNoDeadlineContext, err)
 	}
 
-	dctx, cancel := context.WithTimeout(ctx, 0)
-	defer cancel()
+	dctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	cancel()
 
-	// A context with a deadline should return when expired.
-	if err := nc.FlushWithContext(dctx); err != context.DeadlineExceeded {
-		t.Fatalf("Expected '%v', got '%v'", context.DeadlineExceeded, err)
+	// A closed context should error.
+	if err := nc.FlushWithContext(dctx); err != context.Canceled {
+		t.Fatalf("Expected '%v', got '%v'", context.Canceled, err)
 	}
 }
 
diff --git a/test/drain_test.go b/test/drain_test.go
index 1168f61..c53305e 100644
--- a/test/drain_test.go
+++ b/test/drain_test.go
@@ -1,4 +1,4 @@
-// Copyright 2018-2023 The NATS Authors
+// Copyright 2018-2024 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
@@ -55,6 +55,9 @@ func TestDrain(t *testing.T) {
 
 	// Drain it and make sure we receive all messages.
 	sub.Drain()
+	if !sub.IsDraining() {
+		t.Fatalf("Expected to be draining")
+	}
 	select {
 	case <-done:
 		break
@@ -64,6 +67,10 @@ func TestDrain(t *testing.T) {
 			t.Fatalf("Did not receive all messages: %d of %d", r, expected)
 		}
 	}
+	time.Sleep(100 * time.Millisecond)
+	if sub.IsDraining() {
+		t.Fatalf("Expected to be done draining")
+	}
 }
 
 func TestDrainQueueSub(t *testing.T) {
diff --git a/test/js_test.go b/test/js_test.go
index 4ae2915..540ae41 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2023 The NATS Authors
+// Copyright 2020-2024 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
@@ -593,17 +593,22 @@ func TestJetStreamSubscribe(t *testing.T) {
 
 	// If we are here we have received all of the messages.
 	// We hang the ConsumerInfo option off of the subscription, so we use that to check status.
-	info, _ := sub3.ConsumerInfo()
-	if info.Config.AckPolicy != nats.AckExplicitPolicy {
-		t.Fatalf("Expected ack explicit policy, got %q", info.Config.AckPolicy)
-	}
-	if info.Delivered.Consumer != uint64(toSend) {
-		t.Fatalf("Expected to have received all %d messages, got %d", toSend, info.Delivered.Consumer)
-	}
-	// Make sure we auto-ack'd
-	if info.AckFloor.Consumer != uint64(toSend) {
-		t.Fatalf("Expected to have ack'd all %d messages, got ack floor of %d", toSend, info.AckFloor.Consumer)
-	}
+	// We may need to retry this check since the acks sent by the client have to be processed
+	// on the server.
+	checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
+		info, _ := sub3.ConsumerInfo()
+		if info.Config.AckPolicy != nats.AckExplicitPolicy {
+			t.Fatalf("Expected ack explicit policy, got %q", info.Config.AckPolicy)
+		}
+		if info.Delivered.Consumer != uint64(toSend) {
+			return fmt.Errorf("Expected to have received all %d messages, got %d", toSend, info.Delivered.Consumer)
+		}
+		// Make sure we auto-ack'd
+		if info.AckFloor.Consumer != uint64(toSend) {
+			return fmt.Errorf("Expected to have ack'd all %d messages, got ack floor of %d", toSend, info.AckFloor.Consumer)
+		}
+		return nil
+	})
 	sub3.Unsubscribe()
 	sub2.Unsubscribe()
 	sub1.Unsubscribe()
@@ -619,7 +624,7 @@ func TestJetStreamSubscribe(t *testing.T) {
 	expectConsumers(t, 1)
 
 	// Make sure we registered as a durable.
-	info, _ = sub.ConsumerInfo()
+	info, _ := sub.ConsumerInfo()
 	if info.Config.Durable != dname {
 		t.Fatalf("Expected durable name to be set to %q, got %q", dname, info.Config.Durable)
 	}
diff --git a/test/kv_test.go b/test/kv_test.go
index 5759f5d..4f5d81e 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -1071,6 +1071,35 @@ func TestListKeyValueStores(t *testing.T) {
 }
 
 func TestKeyValueMirrorCrossDomains(t *testing.T) {
+	keyExists := func(t *testing.T, kv nats.KeyValue, key string, expected string) nats.KeyValueEntry {
+		var e nats.KeyValueEntry
+		var err error
+		checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
+			e, err = kv.Get(key)
+			if err != nil {
+				return err
+			}
+			if string(e.Value()) != expected {
+				return fmt.Errorf("Expected value to be %q, got %q", expected, e.Value())
+			}
+			return nil
+		})
+		return e
+	}
+
+	keyDeleted := func(t *testing.T, kv nats.KeyValue, key string) {
+		checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
+			_, err := kv.Get(key)
+			if err == nil {
+				return fmt.Errorf("Expected key to be gone")
+			}
+			if !errors.Is(err, nats.ErrKeyNotFound) {
+				return err
+			}
+			return nil
+		})
+	}
+
 	conf := createConfFile(t, []byte(`
 		server_name: HUB
 		listen: 127.0.0.1:-1
@@ -1159,23 +1188,15 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 	_, err = mkv.PutString("v", "vv")
 	expectOk(t, err)
 	// wait for the key to be propagated to the mirror
-	time.Sleep(10 * time.Millisecond)
-	e, err := mkv.Get("v")
-	expectOk(t, err)
+	e := keyExists(t, kv, "v", "vv")
 	if e.Operation() != nats.KeyValuePut {
 		t.Fatalf("Got wrong value: %q vs %q", e.Operation(), nats.KeyValuePut)
 	}
 	err = mkv.Delete("v")
 	expectOk(t, err)
-	time.Sleep(10 * time.Millisecond)
-	_, err = mkv.Get("v")
-	expectErr(t, err, nats.ErrKeyNotFound)
+	keyDeleted(t, kv, "v")
 
-	e, err = mkv.Get("name")
-	expectOk(t, err)
-	if string(e.Value()) != "rip" {
-		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "rip")
-	}
+	keyExists(t, kv, "name", "rip")
 
 	// Also make sure we can create a watcher on the mirror KV.
 	watcher, err := mkv.WatchAll()
@@ -1192,34 +1213,21 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 	_, err = rkv.PutString("name", "ivan")
 	expectOk(t, err)
 
-	time.Sleep(10 * time.Millisecond)
-	e, err = rkv.Get("name")
-	expectOk(t, err)
-	if string(e.Value()) != "ivan" {
-		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "ivan")
-	}
+	e = keyExists(t, mkv, "name", "ivan")
 	_, err = rkv.PutString("v", "vv")
 	expectOk(t, err)
-	time.Sleep(10 * time.Millisecond)
-	e, err = mkv.Get("v")
-	expectOk(t, err)
+	keyExists(t, mkv, "v", "vv")
 	if e.Operation() != nats.KeyValuePut {
 		t.Fatalf("Got wrong value: %q vs %q", e.Operation(), nats.KeyValuePut)
 	}
 	err = rkv.Delete("v")
 	expectOk(t, err)
-	time.Sleep(10 * time.Millisecond)
-	_, err = rkv.Get("v")
-	expectErr(t, err, nats.ErrKeyNotFound)
+	keyDeleted(t, mkv, "v")
 
 	// Shutdown cluster and test get still work.
 	shutdownJSServerAndRemoveStorage(t, s)
 
-	e, err = mkv.Get("name")
-	expectOk(t, err)
-	if string(e.Value()) != "ivan" {
-		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "ivan")
-	}
+	keyExists(t, mkv, "name", "ivan")
 }
 
 func TestKeyValueNonDirectGet(t *testing.T) {
diff --git a/test/object_test.go b/test/object_test.go
index 616cc1d..f6ecb57 100644
--- a/test/object_test.go
+++ b/test/object_test.go
@@ -892,22 +892,15 @@ func TestObjectList(t *testing.T) {
 			t.Fatalf("Expected 4 total objects, got %d", len(omap))
 		}
 		expected := map[string]struct{}{
-			"A": struct{}{},
-			"B": struct{}{},
-			"C": struct{}{},
-			"b": struct{}{},
+			"A": {},
+			"B": {},
+			"C": {},
+			"b": {},
 		}
 		if !reflect.DeepEqual(omap, expected) {
 			t.Fatalf("Expected %+v but got %+v", expected, omap)
 		}
 	})
-
-	t.Run("context timeout", func(t *testing.T) {
-		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Nanosecond)
-		defer cancel()
-		_, err := root.List(nats.Context(ctx))
-		expectErr(t, err, context.DeadlineExceeded)
-	})
 }
 
 func TestObjectMaxBytes(t *testing.T) {
@@ -1121,3 +1114,71 @@ func TestObjectStoreCompression(t *testing.T) {
 		t.Fatalf("Expected stream to be compressed with S2")
 	}
 }
+
+func TestObjectStoreMirror(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	bucketName := "test-bucket"
+
+	obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: bucketName, Description: "testing"})
+	expectOk(t, err)
+
+	mirrorBucketName := "mirror-test-bucket"
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name: fmt.Sprintf("OBJ_%s", mirrorBucketName),
+		Mirror: &nats.StreamSource{
+			Name: fmt.Sprintf("OBJ_%s", bucketName),
+			SubjectTransforms: []nats.SubjectTransformConfig{
+				{
+					Source:      fmt.Sprintf("$O.%s.>", bucketName),
+					Destination: fmt.Sprintf("$O.%s.>", mirrorBucketName),
+				},
+			},
+		},
+		AllowRollup: true, // meta messages are always rollups
+	})
+	if err != nil {
+		t.Fatalf("Error creating object store bucket mirror: %v", err)
+	}
+
+	_, err = obs.PutString("A", "abc")
+	expectOk(t, err)
+
+	mirrorObs, err := js.ObjectStore(mirrorBucketName)
+	expectOk(t, err)
+
+	// Make sure we sync.
+	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
+		mirrorValue, err := mirrorObs.GetString("A")
+		if err != nil {
+			return err
+		}
+		if mirrorValue != "abc" {
+			t.Fatalf("Expected mirrored object store value to be the same as original")
+		}
+		return nil
+	})
+
+	watcher, err := mirrorObs.Watch()
+	if err != nil {
+		t.Fatalf("Error creating watcher: %v", err)
+	}
+	defer watcher.Stop()
+
+	// expect to get one value and nil
+	for {
+		select {
+		case info := <-watcher.Updates():
+			if info == nil {
+				return
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("Expected to receive an update")
+		}
+	}
+}
diff --git a/test/sub_test.go b/test/sub_test.go
index c359639..0bf2880 100644
--- a/test/sub_test.go
+++ b/test/sub_test.go
@@ -1,4 +1,4 @@
-// Copyright 2013-2023 The NATS Authors
+// Copyright 2013-2024 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
@@ -14,6 +14,7 @@
 package test
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -568,7 +569,7 @@ func TestAsyncErrHandler(t *testing.T) {
 		if s != sub {
 			t.Fatal("Did not receive proper subscription")
 		}
-		if e != nats.ErrSlowConsumer {
+		if !errors.Is(e, nats.ErrSlowConsumer) {
 			t.Fatalf("Did not receive proper error: %v vs %v", e, nats.ErrSlowConsumer)
 		}
 		// Suppress additional calls
@@ -636,7 +637,7 @@ func TestAsyncErrHandlerChanSubscription(t *testing.T) {
 
 	nc.SetErrorHandler(func(c *nats.Conn, s *nats.Subscription, e error) {
 		atomic.AddInt64(&aeCalled, 1)
-		if e != nats.ErrSlowConsumer {
+		if !errors.Is(e, nats.ErrSlowConsumer) {
 			t.Fatalf("Did not receive proper error: %v vs %v",
 				e, nats.ErrSlowConsumer)
 		}
@@ -1614,3 +1615,137 @@ func TestSubscribe_ClosedHandler(t *testing.T) {
 		t.Fatal("Did not receive closed callback")
 	}
 }
+
+func TestSubscriptionEvents(t *testing.T) {
+
+	waitForStatus := func(t *testing.T, ch <-chan nats.SubStatus, expected nats.SubStatus) {
+		t.Helper()
+		select {
+		case s := <-ch:
+			if s != expected {
+				t.Fatalf("Expected status: %s; got: %s", expected, s)
+			}
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Timeout waiting for status %q", expected)
+		}
+	}
+	t.Run("default events", func(t *testing.T) {
+		s := RunDefaultServer()
+		defer s.Shutdown()
+
+		nc := NewDefaultConnection(t)
+		// disable slow consumer prints
+		nc.SetErrorHandler(func(c *nats.Conn, s *nats.Subscription, e error) {})
+		defer nc.Close()
+
+		blockChan := make(chan struct{})
+		sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {
+			// block in subscription callback
+			// to force slow consumer
+			<-blockChan
+		})
+		if err != nil {
+			t.Fatalf("Error subscribing: %v", err)
+		}
+		sub.SetPendingLimits(10, 1024)
+		status := sub.StatusChanged()
+
+		// initial status
+		waitForStatus(t, status, nats.SubscriptionActive)
+
+		for i := 0; i < 11; i++ {
+			nc.Publish("foo", []byte("Hello"))
+		}
+		waitForStatus(t, status, nats.SubscriptionSlowConsumer)
+		close(blockChan)
+
+		sub.Drain()
+
+		waitForStatus(t, status, nats.SubscriptionDraining)
+
+		waitForStatus(t, status, nats.SubscriptionClosed)
+	})
+
+	t.Run("slow consumer event only", func(t *testing.T) {
+		s := RunDefaultServer()
+		defer s.Shutdown()
+
+		nc := NewDefaultConnection(t)
+		defer nc.Close()
+
+		blockChan := make(chan struct{})
+		sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {
+			// block in subscription callback
+			// to force slow consumer
+			<-blockChan
+		})
+		// disable slow consumer prints
+		nc.SetErrorHandler(func(c *nats.Conn, s *nats.Subscription, e error) {})
+		defer sub.Unsubscribe()
+		if err != nil {
+			t.Fatalf("Error subscribing: %v", err)
+		}
+		sub.SetPendingLimits(10, 1024)
+		status := sub.StatusChanged(nats.SubscriptionSlowConsumer)
+
+		for i := 0; i < 20; i++ {
+			nc.Publish("foo", []byte("Hello"))
+		}
+		waitForStatus(t, status, nats.SubscriptionSlowConsumer)
+		close(blockChan)
+
+		// now try with sync sub
+		sub, err = nc.SubscribeSync("foo")
+		if err != nil {
+			t.Fatalf("Error subscribing: %v", err)
+		}
+		defer sub.Unsubscribe()
+		sub.SetPendingLimits(10, 1024)
+		status = sub.StatusChanged(nats.SubscriptionSlowConsumer)
+
+		for i := 0; i < 20; i++ {
+			nc.Publish("foo", []byte("Hello"))
+		}
+		waitForStatus(t, status, nats.SubscriptionSlowConsumer)
+	})
+
+	t.Run("do not block channel if it's not read", func(t *testing.T) {
+		s := RunDefaultServer()
+		defer s.Shutdown()
+
+		nc := NewDefaultConnection(t)
+		// disable slow consumer prints
+		nc.SetErrorHandler(func(c *nats.Conn, s *nats.Subscription, e error) {})
+		defer nc.Close()
+
+		blockChan := make(chan struct{})
+		sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {
+			// block in subscription callback
+			// to force slow consumer
+			<-blockChan
+		})
+		defer sub.Unsubscribe()
+		if err != nil {
+			t.Fatalf("Error subscribing: %v", err)
+		}
+		sub.SetPendingLimits(10, 1024)
+		status := sub.StatusChanged()
+		waitForStatus(t, status, nats.SubscriptionActive)
+
+		// chan length is 10, so make sure we switch state more times
+		for i := 0; i < 20; i++ {
+			// subscription will enter slow consumer state
+			for i := 0; i < 11; i++ {
+				nc.Publish("foo", []byte("Hello"))
+			}
+
+			// messages flow normally, status flips to active
+			for i := 0; i < 10; i++ {
+				nc.Publish("foo", []byte("Hello"))
+				blockChan <- struct{}{}
+			}
+		}
+		// do not read from subscription
+		close(blockChan)
+	})
+}