Repository: nats-io/nats.go
Pull request #1660 (2024-06-20 20:29:24)

Automate updating dependencies report
The dependencies.md file is important for understanding what linceses are used by nats.go dependencies, however it needed manual operation.

This automates it by creating a PR if go.mod dependencies changed.

Signed-off-by: Tomasz Pietrek <tomasz@nats.io>


Patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
new file mode 100644
index 0000000..0ea16de
--- /dev/null
+++ b/.github/workflows/ci.yaml
@@ -0,0 +1,74 @@
+name: ci
+on: 
+  push:
+    branches:
+      - main
+  pull_request:
+  release:
+    types: [published]
+
+jobs:
+    lint:
+        runs-on: ubuntu-latest
+        steps:
+            - name: Checkout code
+              uses: actions/checkout@v4
+      
+            - name: Setup Go
+              uses: actions/setup-go@v5
+              with:
+                go-version: 'stable'
+
+            - name: Install deps
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go get -t ./...
+                go install honnef.co/go/tools/cmd/staticcheck@latest
+                go install github.com/client9/misspell/cmd/misspell@latest
+                go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest
+            
+            - name: Run linters
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                $(exit $(go fmt -modfile=go_test.mod  ./... | wc -l))
+                go vet -modfile=go_test.mod ./... 
+                GOFLAGS="-mod=mod -modfile=go_test.mod" staticcheck ./...
+                find . -type f -name "*.go" | xargs misspell -error -locale US
+                golangci-lint run --timeout 5m0s ./jetstream/...
+
+    test:
+        runs-on: ubuntu-latest
+
+        strategy:
+            matrix:
+              go: [ "1.22", "1.23" ]
+        steps:
+            - name: Checkout code
+              uses: actions/checkout@v4
+      
+            - name: Setup Go
+              uses: actions/setup-go@v5
+              with:
+                go-version: ${{ matrix.go }}
+
+            - name: Install deps
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go install github.com/mattn/goveralls@latest
+                go install github.com/wadey/gocovmerge@latest
+
+            - name: Test and coverage
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
+                if [ "${{ matrix.go }}" = "1.23" ]; then
+                  ./scripts/cov.sh CI
+                else
+                  go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing
+                fi
+
+            - name: Coveralls
+              if: matrix.go == '1.23'
+              uses: coverallsapp/github-action@v2
+              with:
+               file: acc.out
\ No newline at end of file
diff --git a/.github/workflows/dependencies.yaml b/.github/workflows/dependencies.yaml
new file mode 100644
index 0000000..5d9a7dc
--- /dev/null
+++ b/.github/workflows/dependencies.yaml
@@ -0,0 +1,61 @@
+name: License Check
+
+on:
+  push:
+    paths:
+      - 'go.mod'
+    branches:
+      - main
+
+jobs:
+  license-check:
+    runs-on: ubuntu-latest
+
+    env:
+      BRANCH_NAME: update-report-branch-${{ github.run_id }}
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0  # Fetch all history for all branches and tags
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.22'
+
+      - name: Install go-licenses
+        run: go install github.com/google/go-licenses@latest
+        # We need this step because of test dependencies and how they are handled in nats.go
+      - name: Run go mod tidy
+        run: go mod tidy
+      - name: Run license check
+        run: go-licenses report ./... --template dependencies.tpl > dependencies.md
+
+      - name: Configure git
+        run: |
+          git config user.name 'github-actions[bot]'
+          git config user.email 'github-actions[bot]@users.noreply.github.com'
+
+      - name: Check for changes
+        id: git_diff
+        run: |
+          git fetch
+          git diff --exit-code dependencies.md || echo "has_changes=true" >> $GITHUB_ENV
+
+      - name: Commit changes
+        if: env.has_changes == 'true'
+        run: |
+          git checkout -b "$BRANCH_NAME"
+          git add dependencies.md
+          git commit -m "Update dependencies.md"
+          git push -u origin "$BRANCH_NAME"
+
+      - name: Create Pull Request
+        if: env.has_changes == 'true'
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          gh pr create --title "Update dependencies.md" --body "This PR updates the dependencies report" --head "$BRANCH_NAME" --base main
+
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index 9a6b4a8..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,36 +0,0 @@
-language: go
-go:
-- "1.22.x"
-- "1.21.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.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;
-    go install github.com/client9/misspell/cmd/misspell@latest;
-  fi
-before_script:
-- $(exit $(go fmt ./... | wc -l))
-- go vet -modfile=go_test.mod ./...
-- 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.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.22 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
-
-jobs:
-  include:
-  - 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.22.x (nats-server@main)"
diff --git a/README.md b/README.md
index fd64d93..84accda 100644
--- a/README.md
+++ b/README.md
@@ -7,8 +7,8 @@ A [Go](http://golang.org) client for the [NATS messaging system](https://nats.io
 [License-Image]: https://img.shields.io/badge/License-Apache2-blue.svg
 [ReportCard-Url]: https://goreportcard.com/report/github.com/nats-io/nats.go
 [ReportCard-Image]: https://goreportcard.com/badge/github.com/nats-io/nats.go
-[Build-Status-Url]: https://travis-ci.com/github/nats-io/nats.go
-[Build-Status-Image]: https://travis-ci.com/nats-io/nats.go.svg?branch=main
+[Build-Status-Url]: https://github.com/nats-io/nats.go/actions
+[Build-Status-Image]: https://github.com/nats-io/nats.go/actions/workflows/ci.yaml/badge.svg?branch=main
 [GoDoc-Url]: https://pkg.go.dev/github.com/nats-io/nats.go
 [GoDoc-Image]: https://img.shields.io/badge/GoDoc-reference-007d9c
 [Coverage-Url]: https://coveralls.io/r/nats-io/nats.go?branch=main
@@ -31,7 +31,7 @@ When using or transitioning to Go modules support:
 ```bash
 # Go client latest or explicit version
 go get github.com/nats-io/nats.go/@latest
-go get github.com/nats-io/nats.go/@v1.36.0
+go get github.com/nats-io/nats.go/@v1.37.0
 
 # For latest NATS Server, add /v2 at the end
 go get github.com/nats-io/nats-server/v2
@@ -93,11 +93,13 @@ nc.Close()
 ```
 
 ## JetStream
+[![JetStream API Reference](https://pkg.go.dev/badge/github.com/nats-io/nats.go/jetstream.svg)](https://pkg.go.dev/github.com/nats-io/nats.go/jetstream)
 
 JetStream is the built-in NATS persistence system. `nats.go` provides a built-in
 API enabling both managing JetStream assets as well as publishing/consuming
 persistent messages.
 
+
 ### Basic usage
 
 ```go
@@ -134,60 +136,6 @@ To find more information on `nats.go` JetStream API, visit
 The service API (`micro`) allows you to [easily build NATS services](micro/README.md) The
 services API is currently in beta release.
 
-## Encoded Connections
-
-```go
-
-nc, _ := nats.Connect(nats.DefaultURL)
-c, _ := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-defer c.Close()
-
-// Simple Publisher
-c.Publish("foo", "Hello World")
-
-// Simple Async Subscriber
-c.Subscribe("foo", func(s string) {
-    fmt.Printf("Received a message: %s\n", s)
-})
-
-// EncodedConn can Publish any raw Go type using the registered Encoder
-type person struct {
-     Name     string
-     Address  string
-     Age      int
-}
-
-// Go type Subscriber
-c.Subscribe("hello", func(p *person) {
-    fmt.Printf("Received a person: %+v\n", p)
-})
-
-me := &person{Name: "derek", Age: 22, Address: "140 New Montgomery Street, San Francisco, CA"}
-
-// Go type Publisher
-c.Publish("hello", me)
-
-// Unsubscribe
-sub, err := c.Subscribe("foo", nil)
-// ...
-sub.Unsubscribe()
-
-// Requests
-var response string
-err = c.Request("help", "help me", &response, 10*time.Millisecond)
-if err != nil {
-    fmt.Printf("Request failed: %v\n", err)
-}
-
-// Replying
-c.Subscribe("help", func(subj, reply string, msg string) {
-    c.Publish(reply, "I can help!")
-})
-
-// Close connection
-c.Close();
-```
-
 ## New Authentication (Nkeys and User Credentials)
 This requires server with version >= 2.0.0
 
@@ -267,34 +215,6 @@ if err != nil {
 
 ```
 
-## Using Go Channels (netchan)
-
-```go
-nc, _ := nats.Connect(nats.DefaultURL)
-ec, _ := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-defer ec.Close()
-
-type person struct {
-     Name     string
-     Address  string
-     Age      int
-}
-
-recvCh := make(chan *person)
-ec.BindRecvChan("hello", recvCh)
-
-sendCh := make(chan *person)
-ec.BindSendChan("hello", sendCh)
-
-me := &person{Name: "derek", Age: 22, Address: "140 New Montgomery Street"}
-
-// Send via Go channels
-sendCh <- me
-
-// Receive via Go channels
-who := <- recvCh
-```
-
 ## Wildcard Subscriptions
 
 ```go
@@ -461,17 +381,6 @@ msg, err := nc.RequestWithContext(ctx, "foo", []byte("bar"))
 sub, err := nc.SubscribeSync("foo")
 msg, err := sub.NextMsgWithContext(ctx)
 
-// Encoded Request with context
-c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-type request struct {
-	Message string `json:"message"`
-}
-type response struct {
-	Code int `json:"code"`
-}
-req := &request{Message: "Hello"}
-resp := &response{}
-err := c.RequestWithContext(ctx, "foo", req, resp)
 ```
 
 ## Backwards compatibility
diff --git a/context.go b/context.go
index 20f1782..c19673c 100644
--- a/context.go
+++ b/context.go
@@ -217,6 +217,8 @@ func (nc *Conn) FlushWithContext(ctx context.Context) error {
 // RequestWithContext will create an Inbox and perform a Request
 // using the provided cancellation context with the Inbox reply
 // for the data v. A response will be decoded into the vPtr last parameter.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) RequestWithContext(ctx context.Context, subject string, v any, vPtr any) error {
 	if ctx == nil {
 		return ErrInvalidContext
diff --git a/dependencies.tpl b/dependencies.tpl
new file mode 100644
index 0000000..5d01380
--- /dev/null
+++ b/dependencies.tpl
@@ -0,0 +1,8 @@
+# External Dependencies
+
+This file lists the dependencies used in this repository.
+
+| Dependency                                       | License                                 |
+|--------------------------------------------------|-----------------------------------------|
+{{ range . }}| {{.Name}}                           | {{.LicenseName}}                        |
+{{ end }}
diff --git a/enc.go b/enc.go
index 4550f61..78bcc21 100644
--- a/enc.go
+++ b/enc.go
@@ -24,7 +24,11 @@ import (
 	"github.com/nats-io/nats.go/encoders/builtin"
 )
 
+//lint:file-ignore SA1019 Ignore deprecation warnings for EncodedConn
+
 // Encoder interface is for all register encoders
+//
+// Deprecated: Encoded connections are no longer supported.
 type Encoder interface {
 	Encode(subject string, v any) ([]byte, error)
 	Decode(subject string, data []byte, vPtr any) error
@@ -51,6 +55,8 @@ func init() {
 // EncodedConn are the preferred way to interface with NATS. They wrap a bare connection to
 // a nats server and have an extendable encoder system that will encode and decode messages
 // from raw Go types.
+//
+// Deprecated: Encoded connections are no longer supported.
 type EncodedConn struct {
 	Conn *Conn
 	Enc  Encoder
@@ -58,6 +64,8 @@ type EncodedConn struct {
 
 // NewEncodedConn will wrap an existing Connection and utilize the appropriate registered
 // encoder.
+//
+// Deprecated: Encoded connections are no longer supported.
 func NewEncodedConn(c *Conn, encType string) (*EncodedConn, error) {
 	if c == nil {
 		return nil, errors.New("nats: Nil Connection")
@@ -73,6 +81,8 @@ func NewEncodedConn(c *Conn, encType string) (*EncodedConn, error) {
 }
 
 // RegisterEncoder will register the encType with the given Encoder. Useful for customization.
+//
+// Deprecated: Encoded connections are no longer supported.
 func RegisterEncoder(encType string, enc Encoder) {
 	encLock.Lock()
 	defer encLock.Unlock()
@@ -80,6 +90,8 @@ func RegisterEncoder(encType string, enc Encoder) {
 }
 
 // EncoderForType will return the registered Encoder for the encType.
+//
+// Deprecated: Encoded connections are no longer supported.
 func EncoderForType(encType string) Encoder {
 	encLock.Lock()
 	defer encLock.Unlock()
@@ -88,6 +100,8 @@ func EncoderForType(encType string) Encoder {
 
 // Publish publishes the data argument to the given subject. The data argument
 // will be encoded using the associated encoder.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) Publish(subject string, v any) error {
 	b, err := c.Enc.Encode(subject, v)
 	if err != nil {
@@ -99,6 +113,8 @@ func (c *EncodedConn) Publish(subject string, v any) error {
 // PublishRequest will perform a Publish() expecting a response on the
 // reply subject. Use Request() for automatically waiting for a response
 // inline.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) PublishRequest(subject, reply string, v any) error {
 	b, err := c.Enc.Encode(subject, v)
 	if err != nil {
@@ -110,6 +126,8 @@ func (c *EncodedConn) PublishRequest(subject, reply string, v any) error {
 // Request will create an Inbox and perform a Request() call
 // with the Inbox reply for the data v. A response will be
 // decoded into the vPtr Response.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) Request(subject string, v any, vPtr any, timeout time.Duration) error {
 	b, err := c.Enc.Encode(subject, v)
 	if err != nil {
@@ -150,6 +168,8 @@ func (c *EncodedConn) Request(subject string, v any, vPtr any, timeout time.Dura
 // and demarshal it into the given struct, e.g. person.
 // There are also variants where the callback wants either the subject, or the
 // subject and the reply subject.
+//
+// Deprecated: Encoded connections are no longer supported.
 type Handler any
 
 // Dissect the cb Handler's signature
@@ -170,6 +190,8 @@ var emptyMsgType = reflect.TypeOf(&Msg{})
 // Subscribe will create a subscription on the given subject and process incoming
 // messages using the specified Handler. The Handler should be a func that matches
 // a signature from the description of Handler from above.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) Subscribe(subject string, cb Handler) (*Subscription, error) {
 	return c.subscribe(subject, _EMPTY_, cb)
 }
@@ -177,6 +199,8 @@ func (c *EncodedConn) Subscribe(subject string, cb Handler) (*Subscription, erro
 // QueueSubscribe will create a queue subscription on the given subject and process
 // incoming messages using the specified Handler. The Handler should be a func that
 // matches a signature from the description of Handler from above.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) QueueSubscribe(subject, queue string, cb Handler) (*Subscription, error) {
 	return c.subscribe(subject, queue, cb)
 }
@@ -238,18 +262,24 @@ func (c *EncodedConn) subscribe(subject, queue string, cb Handler) (*Subscriptio
 }
 
 // FlushTimeout allows a Flush operation to have an associated timeout.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) FlushTimeout(timeout time.Duration) (err error) {
 	return c.Conn.FlushTimeout(timeout)
 }
 
 // Flush will perform a round trip to the server and return when it
 // receives the internal reply.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) Flush() error {
 	return c.Conn.Flush()
 }
 
 // Close will close the connection to the server. This call will release
 // all blocking calls, such as Flush(), etc.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) Close() {
 	c.Conn.Close()
 }
@@ -259,11 +289,15 @@ func (c *EncodedConn) Close() {
 // will be drained and can not publish any additional messages. Upon draining
 // of the publishers, the connection will be closed. Use the ClosedCB()
 // option to know when the connection has moved from draining to closed.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) Drain() error {
 	return c.Conn.Drain()
 }
 
 // LastError reports the last error encountered via the Connection.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) LastError() error {
 	return c.Conn.LastError()
 }
diff --git a/encoders/builtin/default_enc.go b/encoders/builtin/default_enc.go
index c1d0f6f..e73113d 100644
--- a/encoders/builtin/default_enc.go
+++ b/encoders/builtin/default_enc.go
@@ -26,6 +26,8 @@ import (
 // turn numbers into appropriate strings that can be decoded. It will also
 // properly encoded and decode bools. If will encode a struct, but if you want
 // to properly handle structures you should use JsonEncoder.
+//
+// Deprecated: Encoded connections are no longer supported.
 type DefaultEncoder struct {
 	// Empty
 }
@@ -35,6 +37,8 @@ var falseB = []byte("false")
 var nilB = []byte("")
 
 // Encode
+//
+// Deprecated: Encoded connections are no longer supported.
 func (je *DefaultEncoder) Encode(subject string, v any) ([]byte, error) {
 	switch arg := v.(type) {
 	case string:
@@ -58,6 +62,8 @@ func (je *DefaultEncoder) Encode(subject string, v any) ([]byte, error) {
 }
 
 // Decode
+//
+// Deprecated: Encoded connections are no longer supported.
 func (je *DefaultEncoder) Decode(subject string, data []byte, vPtr any) error {
 	// Figure out what it's pointing to...
 	sData := *(*string)(unsafe.Pointer(&data))
diff --git a/encoders/builtin/gob_enc.go b/encoders/builtin/gob_enc.go
index 7ecf85e..e2e8c32 100644
--- a/encoders/builtin/gob_enc.go
+++ b/encoders/builtin/gob_enc.go
@@ -21,6 +21,8 @@ import (
 // GobEncoder is a Go specific GOB Encoder implementation for EncodedConn.
 // This encoder will use the builtin encoding/gob to Marshal
 // and Unmarshal most types, including structs.
+//
+// Deprecated: Encoded connections are no longer supported.
 type GobEncoder struct {
 	// Empty
 }
@@ -28,6 +30,8 @@ type GobEncoder struct {
 // FIXME(dlc) - This could probably be more efficient.
 
 // Encode
+//
+// Deprecated: Encoded connections are no longer supported.
 func (ge *GobEncoder) Encode(subject string, v any) ([]byte, error) {
 	b := new(bytes.Buffer)
 	enc := gob.NewEncoder(b)
@@ -38,6 +42,8 @@ func (ge *GobEncoder) Encode(subject string, v any) ([]byte, error) {
 }
 
 // Decode
+//
+// Deprecated: Encoded connections are no longer supported.
 func (ge *GobEncoder) Decode(subject string, data []byte, vPtr any) (err error) {
 	dec := gob.NewDecoder(bytes.NewBuffer(data))
 	err = dec.Decode(vPtr)
diff --git a/encoders/builtin/json_enc.go b/encoders/builtin/json_enc.go
index 0540d98..8e4c852 100644
--- a/encoders/builtin/json_enc.go
+++ b/encoders/builtin/json_enc.go
@@ -21,11 +21,15 @@ import (
 // JsonEncoder is a JSON Encoder implementation for EncodedConn.
 // This encoder will use the builtin encoding/json to Marshal
 // and Unmarshal most types, including structs.
+//
+// Deprecated: Encoded connections are no longer supported.
 type JsonEncoder struct {
 	// Empty
 }
 
 // Encode
+//
+// Deprecated: Encoded connections are no longer supported.
 func (je *JsonEncoder) Encode(subject string, v any) ([]byte, error) {
 	b, err := json.Marshal(v)
 	if err != nil {
@@ -35,6 +39,8 @@ func (je *JsonEncoder) Encode(subject string, v any) ([]byte, error) {
 }
 
 // Decode
+//
+// Deprecated: Encoded connections are no longer supported.
 func (je *JsonEncoder) Decode(subject string, data []byte, vPtr any) (err error) {
 	switch arg := vPtr.(type) {
 	case *string:
diff --git a/encoders/protobuf/protobuf_enc.go b/encoders/protobuf/protobuf_enc.go
index 017ffc0..8056577 100644
--- a/encoders/protobuf/protobuf_enc.go
+++ b/encoders/protobuf/protobuf_enc.go
@@ -20,6 +20,8 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+//lint:file-ignore SA1019 Ignore deprecation warnings for EncodedConn
+
 // Additional index for registered Encoders.
 const (
 	PROTOBUF_ENCODER = "protobuf"
@@ -33,6 +35,8 @@ func init() {
 // ProtobufEncoder is a protobuf implementation for EncodedConn
 // This encoder will use the builtin protobuf lib to Marshal
 // and Unmarshal structs.
+//
+// Deprecated: Encoded connections are no longer supported.
 type ProtobufEncoder struct {
 	// Empty
 }
@@ -43,6 +47,8 @@ var (
 )
 
 // Encode
+//
+// Deprecated: Encoded connections are no longer supported.
 func (pb *ProtobufEncoder) Encode(subject string, v any) ([]byte, error) {
 	if v == nil {
 		return nil, nil
@@ -60,6 +66,8 @@ func (pb *ProtobufEncoder) Encode(subject string, v any) ([]byte, error) {
 }
 
 // Decode
+//
+// Deprecated: Encoded connections are no longer supported.
 func (pb *ProtobufEncoder) Decode(subject string, data []byte, vPtr any) error {
 	if _, ok := vPtr.(*any); ok {
 		return nil
diff --git a/internal/syncx/map.go b/internal/syncx/map.go
new file mode 100644
index 0000000..d2278e6
--- /dev/null
+++ b/internal/syncx/map.go
@@ -0,0 +1,73 @@
+// Copyright 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
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package syncx
+
+import "sync"
+
+// Map is a type-safe wrapper around sync.Map.
+// It is safe for concurrent use.
+// The zero value of Map is an empty map ready to use.
+type Map[K comparable, V any] struct {
+	m sync.Map
+}
+
+func (m *Map[K, V]) Load(key K) (V, bool) {
+	v, ok := m.m.Load(key)
+	if !ok {
+		var empty V
+		return empty, false
+	}
+	return v.(V), true
+}
+
+func (m *Map[K, V]) Store(key K, value V) {
+	m.m.Store(key, value)
+}
+
+func (m *Map[K, V]) Delete(key K) {
+	m.m.Delete(key)
+}
+
+func (m *Map[K, V]) Range(f func(key K, value V) bool) {
+	m.m.Range(func(key, value any) bool {
+		return f(key.(K), value.(V))
+	})
+}
+
+func (m *Map[K, V]) LoadOrStore(key K, value V) (V, bool) {
+	v, loaded := m.m.LoadOrStore(key, value)
+	return v.(V), loaded
+}
+
+func (m *Map[K, V]) LoadAndDelete(key K) (V, bool) {
+	v, ok := m.m.LoadAndDelete(key)
+	if !ok {
+		var empty V
+		return empty, false
+	}
+	return v.(V), true
+}
+
+func (m *Map[K, V]) CompareAndSwap(key K, old, new V) bool {
+	return m.m.CompareAndSwap(key, old, new)
+}
+
+func (m *Map[K, V]) CompareAndDelete(key K, value V) bool {
+	return m.m.CompareAndDelete(key, value)
+}
+
+func (m *Map[K, V]) Swap(key K, value V) (V, bool) {
+	previous, loaded := m.m.Swap(key, value)
+	return previous.(V), loaded
+}
diff --git a/jetstream/README.md b/jetstream/README.md
index 75b46c7..4889a1b 100644
--- a/jetstream/README.md
+++ b/jetstream/README.md
@@ -1,4 +1,5 @@
-# JetStream Simplified Client
+
+# JetStream Simplified Client [![JetStream API Reference](https://pkg.go.dev/badge/github.com/nats-io/nats.go/jetstream.svg)](https://pkg.go.dev/github.com/nats-io/nats.go/jetstream)
 
 This doc covers the basic usage of the `jetstream` package in `nats.go` client.
 
diff --git a/jetstream/consumer.go b/jetstream/consumer.go
index aa9003f..ee48a1e 100644
--- a/jetstream/consumer.go
+++ b/jetstream/consumer.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/nats-io/nats.go/internal/syncx"
 	"github.com/nats-io/nuid"
 )
 
@@ -233,12 +234,12 @@ func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg Consu
 	}
 
 	return &pullConsumer{
-		jetStream:     js,
-		stream:        stream,
-		name:          resp.Name,
-		durable:       cfg.Durable != "",
-		info:          resp.ConsumerInfo,
-		subscriptions: make(map[string]*pullSubscription),
+		jetStream: js,
+		stream:    stream,
+		name:      resp.Name,
+		durable:   cfg.Durable != "",
+		info:      resp.ConsumerInfo,
+		subs:      syncx.Map[string, *pullSubscription]{},
 	}, nil
 }
 
@@ -285,12 +286,12 @@ func getConsumer(ctx context.Context, js *jetStream, stream, name string) (Consu
 	}
 
 	cons := &pullConsumer{
-		jetStream:     js,
-		stream:        stream,
-		name:          name,
-		durable:       resp.Config.Durable != "",
-		info:          resp.ConsumerInfo,
-		subscriptions: make(map[string]*pullSubscription, 0),
+		jetStream: js,
+		stream:    stream,
+		name:      name,
+		durable:   resp.Config.Durable != "",
+		info:      resp.ConsumerInfo,
+		subs:      syncx.Map[string, *pullSubscription]{},
 	}
 
 	return cons, nil
diff --git a/jetstream/errors.go b/jetstream/errors.go
index fb36434..8d2fec6 100644
--- a/jetstream/errors.go
+++ b/jetstream/errors.go
@@ -270,6 +270,9 @@ var (
 	// of an ordered consumer which was not yet created.
 	ErrOrderedConsumerNotCreated JetStreamError = &jsError{message: "consumer instance not yet created"}
 
+	// ErrJetStreamPublisherClosed is returned for each unfinished ack future when JetStream.Cleanup is called.
+	ErrJetStreamPublisherClosed JetStreamError = &jsError{message: "jetstream context closed"}
+
 	// KeyValue Errors
 
 	// ErrKeyExists is returned when attempting to create a key that already
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index 554400b..3b94146 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -101,6 +101,19 @@ type (
 		// outstanding asynchronously published messages are acknowledged by the
 		// server.
 		PublishAsyncComplete() <-chan struct{}
+
+		// CleanupPublisher will cleanup the publishing side of JetStreamContext.
+		//
+		// This will unsubscribe from the internal reply subject if needed.
+		// All pending async publishes will fail with ErrJetStreamContextClosed.
+		//
+		// If an error handler was provided, it will be called for each pending async
+		// publish and PublishAsyncComplete will be closed.
+		//
+		// After completing JetStreamContext is still usable - internal subscription
+		// will be recreated on next publish, but the acks from previous publishes will
+		// be lost.
+		CleanupPublisher()
 	}
 
 	// StreamManager provides CRUD API for managing streams. It is available as
@@ -109,7 +122,8 @@ type (
 	// to operate on a stream.
 	StreamManager interface {
 		// CreateStream creates a new stream with given config and returns an
-		// interface to operate on it. If stream with given name already exists,
+		// interface to operate on it. If stream with given name already exists
+		// and its configuration differs from the provided one,
 		// ErrStreamNameAlreadyInUse is returned.
 		CreateStream(ctx context.Context, cfg StreamConfig) (Stream, error)
 
@@ -422,7 +436,7 @@ func NewWithAPIPrefix(nc *nats.Conn, apiPrefix string, opts ...JetStreamOpt) (Je
 		}
 	}
 	if apiPrefix == "" {
-		return nil, fmt.Errorf("API prefix cannot be empty")
+		return nil, errors.New("API prefix cannot be empty")
 	}
 	if !strings.HasSuffix(apiPrefix, ".") {
 		jsOpts.apiPrefix = fmt.Sprintf("%s.", apiPrefix)
@@ -1032,6 +1046,39 @@ func wrapContextWithoutDeadline(ctx context.Context) (context.Context, context.C
 	return context.WithTimeout(ctx, defaultAPITimeout)
 }
 
+// CleanupPublisher will cleanup the publishing side of JetStreamContext.
+//
+// This will unsubscribe from the internal reply subject if needed.
+// All pending async publishes will fail with ErrJetStreamContextClosed.
+//
+// If an error handler was provided, it will be called for each pending async
+// publish and PublishAsyncComplete will be closed.
+//
+// After completing JetStreamContext is still usable - internal subscription
+// will be recreated on next publish, but the acks from previous publishes will
+// be lost.
+func (js *jetStream) CleanupPublisher() {
+	js.cleanupReplySub()
+	js.publisher.Lock()
+	errCb := js.publisher.aecb
+	for id, paf := range js.publisher.acks {
+		paf.err = ErrJetStreamPublisherClosed
+		if paf.errCh != nil {
+			paf.errCh <- paf.err
+		}
+		if errCb != nil {
+			// call error handler after releasing the mutex to avoid contention
+			defer errCb(js, paf.msg, ErrJetStreamPublisherClosed)
+		}
+		delete(js.publisher.acks, id)
+	}
+	if js.publisher.doneCh != nil {
+		close(js.publisher.doneCh)
+		js.publisher.doneCh = nil
+	}
+	js.publisher.Unlock()
+}
+
 func (js *jetStream) cleanupReplySub() {
 	if js.publisher == nil {
 		return
diff --git a/jetstream/kv.go b/jetstream/kv.go
index 7a026a2..ae892b4 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -165,8 +165,8 @@ type (
 		// with the same options as Watch.
 		WatchAll(ctx context.Context, opts ...WatchOpt) (KeyWatcher, error)
 
-		// Keys will return all keys. DEPRECATED: Use ListKeys instead to avoid
-		// memory issues.
+		// Keys will return all keys.
+		// Deprecated: Use ListKeys instead to avoid memory issues.
 		Keys(ctx context.Context, opts ...WatchOpt) ([]string, error)
 
 		// ListKeys will return KeyLister, allowing to retrieve all keys from
diff --git a/jetstream/message.go b/jetstream/message.go
index 81e1512..095f139 100644
--- a/jetstream/message.go
+++ b/jetstream/message.go
@@ -16,6 +16,7 @@ package jetstream
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -434,7 +435,7 @@ func parsePending(msg *nats.Msg) (int, int, error) {
 	if msgsLeftStr != "" {
 		msgsLeft, err = strconv.Atoi(msgsLeftStr)
 		if err != nil {
-			return 0, 0, fmt.Errorf("nats: invalid format of Nats-Pending-Messages")
+			return 0, 0, errors.New("nats: invalid format of Nats-Pending-Messages")
 		}
 	}
 	bytesLeftStr := msg.Header.Get("Nats-Pending-Bytes")
@@ -442,7 +443,7 @@ func parsePending(msg *nats.Msg) (int, int, error) {
 	if bytesLeftStr != "" {
 		bytesLeft, err = strconv.Atoi(bytesLeftStr)
 		if err != nil {
-			return 0, 0, fmt.Errorf("nats: invalid format of Nats-Pending-Bytes")
+			return 0, 0, errors.New("nats: invalid format of Nats-Pending-Bytes")
 		}
 	}
 	return msgsLeft, bytesLeft, nil
diff --git a/jetstream/ordered.go b/jetstream/ordered.go
index 998b83d..0d7f952 100644
--- a/jetstream/ordered.go
+++ b/jetstream/ordered.go
@@ -32,12 +32,13 @@ type (
 		cfg               *OrderedConsumerConfig
 		stream            string
 		currentConsumer   *pullConsumer
+		currentSub        *pullSubscription
 		cursor            cursor
 		namePrefix        string
 		serial            int
 		consumerType      consumerType
 		doReset           chan struct{}
-		resetInProgress   uint32
+		resetInProgress   atomic.Uint32
 		userErrHandler    ConsumeErrHandlerFunc
 		stopAfter         int
 		stopAfterMsgsLeft chan int
@@ -51,7 +52,7 @@ type (
 		consumer *orderedConsumer
 		opts     []PullMessagesOpt
 		done     chan struct{}
-		closed   uint32
+		closed   atomic.Uint32
 	}
 
 	cursor struct {
@@ -68,7 +69,10 @@ const (
 	consumerTypeFetch
 )
 
-var errOrderedSequenceMismatch = errors.New("sequence mismatch")
+var (
+	errOrderedSequenceMismatch = errors.New("sequence mismatch")
+	errOrderedConsumerClosed   = errors.New("ordered consumer closed")
+)
 
 // Consume can be used to continuously receive messages and handle them
 // with the provided callback function. Consume cannot be used concurrently
@@ -116,19 +120,11 @@ func (c *orderedConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt
 			}
 			meta, err := msg.Metadata()
 			if err != nil {
-				sub, ok := c.currentConsumer.getSubscription("")
-				if !ok {
-					return
-				}
-				c.errHandler(serial)(sub, err)
+				c.errHandler(serial)(c.currentSub, err)
 				return
 			}
 			dseq := meta.Sequence.Consumer
 			if dseq != c.cursor.deliverSeq+1 {
-				sub, ok := c.currentConsumer.getSubscription("")
-				if !ok {
-					return
-				}
 				c.errHandler(serial)(sub, errOrderedSequenceMismatch)
 				return
 			}
@@ -138,21 +134,21 @@ func (c *orderedConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt
 		}
 	}
 
-	_, err = c.currentConsumer.Consume(internalHandler(c.serial), opts...)
+	cc, err := c.currentConsumer.Consume(internalHandler(c.serial), opts...)
 	if err != nil {
 		return nil, err
 	}
+	c.currentSub = cc.(*pullSubscription)
 
 	go func() {
 		for {
 			select {
 			case <-c.doReset:
 				if err := c.reset(); err != nil {
-					sub, ok := c.currentConsumer.getSubscription("")
-					if !ok {
-						return
+					if errors.Is(err, errOrderedConsumerClosed) {
+						continue
 					}
-					c.errHandler(c.serial)(sub, err)
+					c.errHandler(c.serial)(c.currentSub, err)
 				}
 				if c.withStopAfter {
 					select {
@@ -175,14 +171,20 @@ func (c *orderedConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt
 				if c.withStopAfter {
 					opts = append(opts, consumeStopAfterNotify(c.stopAfter, c.stopAfterMsgsLeft))
 				}
-				if _, err := c.currentConsumer.Consume(internalHandler(c.serial), opts...); err != nil {
-					sub, ok := c.currentConsumer.getSubscription("")
-					if !ok {
-						return
-					}
-					c.errHandler(c.serial)(sub, err)
+				if cc, err := c.currentConsumer.Consume(internalHandler(c.serial), opts...); err != nil {
+					c.errHandler(c.serial)(cc, err)
+				} else {
+					c.Lock()
+					c.currentSub = cc.(*pullSubscription)
+					c.Unlock()
 				}
 			case <-sub.done:
+				s := sub.consumer.currentSub
+				if s != nil {
+					sub.consumer.Lock()
+					s.Stop()
+					sub.consumer.Unlock()
+				}
 				return
 			case msgsLeft, ok := <-c.stopAfterMsgsLeft:
 				if !ok {
@@ -208,8 +210,8 @@ func (c *orderedConsumer) errHandler(serial int) func(cc ConsumeContext, err err
 			errors.Is(err, ErrConsumerDeleted) ||
 			errors.Is(err, errConnected) {
 			// only reset if serial matches the current consumer serial and there is no reset in progress
-			if serial == c.serial && atomic.LoadUint32(&c.resetInProgress) == 0 {
-				atomic.StoreUint32(&c.resetInProgress, 1)
+			if serial == c.serial && c.resetInProgress.Load() == 0 {
+				c.resetInProgress.Store(1)
 				c.doReset <- struct{}{}
 			}
 		}
@@ -250,10 +252,11 @@ func (c *orderedConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, er
 	if c.stopAfter > 0 {
 		opts = append(opts, messagesStopAfterNotify(c.stopAfter, c.stopAfterMsgsLeft))
 	}
-	_, err = c.currentConsumer.Messages(opts...)
+	cc, err := c.currentConsumer.Messages(opts...)
 	if err != nil {
 		return nil, err
 	}
+	c.currentSub = cc.(*pullSubscription)
 
 	sub := &orderedSubscription{
 		consumer: c,
@@ -267,12 +270,7 @@ func (c *orderedConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, er
 
 func (s *orderedSubscription) Next() (Msg, error) {
 	for {
-		currentConsumer := s.consumer.currentConsumer
-		sub, ok := currentConsumer.getSubscription("")
-		if !ok {
-			return nil, ErrMsgIteratorClosed
-		}
-		msg, err := sub.Next()
+		msg, err := s.consumer.currentSub.Next()
 		if err != nil {
 			if errors.Is(err, ErrMsgIteratorClosed) {
 				s.Stop()
@@ -290,12 +288,16 @@ func (s *orderedSubscription) Next() (Msg, error) {
 				s.opts[len(s.opts)-1] = StopAfter(s.consumer.stopAfter)
 			}
 			if err := s.consumer.reset(); err != nil {
+				if errors.Is(err, errOrderedConsumerClosed) {
+					return nil, ErrMsgIteratorClosed
+				}
 				return nil, err
 			}
-			_, err := s.consumer.currentConsumer.Messages(s.opts...)
+			cc, err := s.consumer.currentConsumer.Messages(s.opts...)
 			if err != nil {
 				return nil, err
 			}
+			s.consumer.currentSub = cc.(*pullSubscription)
 			continue
 		}
 
@@ -310,12 +312,16 @@ func (s *orderedSubscription) Next() (Msg, error) {
 		dseq := meta.Sequence.Consumer
 		if dseq != s.consumer.cursor.deliverSeq+1 {
 			if err := s.consumer.reset(); err != nil {
+				if errors.Is(err, errOrderedConsumerClosed) {
+					return nil, ErrMsgIteratorClosed
+				}
 				return nil, err
 			}
-			_, err := s.consumer.currentConsumer.Messages(s.opts...)
+			cc, err := s.consumer.currentConsumer.Messages(s.opts...)
 			if err != nil {
 				return nil, err
 			}
+			s.consumer.currentSub = cc.(*pullSubscription)
 			continue
 		}
 		s.consumer.cursor.deliverSeq = dseq
@@ -325,33 +331,60 @@ func (s *orderedSubscription) Next() (Msg, error) {
 }
 
 func (s *orderedSubscription) Stop() {
-	if !atomic.CompareAndSwapUint32(&s.closed, 0, 1) {
+	if !s.closed.CompareAndSwap(0, 1) {
 		return
 	}
-	sub, ok := s.consumer.currentConsumer.getSubscription("")
-	if !ok {
-		return
+	s.consumer.Lock()
+	defer s.consumer.Unlock()
+	if s.consumer.currentSub != nil {
+		s.consumer.currentSub.Stop()
 	}
-	s.consumer.currentConsumer.Lock()
-	defer s.consumer.currentConsumer.Unlock()
-	sub.Stop()
 	close(s.done)
 }
 
 func (s *orderedSubscription) Drain() {
-	if !atomic.CompareAndSwapUint32(&s.closed, 0, 1) {
+	if !s.closed.CompareAndSwap(0, 1) {
 		return
 	}
-	sub, ok := s.consumer.currentConsumer.getSubscription("")
-	if !ok {
-		return
+	if s.consumer.currentSub != nil {
+		s.consumer.currentConsumer.Lock()
+		s.consumer.currentSub.Drain()
+		s.consumer.currentConsumer.Unlock()
 	}
-	s.consumer.currentConsumer.Lock()
-	defer s.consumer.currentConsumer.Unlock()
-	sub.Drain()
 	close(s.done)
 }
 
+// Closed returns a channel that is closed when the consuming is
+// fully stopped/drained. When the channel is closed, no more messages
+// will be received and processing is complete.
+func (s *orderedSubscription) Closed() <-chan struct{} {
+	s.consumer.Lock()
+	defer s.consumer.Unlock()
+	closedCh := make(chan struct{})
+
+	go func() {
+		for {
+			s.consumer.Lock()
+			if s.consumer.currentSub == nil {
+				return
+			}
+
+			closed := s.consumer.currentSub.Closed()
+			s.consumer.Unlock()
+
+			// wait until the underlying pull consumer is closed
+			<-closed
+			// if the subscription is closed and ordered consumer is closed as well,
+			// send a signal that the Consume() is fully stopped
+			if s.closed.Load() == 1 {
+				close(closedCh)
+				return
+			}
+		}
+	}()
+	return closedCh
+}
+
 // Fetch is used to retrieve up to a provided number of messages from a
 // stream. This method will always send a single request and wait until
 // either all messages are retrieved or request times out.
@@ -493,12 +526,11 @@ func serialNumberFromConsumer(name string) int {
 func (c *orderedConsumer) reset() error {
 	c.Lock()
 	defer c.Unlock()
-	defer atomic.StoreUint32(&c.resetInProgress, 0)
+	defer c.resetInProgress.Store(0)
 	if c.currentConsumer != nil {
-		sub, ok := c.currentConsumer.getSubscription("")
 		c.currentConsumer.Lock()
-		if ok {
-			sub.Stop()
+		if c.currentSub != nil {
+			c.currentSub.Stop()
 		}
 		consName := c.currentConsumer.CachedInfo().Name
 		c.currentConsumer.Unlock()
@@ -523,9 +555,9 @@ func (c *orderedConsumer) reset() error {
 		cancel:          c.subscription.done,
 	}
 	err = retryWithBackoff(func(attempt int) (bool, error) {
-		isClosed := atomic.LoadUint32(&c.subscription.closed) == 1
+		isClosed := c.subscription.closed.Load() == 1
 		if isClosed {
-			return false, nil
+			return false, errOrderedConsumerClosed
 		}
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()
@@ -533,7 +565,6 @@ func (c *orderedConsumer) reset() error {
 		if err != nil {
 			return true, err
 		}
-		c.currentConsumer = cons.(*pullConsumer)
 		return false, nil
 	}, backoffOpts)
 	if err != nil {
@@ -583,7 +614,8 @@ func (c *orderedConsumer) getConsumerConfig() *ConsumerConfig {
 		cfg.InactiveThreshold = c.cfg.InactiveThreshold
 	}
 
-	if c.serial != 1 {
+	// if the cursor is not yet set, use the provided deliver policy
+	if c.cursor.streamSeq != 0 {
 		return cfg
 	}
 
@@ -715,7 +747,7 @@ func retryWithBackoff(f func(int) (bool, error), opts backoffOpts) error {
 	// if custom backoff is set, use it instead of other options
 	if len(opts.customBackoff) > 0 {
 		if opts.attempts != 0 {
-			return fmt.Errorf("cannot use custom backoff intervals when attempts are set")
+			return errors.New("cannot use custom backoff intervals when attempts are set")
 		}
 		for i, interval := range opts.customBackoff {
 			select {
@@ -742,7 +774,7 @@ func retryWithBackoff(f func(int) (bool, error), opts backoffOpts) error {
 		opts.maxInterval = 1 * time.Minute
 	}
 	if opts.attempts == 0 {
-		return fmt.Errorf("retry attempts have to be set when not using custom backoff intervals")
+		return errors.New("retry attempts have to be set when not using custom backoff intervals")
 	}
 	interval := opts.initialInterval
 	for i := 0; ; i++ {
diff --git a/jetstream/publish.go b/jetstream/publish.go
index 1a3a4fd..d08cdf1 100644
--- a/jetstream/publish.go
+++ b/jetstream/publish.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2023 The NATS Authors
+// Copyright 2022-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
@@ -467,11 +467,7 @@ func (js *jetStream) resetPendingAcksOnReconnect() {
 				paf.errCh <- paf.err
 			}
 			if errCb != nil {
-				js.publisher.Unlock()
-				// clear reply subject so that new one is created on republish
-				paf.msg.Reply = ""
-				errCb(js, paf.msg, nats.ErrDisconnected)
-				js.publisher.Lock()
+				defer errCb(js, paf.msg, nats.ErrDisconnected)
 			}
 			delete(js.publisher.acks, id)
 		}
diff --git a/jetstream/pull.go b/jetstream/pull.go
index 001a0d1..3869681 100644
--- a/jetstream/pull.go
+++ b/jetstream/pull.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/internal/syncx"
 	"github.com/nats-io/nuid"
 )
 
@@ -58,6 +59,11 @@ type (
 		// Drain unsubscribes from the stream and cancels subscription.
 		// All messages that are already in the buffer will be processed in callback function.
 		Drain()
+
+		// Closed returns a channel that is closed when the consuming is
+		// fully stopped/drained. When the channel is closed, no more messages
+		// will be received and processing is complete.
+		Closed() <-chan struct{}
 	}
 
 	// MessageHandler is a handler function used as callback in [Consume].
@@ -75,12 +81,12 @@ type (
 
 	pullConsumer struct {
 		sync.Mutex
-		jetStream     *jetStream
-		stream        string
-		durable       bool
-		name          string
-		info          *ConsumerInfo
-		subscriptions map[string]*pullSubscription
+		jetStream *jetStream
+		stream    string
+		durable   bool
+		name      string
+		info      *ConsumerInfo
+		subs      syncx.Map[string, *pullSubscription]
 	}
 
 	pullRequest struct {
@@ -116,14 +122,15 @@ type (
 		errs              chan error
 		pending           pendingMsgs
 		hbMonitor         *hbMonitor
-		fetchInProgress   uint32
-		closed            uint32
-		draining          uint32
+		fetchInProgress   atomic.Uint32
+		closed            atomic.Uint32
+		draining          atomic.Uint32
 		done              chan struct{}
 		connStatusChanged chan nats.Status
 		fetchNext         chan *pullRequest
 		consumeOpts       *consumeOpts
 		delivered         int
+		closedCh          chan struct{}
 	}
 
 	pendingMsgs struct {
@@ -181,12 +188,7 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 
 	subject := apiSubj(p.jetStream.apiPrefix, fmt.Sprintf(apiRequestNextT, p.stream, p.name))
 
-	// for single consume, use empty string as id
-	// this is useful for ordered consumer, where only a single subscription is valid
-	var consumeID string
-	if len(p.subscriptions) > 0 {
-		consumeID = nuid.Next()
-	}
+	consumeID := nuid.Next()
 	sub := &pullSubscription{
 		id:          consumeID,
 		consumer:    p,
@@ -199,7 +201,7 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 
 	sub.hbMonitor = sub.scheduleHeartbeatCheck(consumeOpts.Heartbeat)
 
-	p.subscriptions[sub.id] = sub
+	p.subs.Store(sub.id, sub)
 	p.Unlock()
 
 	internalHandler := func(msg *nats.Msg) {
@@ -232,7 +234,7 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 			sub.Unlock()
 
 			if err != nil {
-				if atomic.LoadUint32(&sub.closed) == 1 {
+				if sub.closed.Load() == 1 {
 					return
 				}
 				if sub.consumeOpts.ErrHandler != nil {
@@ -259,10 +261,14 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 	}
 	sub.subscription.SetClosedHandler(func(sid string) func(string) {
 		return func(subject string) {
-			p.Lock()
-			defer p.Unlock()
-			delete(p.subscriptions, sid)
-			atomic.CompareAndSwapUint32(&sub.draining, 1, 0)
+			p.subs.Delete(sid)
+			sub.draining.CompareAndSwap(1, 0)
+			sub.Lock()
+			if sub.closedCh != nil {
+				close(sub.closedCh)
+				sub.closedCh = nil
+			}
+			sub.Unlock()
 		}
 	}(sub.id))
 
@@ -286,7 +292,7 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 	go func() {
 		isConnected := true
 		for {
-			if atomic.LoadUint32(&sub.closed) == 1 {
+			if sub.closed.Load() == 1 {
 				return
 			}
 			select {
@@ -383,7 +389,7 @@ func (s *pullSubscription) incrementDeliveredMsgs() {
 func (s *pullSubscription) checkPending() {
 	if (s.pending.msgCount < s.consumeOpts.ThresholdMessages ||
 		(s.pending.byteCount < s.consumeOpts.ThresholdBytes && s.consumeOpts.MaxBytes != 0)) &&
-		atomic.LoadUint32(&s.fetchInProgress) == 0 {
+		s.fetchInProgress.Load() == 0 {
 
 		var batchSize, maxBytes int
 		if s.consumeOpts.MaxBytes == 0 {
@@ -427,12 +433,7 @@ func (p *pullConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, error
 
 	msgs := make(chan *nats.Msg, consumeOpts.MaxMessages)
 
-	// for single consume, use empty string as id
-	// this is useful for ordered consumer, where only a single subscription is valid
-	var consumeID string
-	if len(p.subscriptions) > 0 {
-		consumeID = nuid.Next()
-	}
+	consumeID := nuid.Next()
 	sub := &pullSubscription{
 		id:          consumeID,
 		consumer:    p,
@@ -451,20 +452,18 @@ func (p *pullConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, error
 	}
 	sub.subscription.SetClosedHandler(func(sid string) func(string) {
 		return func(subject string) {
-			p.Lock()
-			defer p.Unlock()
-			if atomic.LoadUint32(&sub.draining) != 1 {
+			if sub.draining.Load() != 1 {
 				// if we're not draining, subscription can be closed as soon
 				// as closed handler is called
 				// otherwise, we need to wait until all messages are drained
 				// in Next
-				delete(p.subscriptions, sid)
+				p.subs.Delete(sid)
 			}
 			close(msgs)
 		}
 	}(sub.id))
 
-	p.subscriptions[sub.id] = sub
+	p.subs.Store(sub.id, sub)
 	p.Unlock()
 
 	go sub.pullMessages(subject)
@@ -502,8 +501,8 @@ var (
 func (s *pullSubscription) Next() (Msg, error) {
 	s.Lock()
 	defer s.Unlock()
-	drainMode := atomic.LoadUint32(&s.draining) == 1
-	closed := atomic.LoadUint32(&s.closed) == 1
+	drainMode := s.draining.Load() == 1
+	closed := s.closed.Load() == 1
 	if closed && !drainMode {
 		return nil, ErrMsgIteratorClosed
 	}
@@ -526,8 +525,8 @@ func (s *pullSubscription) Next() (Msg, error) {
 		case msg, ok := <-s.msgs:
 			if !ok {
 				// if msgs channel is closed, it means that subscription was either drained or stopped
-				delete(s.consumer.subscriptions, s.id)
-				atomic.CompareAndSwapUint32(&s.draining, 1, 0)
+				s.consumer.subs.Delete(s.id)
+				s.draining.CompareAndSwap(1, 0)
 				return nil, ErrMsgIteratorClosed
 			}
 			if hbMonitor != nil {
@@ -630,7 +629,7 @@ func (hb *hbMonitor) Reset(dur time.Duration) {
 // Next after calling Stop will return ErrMsgIteratorClosed error.
 // All messages that are already in the buffer are discarded.
 func (s *pullSubscription) Stop() {
-	if !atomic.CompareAndSwapUint32(&s.closed, 0, 1) {
+	if !s.closed.CompareAndSwap(0, 1) {
 		return
 	}
 	close(s.done)
@@ -648,10 +647,10 @@ func (s *pullSubscription) Stop() {
 // subsequent calls to Next. After the buffer is drained, Next will
 // return ErrMsgIteratorClosed error.
 func (s *pullSubscription) Drain() {
-	if !atomic.CompareAndSwapUint32(&s.closed, 0, 1) {
+	if !s.closed.CompareAndSwap(0, 1) {
 		return
 	}
-	atomic.StoreUint32(&s.draining, 1)
+	s.draining.Store(1)
 	close(s.done)
 	if s.consumeOpts.stopAfterMsgsLeft != nil {
 		if s.delivered >= s.consumeOpts.StopAfter {
@@ -662,6 +661,24 @@ func (s *pullSubscription) Drain() {
 	}
 }
 
+// Closed returns a channel that is closed when consuming is
+// fully stopped/drained. When the channel is closed, no more messages
+// will be received and processing is complete.
+func (s *pullSubscription) Closed() <-chan struct{} {
+	s.Lock()
+	defer s.Unlock()
+	closedCh := s.closedCh
+	if closedCh == nil {
+		closedCh = make(chan struct{})
+		s.closedCh = closedCh
+	}
+	if !s.subscription.IsValid() {
+		close(s.closedCh)
+		s.closedCh = nil
+	}
+	return closedCh
+}
+
 // Fetch sends a single request to retrieve given number of messages.
 // It will wait up to provided expiry time if not all messages are available.
 func (p *pullConsumer) Fetch(batch int, opts ...FetchOpt) (MessageBatch, error) {
@@ -840,7 +857,7 @@ func (s *pullSubscription) pullMessages(subject string) {
 	for {
 		select {
 		case req := <-s.fetchNext:
-			atomic.StoreUint32(&s.fetchInProgress, 1)
+			s.fetchInProgress.Store(1)
 
 			if err := s.pull(req, subject); err != nil {
 				if errors.Is(err, ErrMsgIteratorClosed) {
@@ -849,7 +866,7 @@ func (s *pullSubscription) pullMessages(subject string) {
 				}
 				s.errs <- err
 			}
-			atomic.StoreUint32(&s.fetchInProgress, 0)
+			s.fetchInProgress.Store(0)
 		case <-s.done:
 			s.cleanup()
 			return
@@ -880,13 +897,13 @@ func (s *pullSubscription) cleanup() {
 	if s.hbMonitor != nil {
 		s.hbMonitor.Stop()
 	}
-	drainMode := atomic.LoadUint32(&s.draining) == 1
+	drainMode := s.draining.Load() == 1
 	if drainMode {
 		s.subscription.Drain()
 	} else {
 		s.subscription.Unsubscribe()
 	}
-	atomic.StoreUint32(&s.closed, 1)
+	s.closed.Store(1)
 }
 
 // pull sends a pull request to the server and waits for messages using a subscription from [pullSubscription].
@@ -894,7 +911,7 @@ func (s *pullSubscription) cleanup() {
 func (s *pullSubscription) pull(req *pullRequest, subject string) error {
 	s.consumer.Lock()
 	defer s.consumer.Unlock()
-	if atomic.LoadUint32(&s.closed) == 1 {
+	if s.closed.Load() == 1 {
 		return ErrMsgIteratorClosed
 	}
 	if req.Batch < 1 {
@@ -954,7 +971,7 @@ func parseMessagesOpts(ordered bool, opts ...PullMessagesOpt) (*consumeOpts, err
 
 func (consumeOpts *consumeOpts) setDefaults(ordered bool) error {
 	if consumeOpts.MaxBytes != unset && consumeOpts.MaxMessages != unset {
-		return fmt.Errorf("only one of MaxMessages and MaxBytes can be specified")
+		return errors.New("only one of MaxMessages and MaxBytes can be specified")
 	}
 	if consumeOpts.MaxBytes != unset {
 		// when max_bytes is used, set batch size to a very large number
@@ -990,14 +1007,7 @@ func (consumeOpts *consumeOpts) setDefaults(ordered bool) error {
 		}
 	}
 	if consumeOpts.Heartbeat > consumeOpts.Expires/2 {
-		return fmt.Errorf("the value of Heartbeat must be less than 50%% of expiry")
+		return errors.New("the value of Heartbeat must be less than 50%% of expiry")
 	}
 	return nil
 }
-
-func (c *pullConsumer) getSubscription(id string) (*pullSubscription, bool) {
-	c.Lock()
-	defer c.Unlock()
-	sub, ok := c.subscriptions[id]
-	return sub, ok
-}
diff --git a/jetstream/stream.go b/jetstream/stream.go
index 4741a51..1cd9975 100644
--- a/jetstream/stream.go
+++ b/jetstream/stream.go
@@ -504,16 +504,16 @@ func convertDirectGetMsgResponseToMsg(name string, r *nats.Msg) (*RawStreamMsg,
 	// Check for headers that give us the required information to
 	// reconstruct the message.
 	if len(r.Header) == 0 {
-		return nil, fmt.Errorf("nats: response should have headers")
+		return nil, errors.New("nats: response should have headers")
 	}
 	stream := r.Header.Get(StreamHeader)
 	if stream == "" {
-		return nil, fmt.Errorf("nats: missing stream header")
+		return nil, errors.New("nats: missing stream header")
 	}
 
 	seqStr := r.Header.Get(SequenceHeader)
 	if seqStr == "" {
-		return nil, fmt.Errorf("nats: missing sequence header")
+		return nil, errors.New("nats: missing sequence header")
 	}
 	seq, err := strconv.ParseUint(seqStr, 10, 64)
 	if err != nil {
@@ -521,7 +521,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *nats.Msg) (*RawStreamMsg,
 	}
 	timeStr := r.Header.Get(TimeStampHeaer)
 	if timeStr == "" {
-		return nil, fmt.Errorf("nats: missing timestamp header")
+		return nil, errors.New("nats: missing timestamp header")
 	}
 
 	tm, err := time.Parse(time.RFC3339Nano, timeStr)
@@ -530,7 +530,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *nats.Msg) (*RawStreamMsg,
 	}
 	subj := r.Header.Get(SubjectHeader)
 	if subj == "" {
-		return nil, fmt.Errorf("nats: missing subject header")
+		return nil, errors.New("nats: missing subject header")
 	}
 	return &RawStreamMsg{
 		Subject:  subj,
diff --git a/jetstream/stream_config.go b/jetstream/stream_config.go
index dd1f9d9..304203b 100644
--- a/jetstream/stream_config.go
+++ b/jetstream/stream_config.go
@@ -15,6 +15,7 @@ package jetstream
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 	"time"
@@ -192,8 +193,8 @@ type (
 		// v2.10.0 or later.
 		Metadata map[string]string `json:"metadata,omitempty"`
 
-		// Template identifies the template that manages the Stream. DEPRECATED:
-		// This feature is no longer supported.
+		// Template identifies the template that manages the Stream.
+		// Deprecated: This feature is no longer supported.
 		Template string `json:"template_owner,omitempty"`
 	}
 
@@ -584,7 +585,7 @@ func (alg StoreCompression) MarshalJSON() ([]byte, error) {
 	case NoCompression:
 		str = "none"
 	default:
-		return nil, fmt.Errorf("unknown compression algorithm")
+		return nil, errors.New("unknown compression algorithm")
 	}
 	return json.Marshal(str)
 }
@@ -600,7 +601,7 @@ func (alg *StoreCompression) UnmarshalJSON(b []byte) error {
 	case "none":
 		*alg = NoCompression
 	default:
-		return fmt.Errorf("unknown compression algorithm")
+		return errors.New("unknown compression algorithm")
 	}
 	return nil
 }
diff --git a/js.go b/js.go
index 5f8dfe3..e024fae 100644
--- a/js.go
+++ b/js.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
@@ -58,6 +58,19 @@ type JetStream interface {
 	// PublishAsyncComplete returns a channel that will be closed when all outstanding messages are ack'd.
 	PublishAsyncComplete() <-chan struct{}
 
+	// CleanupPublisher will cleanup the publishing side of JetStreamContext.
+	//
+	// This will unsubscribe from the internal reply subject if needed.
+	// All pending async publishes will fail with ErrJetStreamPublisherClosed.
+	//
+	// If an error handler was provided, it will be called for each pending async
+	// publish and PublishAsyncComplete will be closed.
+	//
+	// After completing JetStreamContext is still usable - internal subscription
+	// will be recreated on next publish, but the acks from previous publishes will
+	// be lost.
+	CleanupPublisher()
+
 	// Subscribe creates an async Subscription for JetStream.
 	// The stream and consumer names can be provided with the nats.Bind() option.
 	// For creating an ephemeral (where the consumer name is picked by the server),
@@ -466,6 +479,9 @@ type pubOpts struct {
 
 	// stallWait is the max wait of a async pub ack.
 	stallWait time.Duration
+
+	// internal option to re-use existing paf in case of retry.
+	pafRetry *pubAckFuture
 }
 
 // pubAckResponse is the ack response from the JetStream API when publishing a message.
@@ -531,7 +547,7 @@ func (js *js) PublishMsg(m *Msg, opts ...PubOpt) (*PubAck, error) {
 		o.ttl = js.opts.wait
 	}
 	if o.stallWait > 0 {
-		return nil, fmt.Errorf("nats: stall wait cannot be set to sync publish")
+		return nil, errors.New("nats: stall wait cannot be set to sync publish")
 	}
 
 	if o.id != _EMPTY_ {
@@ -620,13 +636,17 @@ type PubAckFuture interface {
 }
 
 type pubAckFuture struct {
-	js     *js
-	msg    *Msg
-	pa     *PubAck
-	st     time.Time
-	err    error
-	errCh  chan error
-	doneCh chan *PubAck
+	js         *js
+	msg        *Msg
+	pa         *PubAck
+	st         time.Time
+	err        error
+	errCh      chan error
+	doneCh     chan *PubAck
+	retries    int
+	maxRetries int
+	retryWait  time.Duration
+	reply      string
 }
 
 func (paf *pubAckFuture) Ok() <-chan *PubAck {
@@ -719,10 +739,7 @@ func (js *js) resetPendingAcksOnReconnect() {
 				paf.errCh <- paf.err
 			}
 			if errCb != nil {
-				// clear reply subject so that new one is created on republish
-				js.mu.Unlock()
-				errCb(js, paf.msg, ErrDisconnected)
-				js.mu.Lock()
+				defer errCb(js, paf.msg, ErrDisconnected)
 			}
 			delete(js.pafs, id)
 		}
@@ -734,6 +751,38 @@ func (js *js) resetPendingAcksOnReconnect() {
 	}
 }
 
+// CleanupPublisher will cleanup the publishing side of JetStreamContext.
+//
+// This will unsubscribe from the internal reply subject if needed.
+// All pending async publishes will fail with ErrJetStreamContextClosed.
+//
+// If an error handler was provided, it will be called for each pending async
+// publish and PublishAsyncComplete will be closed.
+//
+// After completing JetStreamContext is still usable - internal subscription
+// will be recreated on next publish, but the acks from previous publishes will
+// be lost.
+func (js *js) CleanupPublisher() {
+	js.cleanupReplySub()
+	js.mu.Lock()
+	errCb := js.opts.aecb
+	for id, paf := range js.pafs {
+		paf.err = ErrJetStreamPublisherClosed
+		if paf.errCh != nil {
+			paf.errCh <- paf.err
+		}
+		if errCb != nil {
+			defer errCb(js, paf.msg, ErrJetStreamPublisherClosed)
+		}
+		delete(js.pafs, id)
+	}
+	if js.dch != nil {
+		close(js.dch)
+		js.dch = nil
+	}
+	js.mu.Unlock()
+}
+
 func (js *js) cleanupReplySub() {
 	js.mu.Lock()
 	if js.rsub != nil {
@@ -806,20 +855,30 @@ func (js *js) handleAsyncReply(m *Msg) {
 		js.mu.Unlock()
 		return
 	}
-	// Remove
-	delete(js.pafs, id)
 
-	// Check on anyone stalled and waiting.
-	if js.stc != nil && len(js.pafs) < js.opts.maxpa {
-		close(js.stc)
-		js.stc = nil
+	closeStc := func() {
+		// Check on anyone stalled and waiting.
+		if js.stc != nil && len(js.pafs) < js.opts.maxpa {
+			close(js.stc)
+			js.stc = nil
+		}
 	}
-	// Check on anyone one waiting on done status.
-	if js.dch != nil && len(js.pafs) == 0 {
-		dch := js.dch
-		js.dch = nil
-		// Defer here so error is processed and can be checked.
-		defer close(dch)
+
+	closeDchFn := func() func() {
+		var dch chan struct{}
+		// Check on anyone one waiting on done status.
+		if js.dch != nil && len(js.pafs) == 0 {
+			dch = js.dch
+			js.dch = nil
+		}
+		// Return function to close done channel which
+		// should be deferred so that error is processed and
+		// can be checked.
+		return func() {
+			if dch != nil {
+				close(dch)
+			}
+		}
 	}
 
 	doErr := func(err error) {
@@ -836,10 +895,39 @@ func (js *js) handleAsyncReply(m *Msg) {
 
 	// Process no responders etc.
 	if len(m.Data) == 0 && m.Header.Get(statusHdr) == noResponders {
+		if paf.retries < paf.maxRetries {
+			paf.retries++
+			time.AfterFunc(paf.retryWait, func() {
+				js.mu.Lock()
+				paf := js.getPAF(id)
+				js.mu.Unlock()
+				if paf == nil {
+					return
+				}
+				_, err := js.PublishMsgAsync(paf.msg, pubOptFn(func(po *pubOpts) error {
+					po.pafRetry = paf
+					return nil
+				}))
+				if err != nil {
+					js.mu.Lock()
+					doErr(err)
+				}
+			})
+			js.mu.Unlock()
+			return
+		}
+		delete(js.pafs, id)
+		closeStc()
+		defer closeDchFn()()
 		doErr(ErrNoResponders)
 		return
 	}
 
+	//remove
+	delete(js.pafs, id)
+	closeStc()
+	defer closeDchFn()()
+
 	var pa pubAckResponse
 	if err := json.Unmarshal(m.Data, &pa); err != nil {
 		doErr(ErrInvalidJSAck)
@@ -906,6 +994,10 @@ func (js *js) PublishMsgAsync(m *Msg, opts ...PubOpt) (PubAckFuture, error) {
 		}
 	}
 
+	if o.rnum < 0 {
+		return nil, fmt.Errorf("%w: retry attempts cannot be negative", ErrInvalidArg)
+	}
+
 	// Timeouts and contexts do not make sense for these.
 	if o.ttl != 0 || o.ctx != nil {
 		return nil, ErrContextAndTimeout
@@ -933,30 +1025,42 @@ func (js *js) PublishMsgAsync(m *Msg, opts ...PubOpt) (PubAckFuture, error) {
 	}
 
 	// Reply
-	if m.Reply != _EMPTY_ {
+	paf := o.pafRetry
+	if paf == nil && m.Reply != _EMPTY_ {
 		return nil, errors.New("nats: reply subject should be empty")
 	}
-	reply := m.Reply
-	m.Reply = js.newAsyncReply()
-	defer func() { m.Reply = reply }()
+	var id string
+	var reply string
 
-	if m.Reply == _EMPTY_ {
-		return nil, errors.New("nats: error creating async reply handler")
-	}
+	// register new paf if not retrying
+	if paf == nil {
+		reply = js.newAsyncReply()
+
+		if reply == _EMPTY_ {
+			return nil, errors.New("nats: error creating async reply handler")
+		}
 
-	id := m.Reply[js.replyPrefixLen:]
-	paf := &pubAckFuture{msg: m, st: time.Now()}
-	numPending, maxPending := js.registerPAF(id, paf)
+		id = reply[js.replyPrefixLen:]
+		paf = &pubAckFuture{msg: m, st: time.Now(), maxRetries: o.rnum, retryWait: o.rwait, reply: reply}
+		numPending, maxPending := js.registerPAF(id, paf)
 
-	if maxPending > 0 && numPending >= maxPending {
-		select {
-		case <-js.asyncStall():
-		case <-time.After(stallWait):
-			js.clearPAF(id)
-			return nil, errors.New("nats: stalled with too many outstanding async published messages")
+		if maxPending > 0 && numPending > maxPending {
+			select {
+			case <-js.asyncStall():
+			case <-time.After(stallWait):
+				js.clearPAF(id)
+				return nil, errors.New("nats: stalled with too many outstanding async published messages")
+			}
 		}
+	} else {
+		reply = paf.reply
+		id = reply[js.replyPrefixLen:]
+	}
+	hdr, err := m.headerBytes()
+	if err != nil {
+		return nil, err
 	}
-	if err := js.nc.PublishMsg(m); err != nil {
+	if err := js.nc.publish(m.Subject, reply, hdr, m.Data); err != nil {
 		js.clearPAF(id)
 		return nil, err
 	}
@@ -1039,7 +1143,7 @@ func RetryAttempts(num int) PubOpt {
 func StallWait(ttl time.Duration) PubOpt {
 	return pubOptFn(func(opts *pubOpts) error {
 		if ttl <= 0 {
-			return fmt.Errorf("nats: stall wait should be more than 0")
+			return errors.New("nats: stall wait should be more than 0")
 		}
 		opts.stallWait = ttl
 		return nil
@@ -1397,11 +1501,11 @@ func processConsInfo(info *ConsumerInfo, userCfg *ConsumerConfig, isPullMode boo
 		// Prevent an user from attempting to create a queue subscription on
 		// a JS consumer that was not created with a deliver group.
 		if queue != _EMPTY_ {
-			return _EMPTY_, fmt.Errorf("cannot create a queue subscription for a consumer without a deliver group")
+			return _EMPTY_, errors.New("cannot create a queue subscription for a consumer without a deliver group")
 		} else if info.PushBound {
 			// Need to reject a non queue subscription to a non queue consumer
 			// if the consumer is already bound.
-			return _EMPTY_, fmt.Errorf("consumer is already bound to a subscription")
+			return _EMPTY_, errors.New("consumer is already bound to a subscription")
 		}
 	} else {
 		// If the JS consumer has a deliver group, we need to fail a non queue
@@ -1503,7 +1607,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 
 	// If no stream name is specified, the subject cannot be empty.
 	if subj == _EMPTY_ && o.stream == _EMPTY_ {
-		return nil, fmt.Errorf("nats: subject required")
+		return nil, errors.New("nats: subject required")
 	}
 
 	// Note that these may change based on the consumer info response we may get.
@@ -1525,7 +1629,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		// would subscribe to and server would send on.
 		if o.cfg.Heartbeat > 0 || o.cfg.FlowControl {
 			// Not making this a public ErrXXX in case we allow in the future.
-			return nil, fmt.Errorf("nats: queue subscription doesn't support idle heartbeat nor flow control")
+			return nil, errors.New("nats: queue subscription doesn't support idle heartbeat nor flow control")
 		}
 
 		// If this is a queue subscription and no consumer nor durable name was specified,
@@ -1563,31 +1667,31 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 	if o.ordered {
 		// Make sure we are not durable.
 		if isDurable {
-			return nil, fmt.Errorf("nats: durable can not be set for an ordered consumer")
+			return nil, errors.New("nats: durable can not be set for an ordered consumer")
 		}
 		// Check ack policy.
 		if o.cfg.AckPolicy != ackPolicyNotSet {
-			return nil, fmt.Errorf("nats: ack policy can not be set for an ordered consumer")
+			return nil, errors.New("nats: ack policy can not be set for an ordered consumer")
 		}
 		// Check max deliver.
 		if o.cfg.MaxDeliver != 1 && o.cfg.MaxDeliver != 0 {
-			return nil, fmt.Errorf("nats: max deliver can not be set for an ordered consumer")
+			return nil, errors.New("nats: max deliver can not be set for an ordered consumer")
 		}
 		// No deliver subject, we pick our own.
 		if o.cfg.DeliverSubject != _EMPTY_ {
-			return nil, fmt.Errorf("nats: deliver subject can not be set for an ordered consumer")
+			return nil, errors.New("nats: deliver subject can not be set for an ordered consumer")
 		}
 		// Queue groups not allowed.
 		if queue != _EMPTY_ {
-			return nil, fmt.Errorf("nats: queues not be set for an ordered consumer")
+			return nil, errors.New("nats: queues not be set for an ordered consumer")
 		}
 		// Check for bound consumers.
 		if consumer != _EMPTY_ {
-			return nil, fmt.Errorf("nats: can not bind existing consumer for an ordered consumer")
+			return nil, errors.New("nats: can not bind existing consumer for an ordered consumer")
 		}
 		// Check for pull mode.
 		if isPullMode {
-			return nil, fmt.Errorf("nats: can not use pull mode for an ordered consumer")
+			return nil, errors.New("nats: can not use pull mode for an ordered consumer")
 		}
 		// Setup how we need it to be here.
 		o.cfg.FlowControl = true
@@ -2321,7 +2425,7 @@ func Description(description string) SubOpt {
 func Durable(consumer string) SubOpt {
 	return subOptFn(func(opts *subOpts) error {
 		if opts.cfg.Durable != _EMPTY_ {
-			return fmt.Errorf("nats: option Durable set more than once")
+			return errors.New("nats: option Durable set more than once")
 		}
 		if opts.consumer != _EMPTY_ && opts.consumer != consumer {
 			return fmt.Errorf("nats: duplicate consumer names (%s and %s)", opts.consumer, consumer)
@@ -2899,10 +3003,11 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 			}
 
 			// Make our request expiration a bit shorter than the current timeout.
-			expires := ttl
-			if ttl >= 20*time.Millisecond {
-				expires = ttl - 10*time.Millisecond
+			expiresDiff := time.Duration(float64(ttl) * 0.1)
+			if expiresDiff > 5*time.Second {
+				expiresDiff = 5 * time.Second
 			}
+			expires := ttl - expiresDiff
 
 			nr.Batch = batch - len(msgs)
 			nr.Expires = expires
@@ -3166,10 +3271,11 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 	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
+	expiresDiff := time.Duration(float64(ttl) * 0.1)
+	if expiresDiff > 5*time.Second {
+		expiresDiff = 5 * time.Second
 	}
+	expires := ttl - expiresDiff
 
 	requestBatch := batch - len(result.msgs)
 	req := nextRequest{
@@ -3844,7 +3950,7 @@ func (alg StoreCompression) MarshalJSON() ([]byte, error) {
 	case NoCompression:
 		str = "none"
 	default:
-		return nil, fmt.Errorf("unknown compression algorithm")
+		return nil, errors.New("unknown compression algorithm")
 	}
 	return json.Marshal(str)
 }
@@ -3860,7 +3966,7 @@ func (alg *StoreCompression) UnmarshalJSON(b []byte) error {
 	case "none":
 		*alg = NoCompression
 	default:
-		return fmt.Errorf("unknown compression algorithm")
+		return errors.New("unknown compression algorithm")
 	}
 	return nil
 }
diff --git a/jserrors.go b/jserrors.go
index f028594..1c22d81 100644
--- a/jserrors.go
+++ b/jserrors.go
@@ -151,7 +151,10 @@ var (
 	// ErrSubscriptionClosed is returned when attempting to send pull request to a closed subscription
 	ErrSubscriptionClosed JetStreamError = &jsError{message: "subscription closed"}
 
-	// DEPRECATED: ErrInvalidDurableName is no longer returned and will be removed in future releases.
+	// ErrJetStreamPublisherClosed is returned for each unfinished ack future when JetStream.Cleanup is called.
+	ErrJetStreamPublisherClosed JetStreamError = &jsError{message: "jetstream context closed"}
+
+	// Deprecated: ErrInvalidDurableName is no longer returned and will be removed in future releases.
 	// Use ErrInvalidConsumerName instead.
 	ErrInvalidDurableName = errors.New("nats: invalid durable name")
 )
diff --git a/jsm.go b/jsm.go
index 9eb5d4b..2ae19c7 100644
--- a/jsm.go
+++ b/jsm.go
@@ -41,7 +41,7 @@ type JetStreamManager interface {
 	PurgeStream(name string, opts ...JSOpt) error
 
 	// StreamsInfo can be used to retrieve a list of StreamInfo objects.
-	// DEPRECATED: Use Streams() instead.
+	// Deprecated: Use Streams() instead.
 	StreamsInfo(opts ...JSOpt) <-chan *StreamInfo
 
 	// Streams can be used to retrieve a list of StreamInfo objects.
@@ -86,7 +86,7 @@ type JetStreamManager interface {
 	ConsumerInfo(stream, name string, opts ...JSOpt) (*ConsumerInfo, error)
 
 	// ConsumersInfo is used to retrieve a list of ConsumerInfo objects.
-	// DEPRECATED: Use Consumers() instead.
+	// Deprecated: Use Consumers() instead.
 	ConsumersInfo(stream string, opts ...JSOpt) <-chan *ConsumerInfo
 
 	// Consumers is used to retrieve a list of ConsumerInfo objects.
@@ -240,7 +240,7 @@ type StreamConfig struct {
 	// v2.10.0 or later.
 	Metadata map[string]string `json:"metadata,omitempty"`
 
-	// Template identifies the template that manages the Stream. DEPRECATED:
+	// Template identifies the template that manages the Stream. Deprecated:
 	// This feature is no longer supported.
 	Template string `json:"template_owner,omitempty"`
 }
@@ -747,7 +747,7 @@ func (jsc *js) Consumers(stream string, opts ...JSOpt) <-chan *ConsumerInfo {
 }
 
 // ConsumersInfo is used to retrieve a list of ConsumerInfo objects.
-// DEPRECATED: Use Consumers() instead.
+// Deprecated: Use Consumers() instead.
 func (jsc *js) ConsumersInfo(stream string, opts ...JSOpt) <-chan *ConsumerInfo {
 	return jsc.Consumers(stream, opts...)
 }
@@ -1330,11 +1330,11 @@ func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error
 	// Check for headers that give us the required information to
 	// reconstruct the message.
 	if len(r.Header) == 0 {
-		return nil, fmt.Errorf("nats: response should have headers")
+		return nil, errors.New("nats: response should have headers")
 	}
 	stream := r.Header.Get(JSStream)
 	if stream == _EMPTY_ {
-		return nil, fmt.Errorf("nats: missing stream header")
+		return nil, errors.New("nats: missing stream header")
 	}
 
 	// Mirrors can now answer direct gets, so removing check for name equality.
@@ -1342,7 +1342,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error
 
 	seqStr := r.Header.Get(JSSequence)
 	if seqStr == _EMPTY_ {
-		return nil, fmt.Errorf("nats: missing sequence header")
+		return nil, errors.New("nats: missing sequence header")
 	}
 	seq, err := strconv.ParseUint(seqStr, 10, 64)
 	if err != nil {
@@ -1350,7 +1350,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error
 	}
 	timeStr := r.Header.Get(JSTimeStamp)
 	if timeStr == _EMPTY_ {
-		return nil, fmt.Errorf("nats: missing timestamp header")
+		return nil, errors.New("nats: missing timestamp header")
 	}
 	// Temporary code: the server in main branch is sending with format
 	// "2006-01-02 15:04:05.999999999 +0000 UTC", but will be changed
@@ -1365,7 +1365,7 @@ func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error
 	}
 	subj := r.Header.Get(JSSubject)
 	if subj == _EMPTY_ {
-		return nil, fmt.Errorf("nats: missing subject header")
+		return nil, errors.New("nats: missing subject header")
 	}
 	return &RawStreamMsg{
 		Subject:  subj,
@@ -1617,7 +1617,7 @@ func (jsc *js) Streams(opts ...JSOpt) <-chan *StreamInfo {
 }
 
 // StreamsInfo can be used to retrieve a list of StreamInfo objects.
-// DEPRECATED: Use Streams() instead.
+// Deprecated: Use Streams() instead.
 func (jsc *js) StreamsInfo(opts ...JSOpt) <-chan *StreamInfo {
 	return jsc.Streams(opts...)
 }
diff --git a/kv.go b/kv.go
index d9f40fd..4e7a3fd 100644
--- a/kv.go
+++ b/kv.go
@@ -65,7 +65,7 @@ type KeyValue interface {
 	// WatchAll will invoke the callback for all updates.
 	WatchAll(opts ...WatchOpt) (KeyWatcher, error)
 	// Keys will return all keys.
-	// DEPRECATED: Use ListKeys instead to avoid memory issues.
+	// Deprecated: Use ListKeys instead to avoid memory issues.
 	Keys(opts ...WatchOpt) ([]string, error)
 	// ListKeys will return all keys in a channel.
 	ListKeys(opts ...WatchOpt) (KeyLister, error)
diff --git a/micro/README.md b/micro/README.md
index e4ffe09..99949eb 100644
--- a/micro/README.md
+++ b/micro/README.md
@@ -206,35 +206,69 @@ Service IDs can be discovered by:
 ```sh
 nats req '$SRV.PING.EchoService' '' --replies=3
 
-8:59:41 Sending request on "$SRV.PING.EchoService"
-18:59:41 Received with rtt 688.042µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxqC","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
+13:03:04 Sending request on "$SRV.PING.EchoService"
+13:03:04 Received with rtt 1.302208ms
+{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4K7","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
 
-18:59:41 Received with rtt 704.167µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxvO","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
+13:03:04 Received with rtt 1.317ms
+{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4Kt","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
 
-18:59:41 Received with rtt 707.875µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdy0a","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
+13:03:04 Received with rtt 1.320291ms
+{"name":"EchoService","id":"x3Yuiq7g7MoxhXdxk7i4Lf","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.ping_response"}
 ```
 
 A specific service instance info can be retrieved:
 
 ```sh
-nats req '$SRV.INFO.EchoService.tNoopzL5Sp1M4qJZdhdxqC' ''
-
-19:40:06 Sending request on "$SRV.INFO.EchoService.tNoopzL5Sp1M4qJZdhdxqC"
-19:40:06 Received with rtt 282.375µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxqC","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.info_response","description":"","subjects":["svc.echo"]}
+nats req '$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7' '' | jq
+
+13:04:19 Sending request on "$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7"
+13:04:19 Received with rtt 318.875µs
+{
+  "name": "EchoService",
+  "id": "x3Yuiq7g7MoxhXdxk7i4K7",
+  "version": "1.0.0",
+  "metadata": {},
+  "type": "io.nats.micro.v1.info_response",
+  "description": "",
+  "endpoints": [
+    {
+      "name": "default",
+      "subject": "svc.echo",
+      "queue_group": "q",
+      "metadata": null
+    }
+  ]
+}
 ```
 
 To get statistics for this service:
 
 ```sh
-nats req '$SRV.STATS.EchoService.tNoopzL5Sp1M4qJZdhdxqC' ''
-
-19:40:47 Sending request on "$SRV.STATS.EchoService.tNoopzL5Sp1M4qJZdhdxqC"
-19:40:47 Received with rtt 421.666µs
-{"name":"EchoService","id":"tNoopzL5Sp1M4qJZdhdxqC","version":"1.0.0","metadata":{},"type":"io.nats.micro.v1.stats_response","started":"2023-05-22T16:59:39.938514Z","endpoints":[{"name":"default","subject":"svc.echo","metadata":null,"num_requests":0,"num_errors":0,"last_error":"","processing_time":0,"average_processing_time":0}]}
+nats req '$SRV.STATS.EchoService.x3Yuiq7g7MoxhXdxk7i4K7' '' | jq
+
+13:04:46 Sending request on "$SRV.STATS.EchoService.x3Yuiq7g7MoxhXdxk7i4K7"
+13:04:46 Received with rtt 678.25µs
+{
+  "name": "EchoService",
+  "id": "x3Yuiq7g7MoxhXdxk7i4K7",
+  "version": "1.0.0",
+  "metadata": {},
+  "type": "io.nats.micro.v1.stats_response",
+  "started": "2024-09-24T11:02:55.564771Z",
+  "endpoints": [
+    {
+      "name": "default",
+      "subject": "svc.echo",
+      "queue_group": "q",
+      "num_requests": 0,
+      "num_errors": 0,
+      "last_error": "",
+      "processing_time": 0,
+      "average_processing_time": 0
+    }
+  ]
+}
 ```
 
 ## Examples
diff --git a/nats.go b/nats.go
index d019cee..83638a5 100644
--- a/nats.go
+++ b/nats.go
@@ -47,7 +47,7 @@ import (
 
 // Default Constants
 const (
-	Version                   = "1.36.0"
+	Version                   = "1.37.0"
 	DefaultURL                = "nats://127.0.0.1:4222"
 	DefaultPort               = 4222
 	DefaultMaxReconnect       = 60
@@ -131,6 +131,7 @@ var (
 	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")
+	ErrUserInfoAlreadySet          = errors.New("nats: cannot set user info callback and user/pass")
 	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")
@@ -160,7 +161,7 @@ func GetDefaultOptions() Options {
 	}
 }
 
-// DEPRECATED: Use GetDefaultOptions() instead.
+// Deprecated: Use GetDefaultOptions() instead.
 // DefaultOptions is not safe for use by multiple clients.
 // For details see #308.
 var DefaultOptions = GetDefaultOptions()
@@ -230,6 +231,9 @@ type SignatureHandler func([]byte) ([]byte, error)
 // AuthTokenHandler is used to generate a new token.
 type AuthTokenHandler func() string
 
+// UserInfoCB is used to pass the username and password when establishing connection.
+type UserInfoCB func() (string, string)
+
 // ReconnectDelayHandler is used to get from the user the desired
 // delay the library should pause before attempting to reconnect
 // again. Note that this is invoked after the library tried the
@@ -386,7 +390,7 @@ type Options struct {
 	// DisconnectedCB sets the disconnected handler that is called
 	// whenever the connection is disconnected.
 	// Will not be called if DisconnectedErrCB is set
-	// DEPRECATED. Use DisconnectedErrCB which passes error that caused
+	// Deprecated. Use DisconnectedErrCB which passes error that caused
 	// the disconnect event.
 	DisconnectedCB ConnHandler
 
@@ -443,6 +447,9 @@ type Options struct {
 	// Password sets the password to be used when connecting to a server.
 	Password string
 
+	// UserInfo sets the callback handler that will fetch the username and password.
+	UserInfo UserInfoCB
+
 	// Token sets the token to be used when connecting to a server.
 	Token string
 
@@ -450,7 +457,7 @@ type Options struct {
 	TokenHandler AuthTokenHandler
 
 	// Dialer allows a custom net.Dialer when forming connections.
-	// DEPRECATED: should use CustomDialer instead.
+	// Deprecated: should use CustomDialer instead.
 	Dialer *net.Dialer
 
 	// CustomDialer allows to specify a custom dialer (not necessarily
@@ -1108,7 +1115,7 @@ func DisconnectErrHandler(cb ConnErrHandler) Option {
 }
 
 // DisconnectHandler is an Option to set the disconnected handler.
-// DEPRECATED: Use DisconnectErrHandler.
+// Deprecated: Use DisconnectErrHandler.
 func DisconnectHandler(cb ConnHandler) Option {
 	return func(o *Options) error {
 		o.DisconnectedCB = cb
@@ -1166,6 +1173,13 @@ func UserInfo(user, password string) Option {
 	}
 }
 
+func UserInfoHandler(cb UserInfoCB) Option {
+	return func(o *Options) error {
+		o.UserInfo = cb
+		return nil
+	}
+}
+
 // Token is an Option to set the token to use
 // when a token is not included directly in the URLs
 // and when a token handler is not provided.
@@ -1280,7 +1294,7 @@ func SyncQueueLen(max int) Option {
 
 // Dialer is an Option to set the dialer which will be used when
 // attempting to establish a connection.
-// DEPRECATED: Should use CustomDialer instead.
+// Deprecated: Should use CustomDialer instead.
 func Dialer(dialer *net.Dialer) Option {
 	return func(o *Options) error {
 		o.Dialer = dialer
@@ -1359,7 +1373,7 @@ func ProxyPath(path string) Option {
 func CustomInboxPrefix(p string) Option {
 	return func(o *Options) error {
 		if p == "" || strings.Contains(p, ">") || strings.Contains(p, "*") || strings.HasSuffix(p, ".") {
-			return fmt.Errorf("nats: invalid custom prefix")
+			return errors.New("nats: invalid custom prefix")
 		}
 		o.InboxPrefix = p
 		return nil
@@ -1397,7 +1411,7 @@ func TLSHandshakeFirst() Option {
 // Handler processing
 
 // SetDisconnectHandler will set the disconnect event handler.
-// DEPRECATED: Use SetDisconnectErrHandler
+// Deprecated: Use SetDisconnectErrHandler
 func (nc *Conn) SetDisconnectHandler(dcb ConnHandler) {
 	if nc == nil {
 		return
@@ -1513,7 +1527,7 @@ func processUrlString(url string) []string {
 	urls := strings.Split(url, ",")
 	var j int
 	for _, s := range urls {
-		u := strings.TrimSpace(s)
+		u := strings.TrimSuffix(strings.TrimSpace(s), "/")
 		if len(u) > 0 {
 			urls[j] = u
 			j++
@@ -1814,7 +1828,7 @@ func (nc *Conn) addURLToPool(sURL string, implicit, saveTLSName bool) error {
 	if len(nc.srvPool) == 0 {
 		nc.ws = isWS
 	} else if isWS && !nc.ws || !isWS && nc.ws {
-		return fmt.Errorf("mixing of websocket and non websocket URLs is not allowed")
+		return errors.New("mixing of websocket and non websocket URLs is not allowed")
 	}
 
 	var tlsName string
@@ -2563,6 +2577,13 @@ func (nc *Conn) connectProto() (string, error) {
 		pass = o.Password
 		token = o.Token
 		nkey = o.Nkey
+
+		if nc.Opts.UserInfo != nil {
+			if user != _EMPTY_ || pass != _EMPTY_ {
+				return _EMPTY_, ErrUserInfoAlreadySet
+			}
+			user, pass = nc.Opts.UserInfo()
+		}
 	}
 
 	// Look for user jwt.
@@ -4902,7 +4923,8 @@ func (s *Subscription) processNextMsgDelivered(msg *Msg) error {
 }
 
 // Queued returns the number of queued messages in the client for this subscription.
-// DEPRECATED: Use Pending()
+//
+// Deprecated: Use Pending()
 func (s *Subscription) QueuedMsgs() (int, error) {
 	m, _, err := s.Pending()
 	return int(m), err
@@ -5791,7 +5813,7 @@ func NkeyOptionFromSeed(seedFile string) (Option, error) {
 		return nil, err
 	}
 	if !nkeys.IsValidPublicUserKey(pub) {
-		return nil, fmt.Errorf("nats: Not a valid nkey user seed")
+		return nil, errors.New("nats: Not a valid nkey user seed")
 	}
 	sigCB := func(nonce []byte) ([]byte, error) {
 		return sigHandler(nonce, seedFile)
diff --git a/netchan.go b/netchan.go
index 6b13690..3722d9f 100644
--- a/netchan.go
+++ b/netchan.go
@@ -23,6 +23,8 @@ import (
 // Data will be encoded and decoded via the EncodedConn and its associated encoders.
 
 // BindSendChan binds a channel for send operations to NATS.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) BindSendChan(subject string, channel any) error {
 	chVal := reflect.ValueOf(channel)
 	if chVal.Kind() != reflect.Chan {
@@ -61,11 +63,15 @@ func chPublish(c *EncodedConn, chVal reflect.Value, subject string) {
 }
 
 // BindRecvChan binds a channel for receive operations from NATS.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) BindRecvChan(subject string, channel any) (*Subscription, error) {
 	return c.bindRecvChan(subject, _EMPTY_, channel)
 }
 
 // BindRecvQueueChan binds a channel for queue-based receive operations from NATS.
+//
+// Deprecated: Encoded connections are no longer supported.
 func (c *EncodedConn) BindRecvQueueChan(subject, queue string, channel any) (*Subscription, error) {
 	return c.bindRecvChan(subject, queue, channel)
 }
diff --git a/scripts/cov.sh b/scripts/cov.sh
index 80828cb..fa0fdc1 100755
--- a/scripts/cov.sh
+++ b/scripts/cov.sh
@@ -5,14 +5,15 @@ rm -rf ./cov
 mkdir cov
 go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/nats.out . -tags=skip_no_race_tests
 go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/test.out -coverpkg=github.com/nats-io/nats.go ./test -tags=skip_no_race_tests,internal_testing
-go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/jetstream.out -coverpkg=github.com/nats-io/nats.go/jetstream ./jetstream/test -tags=skip_no_race_tests
+go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/jetstream.out -coverpkg=github.com/nats-io/nats.go/jetstream ./jetstream/...
+go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/service.out -coverpkg=github.com/nats-io/nats.go/micro ./micro/...
 go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/builtin.out -coverpkg=github.com/nats-io/nats.go/encoders/builtin ./test -run EncBuiltin -tags=skip_no_race_tests
 go test -modfile=go_test.mod --failfast -vet=off -v -covermode=atomic -coverprofile=./cov/protobuf.out -coverpkg=github.com/nats-io/nats.go/encoders/protobuf ./test -run EncProto -tags=skip_no_race_tests
 gocovmerge ./cov/*.out > acc.out
 rm -rf ./cov
 
 # Without argument, launch browser results. We are going to push to coveralls only
-# from Travis.yml and after success of the build (and result of pushing will not affect
+# from ci.yml and after success of the build (and result of pushing will not affect
 # build result).
 if [[ $1 == "" ]]; then
     go tool cover -html=acc.out
diff --git a/ws.go b/ws.go
index 2c2d421..fbc5688 100644
--- a/ws.go
+++ b/ws.go
@@ -237,8 +237,8 @@ func (r *websocketReader) Read(p []byte) (int, error) {
 		case wsPingMessage, wsPongMessage, wsCloseMessage:
 			if rem > wsMaxControlPayloadSize {
 				return 0, fmt.Errorf(
-					fmt.Sprintf("control frame length bigger than maximum allowed of %v bytes",
-						wsMaxControlPayloadSize))
+					"control frame length bigger than maximum allowed of %v bytes",
+					wsMaxControlPayloadSize)
 			}
 			if compressed {
 				return 0, errors.New("control frame should not be compressed")
@@ -622,7 +622,7 @@ func (nc *Conn) wsInitHandshake(u *url.URL) error {
 			!strings.EqualFold(resp.Header.Get("Connection"), "upgrade") ||
 			resp.Header.Get("Sec-Websocket-Accept") != wsAcceptKey(wsKey)) {
 
-		err = fmt.Errorf("invalid websocket connection")
+		err = errors.New("invalid websocket connection")
 	}
 	// Check compression extension...
 	if err == nil && compress {
@@ -634,7 +634,7 @@ func (nc *Conn) wsInitHandshake(u *url.URL) error {
 		if !srvCompress {
 			compress = false
 		} else if !noCtxTakeover {
-			err = fmt.Errorf("compression negotiation error")
+			err = errors.New("compression negotiation error")
 		}
 	}
 	if resp != nil {


Test patch:
diff --git a/.github/workflows/latest-server.yaml b/.github/workflows/latest-server.yaml
new file mode 100644
index 0000000..c44523b
--- /dev/null
+++ b/.github/workflows/latest-server.yaml
@@ -0,0 +1,27 @@
+name: Test nats-server@main
+on:
+  schedule:
+    - cron: "30 8 * * *"
+
+jobs:
+    test:
+        runs-on: ubuntu-latest
+        steps:
+            - name: Checkout code
+              uses: actions/checkout@v4
+      
+            - name: Setup Go
+              uses: actions/setup-go@v5
+              with:
+                go-version: 'stable'
+                
+            - name: Get latest server
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go get -modfile go_test.mod github.com/nats-io/nats-server/v2@main
+
+            - name: Test
+              shell: bash --noprofile --norc -x -eo pipefail {0}
+              run: |
+                go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
+                go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing
\ No newline at end of file
diff --git a/example_test.go b/example_test.go
index 3ad3673..782adc4 100644
--- a/example_test.go
+++ b/example_test.go
@@ -227,106 +227,6 @@ func ExampleConn_Close() {
 	nc.Close()
 }
 
-// Shows how to wrap a Conn into an EncodedConn
-func ExampleNewEncodedConn() {
-	nc, _ := nats.Connect(nats.DefaultURL)
-	c, _ := nats.NewEncodedConn(nc, "json")
-	c.Close()
-}
-
-// EncodedConn can publish virtually anything just
-// by passing it in. The encoder will be used to properly
-// encode the raw Go type
-func ExampleEncodedConn_Publish() {
-	nc, _ := nats.Connect(nats.DefaultURL)
-	c, _ := nats.NewEncodedConn(nc, "json")
-	defer c.Close()
-
-	type person struct {
-		Name    string
-		Address string
-		Age     int
-	}
-
-	me := &person{Name: "derek", Age: 22, Address: "85 Second St"}
-	c.Publish("hello", me)
-}
-
-// EncodedConn's subscribers will automatically decode the
-// wire data into the requested Go type using the Decode()
-// method of the registered Encoder. The callback signature
-// can also vary to include additional data, such as subject
-// and reply subjects.
-func ExampleEncodedConn_Subscribe() {
-	nc, _ := nats.Connect(nats.DefaultURL)
-	c, _ := nats.NewEncodedConn(nc, "json")
-	defer c.Close()
-
-	type person struct {
-		Name    string
-		Address string
-		Age     int
-	}
-
-	c.Subscribe("hello", func(p *person) {
-		fmt.Printf("Received a person! %+v\n", p)
-	})
-
-	c.Subscribe("hello", func(subj, reply string, p *person) {
-		fmt.Printf("Received a person on subject %s! %+v\n", subj, p)
-	})
-
-	me := &person{Name: "derek", Age: 22, Address: "85 Second St"}
-	c.Publish("hello", me)
-}
-
-// BindSendChan() allows binding of a Go channel to a nats
-// subject for publish operations. The Encoder attached to the
-// EncodedConn will be used for marshaling.
-func ExampleEncodedConn_BindSendChan() {
-	nc, _ := nats.Connect(nats.DefaultURL)
-	c, _ := nats.NewEncodedConn(nc, "json")
-	defer c.Close()
-
-	type person struct {
-		Name    string
-		Address string
-		Age     int
-	}
-
-	ch := make(chan *person)
-	c.BindSendChan("hello", ch)
-
-	me := &person{Name: "derek", Age: 22, Address: "85 Second St"}
-	ch <- me
-}
-
-// BindRecvChan() allows binding of a Go channel to a nats
-// subject for subscribe operations. The Encoder attached to the
-// EncodedConn will be used for un-marshaling.
-func ExampleEncodedConn_BindRecvChan() {
-	nc, _ := nats.Connect(nats.DefaultURL)
-	c, _ := nats.NewEncodedConn(nc, "json")
-	defer c.Close()
-
-	type person struct {
-		Name    string
-		Address string
-		Age     int
-	}
-
-	ch := make(chan *person)
-	c.BindRecvChan("hello", ch)
-
-	me := &person{Name: "derek", Age: 22, Address: "85 Second St"}
-	c.Publish("hello", me)
-
-	// Receive the publish directly on a channel
-	who := <-ch
-
-	fmt.Printf("%v says hello!\n", who)
-}
-
 func ExampleJetStream() {
 	nc, err := nats.Connect("localhost")
 	if err != nil {
diff --git a/go_test.mod b/go_test.mod
index f5b731d..20e1ab7 100644
--- a/go_test.mod
+++ b/go_test.mod
@@ -1,23 +1,25 @@
 module github.com/nats-io/nats.go
 
-go 1.19
+go 1.21
+
+toolchain go1.22.5
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/klauspost/compress v1.17.8
+	github.com/klauspost/compress v1.17.9
 	github.com/nats-io/jwt v1.2.2
-	github.com/nats-io/nats-server/v2 v2.10.16
+	github.com/nats-io/nats-server/v2 v2.10.17
 	github.com/nats-io/nkeys v0.4.7
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/goleak v1.3.0
-	golang.org/x/text v0.15.0
+	golang.org/x/text v0.16.0
 	google.golang.org/protobuf v1.23.0
 )
 
 require (
 	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/nats-io/jwt/v2 v2.5.7 // indirect
-	golang.org/x/crypto v0.23.0 // indirect
-	golang.org/x/sys v0.20.0 // indirect
+	golang.org/x/crypto v0.24.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 )
diff --git a/go_test.sum b/go_test.sum
index f89d755..df0ef6d 100644
--- a/go_test.sum
+++ b/go_test.sum
@@ -1,4 +1,5 @@
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
 github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
 github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
@@ -10,38 +11,40 @@ 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.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=
-github.com/klauspost/compress v1.17.8/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
+github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 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 v1.2.2 h1:w3GMTO969dFg+UOKTmmyuu7IGdusK+7Ytlt//OYH/uU=
 github.com/nats-io/jwt v1.2.2/go.mod h1:/xX356yQA6LuXI9xWW7mZNpxgF2mBmGecH+Fj34sP5Q=
 github.com/nats-io/jwt/v2 v2.5.7 h1:j5lH1fUXCnJnY8SsQeB/a/z9Azgu2bYIDvtPVNdxe2c=
 github.com/nats-io/jwt/v2 v2.5.7/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
-github.com/nats-io/nats-server/v2 v2.10.16 h1:2jXaiydp5oB/nAx/Ytf9fdCi9QN6ItIc9eehX8kwVV0=
-github.com/nats-io/nats-server/v2 v2.10.16/go.mod h1:Pksi38H2+6xLe1vQx0/EA4bzetM0NqyIHcIbmgXSkIU=
+github.com/nats-io/nats-server/v2 v2.10.17 h1:PTVObNBD3TZSNUDgzFb1qQsQX4mOgFmOuG9vhT+KBUY=
+github.com/nats-io/nats-server/v2 v2.10.17/go.mod h1:5OUyc4zg42s/p2i92zbbqXvUNsbF0ivdTLKshVMn2YQ=
 github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
 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/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 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.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
-golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
+golang.org/x/crypto v0.24.0 h1:mnl8DM0o513X8fdIkmyFE/5hTYxbwYOjDS/+rK6qpRI=
+golang.org/x/crypto v0.24.0/go.mod h1:Z1PMYSOR5nyMcyAVAIQSKCDwalqy85Aqn1x3Ws4L5DM=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
-golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk=
-golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
+golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
 golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
@@ -54,3 +57,4 @@ google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzi
 google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/internal/syncx/map_test.go b/internal/syncx/map_test.go
new file mode 100644
index 0000000..df34b2f
--- /dev/null
+++ b/internal/syncx/map_test.go
@@ -0,0 +1,152 @@
+// Copyright 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
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package syncx
+
+import (
+	"testing"
+)
+
+func TestMapLoad(t *testing.T) {
+	var m Map[int, string]
+	m.Store(1, "one")
+
+	v, ok := m.Load(1)
+	if !ok || v != "one" {
+		t.Errorf("Load(1) = %v, %v; want 'one', true", v, ok)
+	}
+
+	v, ok = m.Load(2)
+	if ok || v != "" {
+		t.Errorf("Load(2) = %v, %v; want '', false", v, ok)
+	}
+}
+
+func TestMapStore(t *testing.T) {
+	var m Map[int, string]
+	m.Store(1, "one")
+
+	v, ok := m.Load(1)
+	if !ok || v != "one" {
+		t.Errorf("Load(1) after Store(1, 'one') = %v, %v; want 'one', true", v, ok)
+	}
+}
+
+func TestMapDelete(t *testing.T) {
+	var m Map[int, string]
+	m.Store(1, "one")
+	m.Delete(1)
+
+	v, ok := m.Load(1)
+	if ok || v != "" {
+		t.Errorf("Load(1) after Delete(1) = %v, %v; want '', false", v, ok)
+	}
+}
+
+func TestMapRange(t *testing.T) {
+	var m Map[int, string]
+	m.Store(1, "one")
+	m.Store(2, "two")
+
+	var keys []int
+	var values []string
+	m.Range(func(key int, value string) bool {
+		keys = append(keys, key)
+		values = append(values, value)
+		return true
+	})
+
+	if len(keys) != 2 || len(values) != 2 {
+		t.Errorf("Range() keys = %v, values = %v; want 2 keys and 2 values", keys, values)
+	}
+}
+
+func TestMapLoadOrStore(t *testing.T) {
+	var m Map[int, string]
+
+	v, loaded := m.LoadOrStore(1, "one")
+	if loaded || v != "one" {
+		t.Errorf("LoadOrStore(1, 'one') = %v, %v; want 'one', false", v, loaded)
+	}
+
+	v, loaded = m.LoadOrStore(1, "uno")
+	if !loaded || v != "one" {
+		t.Errorf("LoadOrStore(1, 'uno') = %v, %v; want 'one', true", v, loaded)
+	}
+}
+
+func TestMapLoadAndDelete(t *testing.T) {
+	var m Map[int, string]
+	m.Store(1, "one")
+
+	v, ok := m.LoadAndDelete(1)
+	if !ok || v != "one" {
+		t.Errorf("LoadAndDelete(1) = %v, %v; want 'one', true", v, ok)
+	}
+
+	v, ok = m.Load(1)
+	if ok || v != "" {
+		t.Errorf("Load(1) after LoadAndDelete(1) = %v, %v; want '', false", v, ok)
+	}
+
+	// Test that LoadAndDelete on a missing key returns the zero value.
+	v, ok = m.LoadAndDelete(2)
+	if ok || v != "" {
+		t.Errorf("LoadAndDelete(2) = %v, %v; want '', false", v, ok)
+	}
+}
+
+func TestMapCompareAndSwap(t *testing.T) {
+	var m Map[int, string]
+	m.Store(1, "one")
+
+	ok := m.CompareAndSwap(1, "one", "uno")
+	if !ok {
+		t.Errorf("CompareAndSwap(1, 'one', 'uno') = false; want true")
+	}
+
+	v, _ := m.Load(1)
+	if v != "uno" {
+		t.Errorf("Load(1) after CompareAndSwap = %v; want 'uno'", v)
+	}
+}
+
+func TestMapCompareAndDelete(t *testing.T) {
+	var m Map[int, string]
+	m.Store(1, "one")
+
+	ok := m.CompareAndDelete(1, "one")
+	if !ok {
+		t.Errorf("CompareAndDelete(1, 'one') = false; want true")
+	}
+
+	v, _ := m.Load(1)
+	if v != "" {
+		t.Errorf("Load(1) after CompareAndDelete = %v; want ''", v)
+	}
+}
+
+func TestMapSwap(t *testing.T) {
+	var m Map[int, string]
+	m.Store(1, "one")
+
+	v, loaded := m.Swap(1, "uno")
+	if !loaded || v != "one" {
+		t.Errorf("Swap(1, 'uno') = %v, %v; want 'one', true", v, loaded)
+	}
+
+	v, _ = m.Load(1)
+	if v != "uno" {
+		t.Errorf("Load(1) after Swap = %v; want 'uno'", v)
+	}
+}
diff --git a/jetstream/jetstream_test.go b/jetstream/jetstream_test.go
index 878d361..9d429ad 100644
--- a/jetstream/jetstream_test.go
+++ b/jetstream/jetstream_test.go
@@ -276,9 +276,11 @@ func TestRetryWithBackoff(t *testing.T) {
 }
 
 func TestPullConsumer_checkPending(t *testing.T) {
+
 	tests := []struct {
 		name                string
 		givenSub            *pullSubscription
+		fetchInProgress     bool
 		shouldSend          bool
 		expectedPullRequest *pullRequest
 	}{
@@ -292,7 +294,6 @@ func TestPullConsumer_checkPending(t *testing.T) {
 					ThresholdMessages: 5,
 					MaxMessages:       10,
 				},
-				fetchInProgress: 0,
 			},
 			shouldSend: false,
 		},
@@ -307,7 +308,6 @@ func TestPullConsumer_checkPending(t *testing.T) {
 					ThresholdMessages: 5,
 					MaxMessages:       10,
 				},
-				fetchInProgress: 0,
 			},
 			shouldSend: true,
 			expectedPullRequest: &pullRequest{
@@ -325,9 +325,9 @@ func TestPullConsumer_checkPending(t *testing.T) {
 					ThresholdMessages: 5,
 					MaxMessages:       10,
 				},
-				fetchInProgress: 1,
 			},
-			shouldSend: false,
+			fetchInProgress: true,
+			shouldSend:      false,
 		},
 		{
 			name: "pending bytes below threshold, send pull request",
@@ -341,7 +341,6 @@ func TestPullConsumer_checkPending(t *testing.T) {
 					ThresholdBytes: 500,
 					MaxBytes:       1000,
 				},
-				fetchInProgress: 0,
 			},
 			shouldSend: true,
 			expectedPullRequest: &pullRequest{
@@ -359,7 +358,6 @@ func TestPullConsumer_checkPending(t *testing.T) {
 					ThresholdBytes: 500,
 					MaxBytes:       1000,
 				},
-				fetchInProgress: 0,
 			},
 			shouldSend: false,
 		},
@@ -373,9 +371,9 @@ func TestPullConsumer_checkPending(t *testing.T) {
 					ThresholdBytes: 500,
 					MaxBytes:       1000,
 				},
-				fetchInProgress: 1,
 			},
-			shouldSend: false,
+			fetchInProgress: true,
+			shouldSend:      false,
 		},
 		{
 			name: "StopAfter set, pending msgs below StopAfter, send pull request",
@@ -388,8 +386,7 @@ func TestPullConsumer_checkPending(t *testing.T) {
 					MaxMessages:       10,
 					StopAfter:         8,
 				},
-				fetchInProgress: 0,
-				delivered:       2,
+				delivered: 2,
 			},
 			shouldSend: true,
 			expectedPullRequest: &pullRequest{
@@ -408,8 +405,7 @@ func TestPullConsumer_checkPending(t *testing.T) {
 					MaxMessages:       10,
 					StopAfter:         6,
 				},
-				fetchInProgress: 0,
-				delivered:       0,
+				delivered: 0,
 			},
 			shouldSend: false,
 		},
@@ -419,6 +415,9 @@ func TestPullConsumer_checkPending(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			prChan := make(chan *pullRequest, 1)
 			test.givenSub.fetchNext = prChan
+			if test.fetchInProgress {
+				test.givenSub.fetchInProgress.Store(1)
+			}
 			errs := make(chan error, 1)
 			ok := make(chan struct{}, 1)
 			go func() {
@@ -431,13 +430,13 @@ func TestPullConsumer_checkPending(t *testing.T) {
 						}
 						ok <- struct{}{}
 					case <-time.After(1 * time.Second):
-						errs <- fmt.Errorf("Timeout")
+						errs <- errors.New("Timeout")
 						return
 					}
 				} else {
 					select {
 					case <-prChan:
-						errs <- fmt.Errorf("Unexpected pull request")
+						errs <- errors.New("Unexpected pull request")
 					case <-time.After(100 * time.Millisecond):
 						ok <- struct{}{}
 						return
diff --git a/jetstream/test/helper_test.go b/jetstream/test/helper_test.go
index a9dbae2..9b7c1b7 100644
--- a/jetstream/test/helper_test.go
+++ b/jetstream/test/helper_test.go
@@ -82,15 +82,6 @@ func NewConnection(t *testing.T, port int) *nats.Conn {
 	return nc
 }
 
-// NewEConn
-func NewEConn(t *testing.T) *nats.EncodedConn {
-	ec, err := nats.NewEncodedConn(NewDefaultConnection(t), nats.DEFAULT_ENCODER)
-	if err != nil {
-		t.Fatalf("Failed to create an encoded connection: %v\n", err)
-	}
-	return ec
-}
-
 ////////////////////////////////////////////////////////////////////////////////
 // Running nats server in separate Go routines
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/jetstream/test/jetstream_test.go b/jetstream/test/jetstream_test.go
index f5c9c8e..3b25309 100644
--- a/jetstream/test/jetstream_test.go
+++ b/jetstream/test/jetstream_test.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2023 The NATS Authors
+// Copyright 2022-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
@@ -1963,3 +1963,125 @@ func TestConsumerConfigMatches(t *testing.T) {
 		t.Fatalf("ConsumerConfig doesn't match")
 	}
 }
+
+func TestJetStreamCleanupPublisher(t *testing.T) {
+
+	t.Run("cleanup js publisher", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		// Create a stream
+		if _, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "TEST", Subjects: []string{"FOO"}}); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		numSubs := nc.NumSubscriptions()
+		if _, err := js.PublishAsync("FOO", []byte("hello")); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		select {
+		case <-js.PublishAsyncComplete():
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Did not receive completion signal")
+		}
+
+		if numSubs+1 != nc.NumSubscriptions() {
+			t.Fatalf("Expected an additional subscription after publish, got %d", nc.NumSubscriptions())
+		}
+
+		js.CleanupPublisher()
+
+		if numSubs != nc.NumSubscriptions() {
+			t.Fatalf("Expected subscriptions to be back to original count")
+		}
+	})
+
+	t.Run("cleanup js publisher, cancel pending acks", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, err := nats.Connect(s.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		cbErr := make(chan error, 10)
+		js, err := jetstream.New(nc, jetstream.WithPublishAsyncErrHandler(func(js jetstream.JetStream, m *nats.Msg, err error) {
+			cbErr <- err
+		}))
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		// Create a stream with NoAck so that we can test that we cancel ack futures.
+		if _, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "TEST", Subjects: []string{"FOO"}, NoAck: true}); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		numSubs := nc.NumSubscriptions()
+
+		var acks []jetstream.PubAckFuture
+		for i := 0; i < 10; i++ {
+			ack, err := js.PublishAsync("FOO", []byte("hello"))
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			acks = append(acks, ack)
+		}
+
+		asyncComplete := js.PublishAsyncComplete()
+		select {
+		case <-asyncComplete:
+			t.Fatalf("Should not complete, NoAck is set")
+		case <-time.After(200 * time.Millisecond):
+		}
+
+		if numSubs+1 != nc.NumSubscriptions() {
+			t.Fatalf("Expected an additional subscription after publish, got %d", nc.NumSubscriptions())
+		}
+
+		js.CleanupPublisher()
+
+		if numSubs != nc.NumSubscriptions() {
+			t.Fatalf("Expected subscriptions to be back to original count")
+		}
+
+		// check that PublishAsyncComplete channel is closed
+		select {
+		case <-asyncComplete:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Did not receive completion signal")
+		}
+
+		// check that all ack futures are canceled
+		for _, ack := range acks {
+			select {
+			case err := <-ack.Err():
+				if !errors.Is(err, jetstream.ErrJetStreamPublisherClosed) {
+					t.Fatalf("Expected JetStreamContextClosed error, got %v", err)
+				}
+			case <-ack.Ok():
+				t.Fatalf("Expected error on the ack future")
+			case <-time.After(200 * time.Millisecond):
+				t.Fatalf("Expected an error on the ack future")
+			}
+		}
+
+		// check that async error handler is called for each pending ack
+		for i := 0; i < 10; i++ {
+			select {
+			case err := <-cbErr:
+				if !errors.Is(err, jetstream.ErrJetStreamPublisherClosed) {
+					t.Fatalf("Expected JetStreamContextClosed error, got %v", err)
+				}
+			case <-time.After(200 * time.Millisecond):
+				t.Fatalf("Expected errors to be passed from the async handler")
+			}
+		}
+	})
+
+}
diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go
index 26a038c..50d2ffa 100644
--- a/jetstream/test/kv_test.go
+++ b/jetstream/test/kv_test.go
@@ -967,6 +967,7 @@ func TestKeyValueListKeys(t *testing.T) {
 
 func TestKeyValueCrossAccounts(t *testing.T) {
 	conf := createConfFile(t, []byte(`
+		listen: 127.0.0.1:-1
         jetstream: enabled
         accounts: {
            A: {
@@ -1262,7 +1263,7 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 		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")
+				return errors.New("Expected key to be gone")
 			}
 			if !errors.Is(err, jetstream.ErrKeyNotFound) {
 				return err
diff --git a/jetstream/test/ordered_test.go b/jetstream/test/ordered_test.go
index 01a67a8..6680955 100644
--- a/jetstream/test/ordered_test.go
+++ b/jetstream/test/ordered_test.go
@@ -539,6 +539,174 @@ func TestOrderedConsumerConsume(t *testing.T) {
 		cc.Drain()
 		wg.Wait()
 	})
+
+	t.Run("stop consume during reset", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		for i := 0; i < 10; i++ {
+			c, err := s.OrderedConsumer(context.Background(), jetstream.OrderedConsumerConfig{})
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			cc, err := c.Consume(func(msg jetstream.Msg) {
+				msg.Ack()
+			})
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if err := s.DeleteConsumer(context.Background(), c.CachedInfo().Name); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			cc.Stop()
+			time.Sleep(50 * time.Millisecond)
+		}
+	})
+
+	t.Run("wait for closed after drain", func(t *testing.T) {
+		for i := 0; i < 10; i++ {
+			t.Run(fmt.Sprintf("run %d", i), func(t *testing.T) {
+				srv := RunBasicJetStreamServer()
+				defer shutdownJSServerAndRemoveStorage(t, srv)
+				nc, err := nats.Connect(srv.ClientURL())
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				js, err := jetstream.New(nc)
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				defer nc.Close()
+				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+				defer cancel()
+				s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				msgs := make([]jetstream.Msg, 0)
+				lock := sync.Mutex{}
+				publishTestMsgs(t, js)
+				cc, err := c.Consume(func(msg jetstream.Msg) {
+					time.Sleep(50 * time.Millisecond)
+					msg.Ack()
+					lock.Lock()
+					msgs = append(msgs, msg)
+					lock.Unlock()
+				})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				closed := cc.Closed()
+				time.Sleep(100 * time.Millisecond)
+				if err := s.DeleteConsumer(context.Background(), c.CachedInfo().Name); err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				publishTestMsgs(t, js)
+
+				// wait for the consumer to be recreated before calling drain
+				for i := 0; i < 5; i++ {
+					_, err = c.Info(ctx)
+					if err != nil {
+						if errors.Is(err, jetstream.ErrConsumerNotFound) {
+							time.Sleep(100 * time.Millisecond)
+							continue
+						}
+						t.Fatalf("Unexpected error: %v", err)
+					}
+					break
+				}
+
+				cc.Drain()
+
+				select {
+				case <-closed:
+				case <-time.After(5 * time.Second):
+					t.Fatalf("Timeout waiting for consume to be closed")
+				}
+
+				if len(msgs) != 2*len(testMsgs) {
+					t.Fatalf("Unexpected received message count after consume closed; want %d; got %d", 2*len(testMsgs), len(msgs))
+				}
+			})
+		}
+	})
+
+	t.Run("wait for closed on already closed consume", func(t *testing.T) {
+		for i := 0; i < 10; i++ {
+			t.Run(fmt.Sprintf("run %d", i), func(t *testing.T) {
+				srv := RunBasicJetStreamServer()
+				defer shutdownJSServerAndRemoveStorage(t, srv)
+				nc, err := nats.Connect(srv.ClientURL())
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				js, err := jetstream.New(nc)
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				defer nc.Close()
+				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+				defer cancel()
+				s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				msgs := make([]jetstream.Msg, 0)
+				lock := sync.Mutex{}
+				publishTestMsgs(t, js)
+				cc, err := c.Consume(func(msg jetstream.Msg) {
+					time.Sleep(50 * time.Millisecond)
+					msg.Ack()
+					lock.Lock()
+					msgs = append(msgs, msg)
+					lock.Unlock()
+				})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				time.Sleep(100 * time.Millisecond)
+				if err := s.DeleteConsumer(context.Background(), c.CachedInfo().Name); err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				cc.Stop()
+
+				time.Sleep(100 * time.Millisecond)
+
+				select {
+				case <-cc.Closed():
+				case <-time.After(5 * time.Second):
+					t.Fatalf("Timeout waiting for consume to be closed")
+				}
+			})
+		}
+	})
 }
 
 func TestOrderedConsumerMessages(t *testing.T) {
@@ -1080,6 +1248,71 @@ func TestOrderedConsumerFetch(t *testing.T) {
 		}
 	})
 
+	t.Run("with custom deliver policy", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		msgs := make([]jetstream.Msg, 0)
+
+		for i := 0; i < 5; i++ {
+			if _, err := js.Publish(context.Background(), "FOO.A", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+		for i := 0; i < 5; i++ {
+			if _, err := js.Publish(context.Background(), "FOO.B", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{
+			DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy,
+		})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		res, err := c.Fetch(int(c.CachedInfo().NumPending), jetstream.FetchMaxWait(1*time.Second))
+		if err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+
+		for msg := range res.Messages() {
+			msgs = append(msgs, msg)
+		}
+
+		if res.Error() != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+
+		if len(msgs) != 2 {
+			t.Fatalf("Expected %d messages; got: %d", 2, len(msgs))
+		}
+		expectedSubjects := []string{"FOO.A", "FOO.B"}
+
+		for i := range msgs {
+			if msgs[i].Subject() != expectedSubjects[i] {
+				t.Fatalf("Expected subject: %s; got: %s", expectedSubjects[i], msgs[i].Subject())
+			}
+		}
+	})
+
 	t.Run("consumer used as consume", func(t *testing.T) {
 		srv := RunBasicJetStreamServer()
 		defer shutdownJSServerAndRemoveStorage(t, srv)
diff --git a/jetstream/test/publish_test.go b/jetstream/test/publish_test.go
index 2233a16..2ab6de8 100644
--- a/jetstream/test/publish_test.go
+++ b/jetstream/test/publish_test.go
@@ -1383,7 +1383,7 @@ func TestPublishAsyncResetPendingOnReconnect(t *testing.T) {
 					errs <- fmt.Errorf("Expected error: %v or %v; got: %v", nats.ErrDisconnected, nats.ErrNoResponders, err)
 				}
 			case <-time.After(5 * time.Second):
-				errs <- fmt.Errorf("Did not receive completion signal")
+				errs <- errors.New("Did not receive completion signal")
 			}
 			wg.Done()
 		}(ack)
diff --git a/jetstream/test/pull_test.go b/jetstream/test/pull_test.go
index f35aae3..4042e52 100644
--- a/jetstream/test/pull_test.go
+++ b/jetstream/test/pull_test.go
@@ -2584,6 +2584,154 @@ func TestPullConsumerConsume(t *testing.T) {
 		cc.Drain()
 		wg.Wait()
 	})
+
+	t.Run("wait for closed after drain", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		msgs := make([]jetstream.Msg, 0)
+		lock := sync.Mutex{}
+		publishTestMsgs(t, js)
+		cc, err := c.Consume(func(msg jetstream.Msg) {
+			time.Sleep(50 * time.Millisecond)
+			msg.Ack()
+			lock.Lock()
+			msgs = append(msgs, msg)
+			lock.Unlock()
+		})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		closed := cc.Closed()
+		time.Sleep(100 * time.Millisecond)
+
+		cc.Drain()
+
+		select {
+		case <-closed:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Timeout waiting for consume to be closed")
+		}
+
+		if len(msgs) != len(testMsgs) {
+			t.Fatalf("Unexpected received message count after consume closed; want %d; got %d", len(testMsgs), len(msgs))
+		}
+	})
+
+	t.Run("wait for closed after stop", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		msgs := make([]jetstream.Msg, 0)
+		lock := sync.Mutex{}
+		publishTestMsgs(t, js)
+		cc, err := c.Consume(func(msg jetstream.Msg) {
+			time.Sleep(50 * time.Millisecond)
+			msg.Ack()
+			lock.Lock()
+			msgs = append(msgs, msg)
+			lock.Unlock()
+		})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		time.Sleep(100 * time.Millisecond)
+		closed := cc.Closed()
+
+		cc.Stop()
+
+		select {
+		case <-closed:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Timeout waiting for consume to be closed")
+		}
+
+		if len(msgs) < 1 || len(msgs) > 3 {
+			t.Fatalf("Unexpected received message count after consume closed; want 1-3; got %d", len(msgs))
+		}
+	})
+
+	t.Run("wait for closed on already closed consume", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		publishTestMsgs(t, js)
+		cc, err := c.Consume(func(msg jetstream.Msg) {
+			time.Sleep(50 * time.Millisecond)
+			msg.Ack()
+		})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		time.Sleep(100 * time.Millisecond)
+
+		cc.Stop()
+
+		time.Sleep(100 * time.Millisecond)
+
+		select {
+		case <-cc.Closed():
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Timeout waiting for consume to be closed")
+		}
+	})
 }
 
 func TestPullConsumerConsume_WithCluster(t *testing.T) {
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index b9e0049..1dfafa0 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -504,7 +504,7 @@ func TestAddService(t *testing.T) {
 			}
 
 			if test.givenConfig.ErrorHandler != nil {
-				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.asyncErrorSubject}, fmt.Errorf("oops"))
+				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.asyncErrorSubject}, errors.New("oops"))
 				select {
 				case <-errService:
 				case <-time.After(1 * time.Second):
@@ -536,7 +536,7 @@ func TestAddService(t *testing.T) {
 				}
 			}
 			if test.natsErrorHandler != nil {
-				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.asyncErrorSubject}, fmt.Errorf("oops"))
+				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.asyncErrorSubject}, errors.New("oops"))
 				select {
 				case <-errService:
 					t.Fatalf("Expected to restore nats error handler")
@@ -634,7 +634,7 @@ func TestErrHandlerSubjectMatch(t *testing.T) {
 				}
 				defer svc.Stop()
 
-				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.errSubject}, fmt.Errorf("oops"))
+				go nc.Opts.AsyncErrorCB(nc, &nats.Subscription{Subject: test.errSubject}, errors.New("oops"))
 				if test.expectServiceErr {
 					select {
 					case <-errChan:
diff --git a/nats_test.go b/nats_test.go
index 7dcdce2..8f4546a 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -94,10 +94,11 @@ func checkErrChannel(t *testing.T, errCh chan error) {
 }
 
 func TestVersionMatchesTag(t *testing.T) {
-	tag := os.Getenv("TRAVIS_TAG")
-	if tag == "" {
+	refType := os.Getenv("GITHUB_REF_TYPE")
+	if refType != "tag" {
 		t.SkipNow()
 	}
+	tag := os.Getenv("GITHUB_REF_NAME")
 	// We expect a tag of the form vX.Y.Z. If that's not the case,
 	// we need someone to have a look. So fail if first letter is not
 	// a `v`
@@ -229,6 +230,7 @@ func TestSimplifiedURLs(t *testing.T) {
 		{
 			"nats",
 			[]string{
+				"nats://host1:1234/",
 				"nats://host1:1234",
 				"nats://host2:",
 				"nats://host3",
@@ -242,6 +244,7 @@ func TestSimplifiedURLs(t *testing.T) {
 				"[17:18:19:20]:1234",
 			},
 			[]string{
+				"nats://host1:1234/",
 				"nats://host1:1234",
 				"nats://host2:4222",
 				"nats://host3:4222",
@@ -434,6 +437,7 @@ func TestUrlArgument(t *testing.T) {
 	check("nats://localhost:1222 ", oneExpected)
 	check(" nats://localhost:1222", oneExpected)
 	check(" nats://localhost:1222 ", oneExpected)
+	check("nats://localhost:1222/", oneExpected)
 
 	var multiExpected = []string{
 		"nats://localhost:1222",
@@ -445,6 +449,7 @@ func TestUrlArgument(t *testing.T) {
 	check("nats://localhost:1222, nats://localhost:1223, nats://localhost:1224", multiExpected)
 	check(" nats://localhost:1222, nats://localhost:1223, nats://localhost:1224 ", multiExpected)
 	check("nats://localhost:1222,   nats://localhost:1223  ,nats://localhost:1224", multiExpected)
+	check("nats://localhost:1222/,nats://localhost:1223/,nats://localhost:1224/", multiExpected)
 }
 
 func TestParserPing(t *testing.T) {
diff --git a/test/auth_test.go b/test/auth_test.go
index a55b512..8fd0982 100644
--- a/test/auth_test.go
+++ b/test/auth_test.go
@@ -17,6 +17,8 @@ import (
 	"errors"
 	"fmt"
 	"io/fs"
+	"net"
+	"os"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -377,3 +379,75 @@ func TestConnectMissingCreds(t *testing.T) {
 		t.Fatalf("Expected not exists error, got: %v", err)
 	}
 }
+
+func TestUserInfoHandler(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+	listen: 127.0.0.1:-1
+	accounts: {
+	  A {
+		users: [{ user: "pp", password: "foo" }]
+	  }
+	}
+`))
+	defer os.Remove(conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	user, pass := "pp", "foo"
+	userInfoCB := func() (string, string) {
+		return user, pass
+	}
+
+	// check that we cannot set the user info twice
+	_, err := nats.Connect(s.ClientURL(), nats.UserInfo("pp", "foo"), nats.UserInfoHandler(userInfoCB))
+	if !errors.Is(err, nats.ErrUserInfoAlreadySet) {
+		t.Fatalf("Expected ErrUserInfoAlreadySet, got: %v", err)
+	}
+
+	addr, ok := s.Addr().(*net.TCPAddr)
+	if !ok {
+		t.Fatalf("Expected a TCP address, got %T", addr)
+	}
+
+	// check that user/pass from url takes precedence
+	_, err = nats.Connect(fmt.Sprintf("nats://bad:bad@localhost:%d", addr.Port),
+		nats.UserInfoHandler(userInfoCB))
+	if !errors.Is(err, nats.ErrAuthorization) {
+		t.Fatalf("Expected ErrAuthorization, got: %v", err)
+	}
+
+	// connect using the handler
+	nc, err := nats.Connect(s.ClientURL(),
+		nats.ReconnectWait(100*time.Millisecond),
+		nats.UserInfoHandler(userInfoCB))
+	if err != nil {
+		t.Fatalf("Error on connect: %v", err)
+	}
+	defer nc.Close()
+
+	// now change the password and reload the server
+	newConfig := []byte(`
+	listen: 127.0.0.1:-1
+	accounts: {
+	  A {
+		users: [{ user: "dd", password: "bar" }]
+	  }
+	}
+`)
+	if err := os.WriteFile(conf, newConfig, 0666); err != nil {
+		t.Fatalf("Error writing conf file: %v", err)
+	}
+
+	// update the user info used by the callback
+	user, pass = "dd", "bar"
+
+	status := nc.StatusChanged(nats.CONNECTED)
+
+	if err := s.Reload(); err != nil {
+		t.Fatalf("Error on reload: %v", err)
+	}
+
+	// we should get a reconnected event meaning the new credentials were used
+	WaitOnChannel(t, status, nats.CONNECTED)
+}
diff --git a/test/basic_test.go b/test/basic_test.go
index 75a187d..3161930 100644
--- a/test/basic_test.go
+++ b/test/basic_test.go
@@ -464,8 +464,8 @@ func TestQueueSubscriber(t *testing.T) {
 	omsg := []byte("Hello World")
 	nc.Publish("foo", omsg)
 	nc.Flush()
-	r1, _ := s1.QueuedMsgs()
-	r2, _ := s2.QueuedMsgs()
+	r1, _, _ := s1.Pending()
+	r2, _, _ := s2.Pending()
 	if (r1 + r2) != 1 {
 		t.Fatal("Received too many messages for multiple queue subscribers")
 	}
@@ -479,8 +479,8 @@ func TestQueueSubscriber(t *testing.T) {
 	}
 	nc.Flush()
 	v := uint(float32(total) * 0.15)
-	r1, _ = s1.QueuedMsgs()
-	r2, _ = s2.QueuedMsgs()
+	r1, _, _ = s1.Pending()
+	r2, _, _ = s2.Pending()
 	if r1+r2 != total {
 		t.Fatalf("Incorrect number of messages: %d vs %d", (r1 + r2), total)
 	}
@@ -1032,7 +1032,7 @@ func TestNilConnection(t *testing.T) {
 	if _, err := sub.NextMsg(time.Millisecond); err == nil || err != nats.ErrBadSubscription {
 		t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
 	}
-	if _, err := sub.QueuedMsgs(); err == nil || err != nats.ErrBadSubscription {
+	if _, _, err := sub.Pending(); err == nil || err != nats.ErrBadSubscription {
 		t.Fatalf("Expected ErrBadSubscription error, got %v\n", err)
 	}
 	if _, _, err := sub.Pending(); err == nil || err != nats.ErrBadSubscription {
diff --git a/test/cluster_test.go b/test/cluster_test.go
index a075296..05fab49 100644
--- a/test/cluster_test.go
+++ b/test/cluster_test.go
@@ -682,7 +682,7 @@ func (d *checkPoolUpdatedDialer) Dial(network, address string) (net.Conn, error)
 		doReal = true
 	} else if d.final {
 		d.ra++
-		return nil, fmt.Errorf("On purpose")
+		return nil, errors.New("On purpose")
 	} else {
 		d.ra++
 		if d.ra == 15 {
@@ -698,12 +698,12 @@ func (d *checkPoolUpdatedDialer) Dial(network, address string) (net.Conn, error)
 		d.conn = c
 		return c, nil
 	}
-	return nil, fmt.Errorf("On purpose")
+	return nil, errors.New("On purpose")
 }
 
 func TestServerPoolUpdatedWhenRouteGoesAway(t *testing.T) {
 	if err := serverVersionAtLeast(1, 0, 7); err != nil {
-		t.Skipf(err.Error())
+		t.Skip(err.Error())
 	}
 	s1Opts := test.DefaultTestOptions
 	s1Opts.Host = "127.0.0.1"
diff --git a/test/configs/docker/Dockerfile b/test/configs/docker/Dockerfile
index 037430d..96c69c7 100644
--- a/test/configs/docker/Dockerfile
+++ b/test/configs/docker/Dockerfile
@@ -1,4 +1,4 @@
-FROM golang:1.20
+FROM golang:1.22
 WORKDIR /usr/src/nats.go
 COPY . /usr/src/nats.go
 RUN go mod tidy -modfile go_test.mod
diff --git a/test/conn_test.go b/test/conn_test.go
index afc5025..1a47050 100644
--- a/test/conn_test.go
+++ b/test/conn_test.go
@@ -277,7 +277,7 @@ func TestClientTLSConfig(t *testing.T) {
 		pool := x509.NewCertPool()
 		ok := pool.AppendCertsFromPEM(rootCAs)
 		if !ok {
-			return nil, fmt.Errorf("nats: failed to parse root certificate from")
+			return nil, errors.New("nats: failed to parse root certificate from")
 		}
 		return pool, nil
 	}
@@ -614,7 +614,7 @@ func TestErrOnConnectAndDeadlock(t *testing.T) {
 		nc, err := nats.Connect(natsURL)
 		if err == nil {
 			nc.Close()
-			errCh <- fmt.Errorf("expected bad INFO err, got none")
+			errCh <- errors.New("expected bad INFO err, got none")
 			return
 		}
 		errCh <- nil
@@ -1749,7 +1749,7 @@ type customDialer struct {
 
 func (cd *customDialer) Dial(network, address string) (net.Conn, error) {
 	cd.ch <- true
-	return nil, fmt.Errorf("on purpose")
+	return nil, errors.New("on purpose")
 }
 
 func TestUseCustomDialer(t *testing.T) {
@@ -1847,8 +1847,8 @@ func TestDefaultOptionsDialer(t *testing.T) {
 	s := RunDefaultServer()
 	defer s.Shutdown()
 
-	opts1 := nats.DefaultOptions
-	opts2 := nats.DefaultOptions
+	opts1 := nats.GetDefaultOptions()
+	opts2 := nats.GetDefaultOptions()
 
 	nc1, err := opts1.Connect()
 	if err != nil {
diff --git a/test/context_test.go b/test/context_test.go
index b9c2f24..f2df307 100644
--- a/test/context_test.go
+++ b/test/context_test.go
@@ -654,324 +654,6 @@ func TestContextSubNextMsgWithDeadline(t *testing.T) {
 	}
 }
 
-func TestContextEncodedRequestWithTimeout(t *testing.T) {
-	s := RunDefaultServer()
-	defer s.Shutdown()
-
-	nc := NewDefaultConnection(t)
-	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-	if err != nil {
-		t.Fatalf("Unable to create encoded connection: %v", err)
-	}
-	defer c.Close()
-
-	deadline := time.Now().Add(100 * time.Millisecond)
-	ctx, cancelCB := context.WithDeadline(context.Background(), deadline)
-	defer cancelCB() // should always be called, not discarded, to prevent context leak
-
-	type request struct {
-		Message string `json:"message"`
-	}
-	type response struct {
-		Code int `json:"code"`
-	}
-	c.Subscribe("slow", func(_, reply string, req *request) {
-		got := req.Message
-		expected := "Hello"
-		if got != expected {
-			t.Errorf("Expected to receive request with %q, got %q", got, expected)
-		}
-
-		// simulates latency into the client so that timeout is hit.
-		time.Sleep(40 * time.Millisecond)
-		c.Publish(reply, &response{Code: 200})
-	})
-
-	for i := 0; i < 2; i++ {
-		req := &request{Message: "Hello"}
-		resp := &response{}
-		err := c.RequestWithContext(ctx, "slow", req, resp)
-		if err != nil {
-			t.Fatalf("Expected encoded request with context to not fail: %s", err)
-		}
-		got := resp.Code
-		expected := 200
-		if got != expected {
-			t.Errorf("Expected to receive %v, got: %v", expected, got)
-		}
-	}
-
-	// A third request with latency would make the context
-	// reach the deadline.
-	req := &request{Message: "Hello"}
-	resp := &response{}
-	err = c.RequestWithContext(ctx, "slow", req, resp)
-	if err == nil {
-		t.Fatal("Expected request with context to reach deadline")
-	}
-
-	// Reported error is "context deadline exceeded" from Context package,
-	// which implements net.Error Timeout interface.
-	type timeoutError interface {
-		Timeout() bool
-	}
-	timeoutErr, ok := err.(timeoutError)
-	if !ok || !timeoutErr.Timeout() {
-		t.Errorf("Expected to have a timeout error")
-	}
-	expected := `context deadline exceeded`
-	if !strings.Contains(err.Error(), expected) {
-		t.Errorf("Expected %q error, got: %q", expected, err.Error())
-	}
-}
-
-func TestContextEncodedRequestWithTimeoutCanceled(t *testing.T) {
-	s := RunDefaultServer()
-	defer s.Shutdown()
-
-	nc := NewDefaultConnection(t)
-	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-	if err != nil {
-		t.Fatalf("Unable to create encoded connection: %v", err)
-	}
-	defer c.Close()
-
-	ctx, cancelCB := context.WithTimeout(context.Background(), 100*time.Millisecond)
-	defer cancelCB() // should always be called, not discarded, to prevent context leak
-
-	type request struct {
-		Message string `json:"message"`
-	}
-	type response struct {
-		Code int `json:"code"`
-	}
-
-	c.Subscribe("fast", func(_, reply string, req *request) {
-		got := req.Message
-		expected := "Hello"
-		if got != expected {
-			t.Errorf("Expected to receive request with %q, got %q", got, expected)
-		}
-
-		// simulates latency into the client so that timeout is hit.
-		time.Sleep(40 * time.Millisecond)
-
-		c.Publish(reply, &response{Code: 200})
-	})
-
-	// Fast request should not fail
-	req := &request{Message: "Hello"}
-	resp := &response{}
-	c.RequestWithContext(ctx, "fast", req, resp)
-	expectedCode := 200
-	if resp.Code != expectedCode {
-		t.Errorf("Expected to receive %d, got: %d", expectedCode, resp.Code)
-	}
-
-	// Cancel the context already so that rest of requests fail.
-	cancelCB()
-
-	err = c.RequestWithContext(ctx, "fast", req, resp)
-	if err == nil {
-		t.Fatal("Expected request with timeout context to fail")
-	}
-
-	// Reported error is "context canceled" from Context package,
-	// which is not a timeout error.
-	type timeoutError interface {
-		Timeout() bool
-	}
-	if _, ok := err.(timeoutError); ok {
-		t.Errorf("Expected to not have a timeout error")
-	}
-	expected := `context canceled`
-	if !strings.Contains(err.Error(), expected) {
-		t.Errorf("Expected %q error, got: %q", expected, err.Error())
-	}
-
-	// 2nd request should fail again even if fast because context has already been canceled
-	err = c.RequestWithContext(ctx, "fast", req, resp)
-	if err == nil {
-		t.Fatal("Expected request with timeout context to fail")
-	}
-}
-
-func TestContextEncodedRequestWithCancel(t *testing.T) {
-	s := RunDefaultServer()
-	defer s.Shutdown()
-
-	nc := NewDefaultConnection(t)
-	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-	if err != nil {
-		t.Fatalf("Unable to create encoded connection: %v", err)
-	}
-	defer c.Close()
-
-	ctx, cancelCB := context.WithCancel(context.Background())
-	defer cancelCB() // should always be called, not discarded, to prevent context leak
-
-	// timer which cancels the context though can also be arbitrarily extended
-	expirationTimer := time.AfterFunc(100*time.Millisecond, func() {
-		cancelCB()
-	})
-
-	type request struct {
-		Message string `json:"message"`
-	}
-	type response struct {
-		Code int `json:"code"`
-	}
-	c.Subscribe("slow", func(_, reply string, req *request) {
-		got := req.Message
-		expected := "Hello"
-		if got != expected {
-			t.Errorf("Expected to receive request with %q, got %q", got, expected)
-		}
-
-		// simulates latency into the client so that timeout is hit.
-		time.Sleep(40 * time.Millisecond)
-		c.Publish(reply, &response{Code: 200})
-	})
-	c.Subscribe("slower", func(_, reply string, req *request) {
-		got := req.Message
-		expected := "World"
-		if got != expected {
-			t.Errorf("Expected to receive request with %q, got %q", got, expected)
-		}
-
-		// 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)
-		c.Publish(reply, &response{Code: 200})
-	})
-
-	for i := 0; i < 2; i++ {
-		req := &request{Message: "Hello"}
-		resp := &response{}
-		err := c.RequestWithContext(ctx, "slow", req, resp)
-		if err != nil {
-			t.Fatalf("Expected encoded request with context to not fail: %s", err)
-		}
-		got := resp.Code
-		expected := 200
-		if got != expected {
-			t.Errorf("Expected to receive %v, got: %v", expected, got)
-		}
-	}
-
-	// A third request with latency would make the context
-	// get canceled, but these reset the timer so deadline
-	// gets extended:
-	for i := 0; i < 10; i++ {
-		req := &request{Message: "World"}
-		resp := &response{}
-		err := c.RequestWithContext(ctx, "slower", req, resp)
-		if err != nil {
-			t.Fatalf("Expected request with context to not fail: %s", err)
-		}
-		got := resp.Code
-		expected := 200
-		if got != expected {
-			t.Errorf("Expected to receive %d, got: %d", expected, got)
-		}
-	}
-
-	req := &request{Message: "Hello"}
-	resp := &response{}
-
-	// One more slow request will expire the timer and cause an error...
-	err = c.RequestWithContext(ctx, "slow", req, resp)
-	if err == nil {
-		t.Fatal("Expected request with cancellation context to fail")
-	}
-
-	// ...though reported error is "context canceled" from Context package,
-	// which is not a timeout error.
-	type timeoutError interface {
-		Timeout() bool
-	}
-	if _, ok := err.(timeoutError); ok {
-		t.Errorf("Expected to not have a timeout error")
-	}
-	expected := `context canceled`
-	if !strings.Contains(err.Error(), expected) {
-		t.Errorf("Expected %q error, got: %q", expected, err.Error())
-	}
-}
-
-func TestContextEncodedRequestWithDeadline(t *testing.T) {
-	s := RunDefaultServer()
-	defer s.Shutdown()
-
-	nc := NewDefaultConnection(t)
-	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-	if err != nil {
-		t.Fatalf("Unable to create encoded connection: %v", err)
-	}
-	defer c.Close()
-
-	deadline := time.Now().Add(100 * time.Millisecond)
-	ctx, cancelCB := context.WithDeadline(context.Background(), deadline)
-	defer cancelCB() // should always be called, not discarded, to prevent context leak
-
-	type request struct {
-		Message string `json:"message"`
-	}
-	type response struct {
-		Code int `json:"code"`
-	}
-	c.Subscribe("slow", func(_, reply string, req *request) {
-		got := req.Message
-		expected := "Hello"
-		if got != expected {
-			t.Errorf("Expected to receive request with %q, got %q", got, expected)
-		}
-
-		// simulates latency into the client so that timeout is hit.
-		time.Sleep(40 * time.Millisecond)
-		c.Publish(reply, &response{Code: 200})
-	})
-
-	for i := 0; i < 2; i++ {
-		req := &request{Message: "Hello"}
-		resp := &response{}
-		err := c.RequestWithContext(ctx, "slow", req, resp)
-		if err != nil {
-			t.Fatalf("Expected encoded request with context to not fail: %s", err)
-		}
-		got := resp.Code
-		expected := 200
-		if got != expected {
-			t.Errorf("Expected to receive %v, got: %v", expected, got)
-		}
-	}
-
-	// A third request with latency would make the context
-	// reach the deadline.
-	req := &request{Message: "Hello"}
-	resp := &response{}
-	err = c.RequestWithContext(ctx, "slow", req, resp)
-	if err == nil {
-		t.Fatal("Expected request with context to reach deadline")
-	}
-
-	// Reported error is "context deadline exceeded" from Context package,
-	// which implements net.Error Timeout interface.
-	type timeoutError interface {
-		Timeout() bool
-	}
-	timeoutErr, ok := err.(timeoutError)
-	if !ok || !timeoutErr.Timeout() {
-		t.Errorf("Expected to have a timeout error")
-	}
-	expected := `context deadline exceeded`
-	if !strings.Contains(err.Error(), expected) {
-		t.Errorf("Expected %q error, got: %q", expected, err.Error())
-	}
-}
-
 func TestContextRequestConnClosed(t *testing.T) {
 	s := RunDefaultServer()
 	defer s.Shutdown()
@@ -1026,58 +708,6 @@ func TestContextBadSubscription(t *testing.T) {
 	}
 }
 
-func TestContextInvalid(t *testing.T) {
-	s := RunDefaultServer()
-	defer s.Shutdown()
-
-	nc := NewDefaultConnection(t)
-	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-	if err != nil {
-		t.Fatalf("Unable to create encoded connection: %v", err)
-	}
-	defer c.Close()
-
-	//lint:ignore SA1012 testing that passing nil fails
-	_, err = nc.RequestWithContext(nil, "foo", []byte(""))
-	if err == nil {
-		t.Fatal("Expected request to fail with error")
-	}
-	if err != nats.ErrInvalidContext {
-		t.Errorf("Expected request to fail with connection closed error: %s", err)
-	}
-
-	sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {})
-	if err != nil {
-		t.Fatalf("Expected to be able to subscribe: %s", err)
-	}
-
-	//lint:ignore SA1012 testing that passing nil fails
-	_, err = sub.NextMsgWithContext(nil)
-	if err == nil {
-		t.Fatal("Expected request to fail with error")
-	}
-	if err != nats.ErrInvalidContext {
-		t.Errorf("Expected request to fail with connection closed error: %s", err)
-	}
-
-	type request struct {
-		Message string `json:"message"`
-	}
-	type response struct {
-		Code int `json:"code"`
-	}
-	req := &request{Message: "Hello"}
-	resp := &response{}
-	//lint:ignore SA1012 testing that passing nil fails
-	err = c.RequestWithContext(nil, "slow", req, resp)
-	if err == nil {
-		t.Fatal("Expected request to fail with error")
-	}
-	if err != nats.ErrInvalidContext {
-		t.Errorf("Expected request to fail with invalid context: %s", err)
-	}
-}
-
 func TestFlushWithContext(t *testing.T) {
 	s := RunDefaultServer()
 	defer s.Shutdown()
@@ -1148,3 +778,34 @@ func TestUnsubscribeAndNextMsgWithContext(t *testing.T) {
 	}
 	wg.Wait()
 }
+
+func TestContextInvalid(t *testing.T) {
+	s := RunDefaultServer()
+	defer s.Shutdown()
+
+	nc := NewDefaultConnection(t)
+	defer nc.Close()
+
+	//lint:ignore SA1012 testing that passing nil fails
+	_, err := nc.RequestWithContext(nil, "foo", []byte(""))
+	if err == nil {
+		t.Fatal("Expected request to fail with error")
+	}
+	if err != nats.ErrInvalidContext {
+		t.Errorf("Expected request to fail with connection closed error: %s", err)
+	}
+
+	sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {})
+	if err != nil {
+		t.Fatalf("Expected to be able to subscribe: %s", err)
+	}
+
+	//lint:ignore SA1012 testing that passing nil fails
+	_, err = sub.NextMsgWithContext(nil)
+	if err == nil {
+		t.Fatal("Expected request to fail with error")
+	}
+	if err != nats.ErrInvalidContext {
+		t.Errorf("Expected request to fail with connection closed error: %s", err)
+	}
+}
diff --git a/test/drain_test.go b/test/drain_test.go
index c53305e..da07c89 100644
--- a/test/drain_test.go
+++ b/test/drain_test.go
@@ -14,6 +14,7 @@
 package test
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -225,7 +226,7 @@ func TestDrainSlowSubscriber(t *testing.T) {
 		// Wait for it to become invalid. Once drained it is unsubscribed.
 		_, _, err := sub.Pending()
 		if err != nats.ErrBadSubscription {
-			return fmt.Errorf("Still valid")
+			return errors.New("Still valid")
 		}
 		r := int(atomic.LoadInt32(&received))
 		if r != total {
@@ -478,7 +479,7 @@ func TestDrainConnDuringReconnect(t *testing.T) {
 		if nc.IsReconnecting() {
 			return nil
 		}
-		return fmt.Errorf("Not reconnecting yet")
+		return errors.New("Not reconnecting yet")
 	})
 
 	// This should work correctly.
diff --git a/test/enc_test.go b/test/enc_test.go
index e40abbf..c8109e7 100644
--- a/test/enc_test.go
+++ b/test/enc_test.go
@@ -15,7 +15,9 @@ package test
 
 import (
 	"bytes"
+	"context"
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -25,6 +27,8 @@ import (
 	"github.com/nats-io/nats.go/encoders/protobuf/testdata"
 )
 
+//lint:file-ignore SA1019 Ignore deprecation warnings for EncodedConn
+
 func NewDefaultEConn(t *testing.T) *nats.EncodedConn {
 	ec, err := nats.NewEncodedConn(NewConnection(t, TEST_PORT), nats.DEFAULT_ENCODER)
 	if err != nil {
@@ -753,3 +757,350 @@ func TestRequestGOB(t *testing.T) {
 		t.Fatalf("Did not receive proper response, %+v", reply)
 	}
 }
+
+func TestContextEncodedRequestWithTimeout(t *testing.T) {
+	s := RunDefaultServer()
+	defer s.Shutdown()
+
+	nc := NewDefaultConnection(t)
+	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
+	if err != nil {
+		t.Fatalf("Unable to create encoded connection: %v", err)
+	}
+	defer c.Close()
+
+	deadline := time.Now().Add(100 * time.Millisecond)
+	ctx, cancelCB := context.WithDeadline(context.Background(), deadline)
+	defer cancelCB() // should always be called, not discarded, to prevent context leak
+
+	type request struct {
+		Message string `json:"message"`
+	}
+	type response struct {
+		Code int `json:"code"`
+	}
+	c.Subscribe("slow", func(_, reply string, req *request) {
+		got := req.Message
+		expected := "Hello"
+		if got != expected {
+			t.Errorf("Expected to receive request with %q, got %q", got, expected)
+		}
+
+		// simulates latency into the client so that timeout is hit.
+		time.Sleep(40 * time.Millisecond)
+		c.Publish(reply, &response{Code: 200})
+	})
+
+	for i := 0; i < 2; i++ {
+		req := &request{Message: "Hello"}
+		resp := &response{}
+		err := c.RequestWithContext(ctx, "slow", req, resp)
+		if err != nil {
+			t.Fatalf("Expected encoded request with context to not fail: %s", err)
+		}
+		got := resp.Code
+		expected := 200
+		if got != expected {
+			t.Errorf("Expected to receive %v, got: %v", expected, got)
+		}
+	}
+
+	// A third request with latency would make the context
+	// reach the deadline.
+	req := &request{Message: "Hello"}
+	resp := &response{}
+	err = c.RequestWithContext(ctx, "slow", req, resp)
+	if err == nil {
+		t.Fatal("Expected request with context to reach deadline")
+	}
+
+	// Reported error is "context deadline exceeded" from Context package,
+	// which implements net.Error Timeout interface.
+	type timeoutError interface {
+		Timeout() bool
+	}
+	timeoutErr, ok := err.(timeoutError)
+	if !ok || !timeoutErr.Timeout() {
+		t.Errorf("Expected to have a timeout error")
+	}
+	expected := `context deadline exceeded`
+	if !strings.Contains(err.Error(), expected) {
+		t.Errorf("Expected %q error, got: %q", expected, err.Error())
+	}
+}
+
+func TestContextEncodedRequestWithTimeoutCanceled(t *testing.T) {
+	s := RunDefaultServer()
+	defer s.Shutdown()
+
+	nc := NewDefaultConnection(t)
+	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
+	if err != nil {
+		t.Fatalf("Unable to create encoded connection: %v", err)
+	}
+	defer c.Close()
+
+	ctx, cancelCB := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancelCB() // should always be called, not discarded, to prevent context leak
+
+	type request struct {
+		Message string `json:"message"`
+	}
+	type response struct {
+		Code int `json:"code"`
+	}
+
+	c.Subscribe("fast", func(_, reply string, req *request) {
+		got := req.Message
+		expected := "Hello"
+		if got != expected {
+			t.Errorf("Expected to receive request with %q, got %q", got, expected)
+		}
+
+		// simulates latency into the client so that timeout is hit.
+		time.Sleep(40 * time.Millisecond)
+
+		c.Publish(reply, &response{Code: 200})
+	})
+
+	// Fast request should not fail
+	req := &request{Message: "Hello"}
+	resp := &response{}
+	c.RequestWithContext(ctx, "fast", req, resp)
+	expectedCode := 200
+	if resp.Code != expectedCode {
+		t.Errorf("Expected to receive %d, got: %d", expectedCode, resp.Code)
+	}
+
+	// Cancel the context already so that rest of requests fail.
+	cancelCB()
+
+	err = c.RequestWithContext(ctx, "fast", req, resp)
+	if err == nil {
+		t.Fatal("Expected request with timeout context to fail")
+	}
+
+	// Reported error is "context canceled" from Context package,
+	// which is not a timeout error.
+	type timeoutError interface {
+		Timeout() bool
+	}
+	if _, ok := err.(timeoutError); ok {
+		t.Errorf("Expected to not have a timeout error")
+	}
+	expected := `context canceled`
+	if !strings.Contains(err.Error(), expected) {
+		t.Errorf("Expected %q error, got: %q", expected, err.Error())
+	}
+
+	// 2nd request should fail again even if fast because context has already been canceled
+	err = c.RequestWithContext(ctx, "fast", req, resp)
+	if err == nil {
+		t.Fatal("Expected request with timeout context to fail")
+	}
+}
+
+func TestContextEncodedRequestWithCancel(t *testing.T) {
+	s := RunDefaultServer()
+	defer s.Shutdown()
+
+	nc := NewDefaultConnection(t)
+	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
+	if err != nil {
+		t.Fatalf("Unable to create encoded connection: %v", err)
+	}
+	defer c.Close()
+
+	ctx, cancelCB := context.WithCancel(context.Background())
+	defer cancelCB() // should always be called, not discarded, to prevent context leak
+
+	// timer which cancels the context though can also be arbitrarily extended
+	expirationTimer := time.AfterFunc(100*time.Millisecond, func() {
+		cancelCB()
+	})
+
+	type request struct {
+		Message string `json:"message"`
+	}
+	type response struct {
+		Code int `json:"code"`
+	}
+	c.Subscribe("slow", func(_, reply string, req *request) {
+		got := req.Message
+		expected := "Hello"
+		if got != expected {
+			t.Errorf("Expected to receive request with %q, got %q", got, expected)
+		}
+
+		// simulates latency into the client so that timeout is hit.
+		time.Sleep(40 * time.Millisecond)
+		c.Publish(reply, &response{Code: 200})
+	})
+	c.Subscribe("slower", func(_, reply string, req *request) {
+		got := req.Message
+		expected := "World"
+		if got != expected {
+			t.Errorf("Expected to receive request with %q, got %q", got, expected)
+		}
+
+		// 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)
+		c.Publish(reply, &response{Code: 200})
+	})
+
+	for i := 0; i < 2; i++ {
+		req := &request{Message: "Hello"}
+		resp := &response{}
+		err := c.RequestWithContext(ctx, "slow", req, resp)
+		if err != nil {
+			t.Fatalf("Expected encoded request with context to not fail: %s", err)
+		}
+		got := resp.Code
+		expected := 200
+		if got != expected {
+			t.Errorf("Expected to receive %v, got: %v", expected, got)
+		}
+	}
+
+	// A third request with latency would make the context
+	// get canceled, but these reset the timer so deadline
+	// gets extended:
+	for i := 0; i < 10; i++ {
+		req := &request{Message: "World"}
+		resp := &response{}
+		err := c.RequestWithContext(ctx, "slower", req, resp)
+		if err != nil {
+			t.Fatalf("Expected request with context to not fail: %s", err)
+		}
+		got := resp.Code
+		expected := 200
+		if got != expected {
+			t.Errorf("Expected to receive %d, got: %d", expected, got)
+		}
+	}
+
+	req := &request{Message: "Hello"}
+	resp := &response{}
+
+	// One more slow request will expire the timer and cause an error...
+	err = c.RequestWithContext(ctx, "slow", req, resp)
+	if err == nil {
+		t.Fatal("Expected request with cancellation context to fail")
+	}
+
+	// ...though reported error is "context canceled" from Context package,
+	// which is not a timeout error.
+	type timeoutError interface {
+		Timeout() bool
+	}
+	if _, ok := err.(timeoutError); ok {
+		t.Errorf("Expected to not have a timeout error")
+	}
+	expected := `context canceled`
+	if !strings.Contains(err.Error(), expected) {
+		t.Errorf("Expected %q error, got: %q", expected, err.Error())
+	}
+}
+
+func TestContextEncodedRequestWithDeadline(t *testing.T) {
+	s := RunDefaultServer()
+	defer s.Shutdown()
+
+	nc := NewDefaultConnection(t)
+	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
+	if err != nil {
+		t.Fatalf("Unable to create encoded connection: %v", err)
+	}
+	defer c.Close()
+
+	deadline := time.Now().Add(100 * time.Millisecond)
+	ctx, cancelCB := context.WithDeadline(context.Background(), deadline)
+	defer cancelCB() // should always be called, not discarded, to prevent context leak
+
+	type request struct {
+		Message string `json:"message"`
+	}
+	type response struct {
+		Code int `json:"code"`
+	}
+	c.Subscribe("slow", func(_, reply string, req *request) {
+		got := req.Message
+		expected := "Hello"
+		if got != expected {
+			t.Errorf("Expected to receive request with %q, got %q", got, expected)
+		}
+
+		// simulates latency into the client so that timeout is hit.
+		time.Sleep(40 * time.Millisecond)
+		c.Publish(reply, &response{Code: 200})
+	})
+
+	for i := 0; i < 2; i++ {
+		req := &request{Message: "Hello"}
+		resp := &response{}
+		err := c.RequestWithContext(ctx, "slow", req, resp)
+		if err != nil {
+			t.Fatalf("Expected encoded request with context to not fail: %s", err)
+		}
+		got := resp.Code
+		expected := 200
+		if got != expected {
+			t.Errorf("Expected to receive %v, got: %v", expected, got)
+		}
+	}
+
+	// A third request with latency would make the context
+	// reach the deadline.
+	req := &request{Message: "Hello"}
+	resp := &response{}
+	err = c.RequestWithContext(ctx, "slow", req, resp)
+	if err == nil {
+		t.Fatal("Expected request with context to reach deadline")
+	}
+
+	// Reported error is "context deadline exceeded" from Context package,
+	// which implements net.Error Timeout interface.
+	type timeoutError interface {
+		Timeout() bool
+	}
+	timeoutErr, ok := err.(timeoutError)
+	if !ok || !timeoutErr.Timeout() {
+		t.Errorf("Expected to have a timeout error")
+	}
+	expected := `context deadline exceeded`
+	if !strings.Contains(err.Error(), expected) {
+		t.Errorf("Expected %q error, got: %q", expected, err.Error())
+	}
+}
+
+func TestEncodedContextInvalid(t *testing.T) {
+	s := RunDefaultServer()
+	defer s.Shutdown()
+
+	nc := NewDefaultConnection(t)
+	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
+	if err != nil {
+		t.Fatalf("Unable to create encoded connection: %v", err)
+	}
+	defer c.Close()
+
+	type request struct {
+		Message string `json:"message"`
+	}
+	type response struct {
+		Code int `json:"code"`
+	}
+	req := &request{Message: "Hello"}
+	resp := &response{}
+	//lint:ignore SA1012 testing that passing nil fails
+	err = c.RequestWithContext(nil, "slow", req, resp)
+	if err == nil {
+		t.Fatal("Expected request to fail with error")
+	}
+	if err != nats.ErrInvalidContext {
+		t.Errorf("Expected request to fail with invalid context: %s", err)
+	}
+}
diff --git a/test/gob_test.go b/test/gob_test.go
index c772e10..f326f8c 100644
--- a/test/gob_test.go
+++ b/test/gob_test.go
@@ -20,6 +20,8 @@ import (
 	"github.com/nats-io/nats.go"
 )
 
+//lint:file-ignore SA1019 Ignore deprecation warnings for EncodedConn
+
 func NewGobEncodedConn(tl TestLogger) *nats.EncodedConn {
 	ec, err := nats.NewEncodedConn(NewConnection(tl, TEST_PORT), nats.GOB_ENCODER)
 	if err != nil {
diff --git a/test/helper_test.go b/test/helper_test.go
index 7f2aedf..36d47e0 100644
--- a/test/helper_test.go
+++ b/test/helper_test.go
@@ -103,15 +103,6 @@ func NewConnection(t tLogger, port int) *nats.Conn {
 	return nc
 }
 
-// NewEConn
-func NewEConn(t tLogger) *nats.EncodedConn {
-	ec, err := nats.NewEncodedConn(NewDefaultConnection(t), nats.DEFAULT_ENCODER)
-	if err != nil {
-		t.Fatalf("Failed to create an encoded connection: %v\n", err)
-	}
-	return ec
-}
-
 ////////////////////////////////////////////////////////////////////////////////
 // Running nats server in separate Go routines
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/test/js_test.go b/test/js_test.go
index 0363706..db791eb 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -1357,7 +1357,7 @@ func TestPullSubscribeFetchBatchWithHeartbeat(t *testing.T) {
 	if msgs.Error() != nil {
 		t.Fatalf("Unexpected error: %s", msgs.Error())
 	}
-	if elapsed < 290*time.Millisecond {
+	if elapsed < 250*time.Millisecond {
 		t.Fatalf("Expected timeout after 300ms; got: %v", elapsed)
 	}
 
@@ -8207,6 +8207,215 @@ func TestJetStreamPublishAsyncPerf(t *testing.T) {
 	fmt.Printf("%.0f msgs/sec\n\n", float64(toSend)/tt.Seconds())
 }
 
+func TestPublishAsyncRetry(t *testing.T) {
+	tests := []struct {
+		name     string
+		pubOpts  []nats.PubOpt
+		ackError error
+		pubErr   error
+	}{
+		{
+			name: "retry until stream is ready",
+			pubOpts: []nats.PubOpt{
+				nats.RetryAttempts(10),
+				nats.RetryWait(100 * time.Millisecond),
+			},
+		},
+		{
+			name: "fail after max retries",
+			pubOpts: []nats.PubOpt{
+				nats.RetryAttempts(2),
+				nats.RetryWait(50 * time.Millisecond),
+			},
+			ackError: nats.ErrNoResponders,
+		},
+		{
+			name:     "no retries",
+			pubOpts:  nil,
+			ackError: nats.ErrNoResponders,
+		},
+		{
+			name: "invalid retry attempts",
+			pubOpts: []nats.PubOpt{
+				nats.RetryAttempts(-1),
+			},
+			pubErr: nats.ErrInvalidArg,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s := RunBasicJetStreamServer()
+			defer shutdownJSServerAndRemoveStorage(t, s)
+
+			nc, err := nats.Connect(s.ClientURL())
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			// set max pending to 1 so that we can test if retries don't cause stall
+			js, err := nc.JetStream(nats.PublishAsyncMaxPending(1))
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			defer nc.Close()
+
+			test.pubOpts = append(test.pubOpts, nats.StallWait(1*time.Nanosecond))
+			ack, err := js.PublishAsync("foo", []byte("hello"), test.pubOpts...)
+			if !errors.Is(err, test.pubErr) {
+				t.Fatalf("Expected error: %v; got: %v", test.pubErr, err)
+			}
+			if err != nil {
+				return
+			}
+			errs := make(chan error, 1)
+			go func() {
+				// create stream with delay so that publish will receive no responders
+				time.Sleep(300 * time.Millisecond)
+				if _, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Subjects: []string{"foo"}}); err != nil {
+					errs <- err
+				}
+			}()
+			select {
+			case <-ack.Ok():
+			case err := <-ack.Err():
+				if test.ackError != nil {
+					if !errors.Is(err, test.ackError) {
+						t.Fatalf("Expected error: %v; got: %v", test.ackError, err)
+					}
+				} else {
+					t.Fatalf("Unexpected ack error: %v", err)
+				}
+			case err := <-errs:
+				t.Fatalf("Error creating stream: %v", err)
+			case <-time.After(5 * time.Second):
+				t.Fatalf("Timeout waiting for ack")
+			}
+		})
+	}
+}
+func TestJetStreamCleanupPublisher(t *testing.T) {
+
+	t.Run("cleanup js publisher", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		// Create a stream.
+		if _, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Subjects: []string{"FOO"}}); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		numSubs := nc.NumSubscriptions()
+		if _, err := js.PublishAsync("FOO", []byte("hello")); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		select {
+		case <-js.PublishAsyncComplete():
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Did not receive completion signal")
+		}
+
+		if numSubs+1 != nc.NumSubscriptions() {
+			t.Fatalf("Expected an additional subscription after publish, got %d", nc.NumSubscriptions())
+		}
+
+		js.CleanupPublisher()
+
+		if numSubs != nc.NumSubscriptions() {
+			t.Fatalf("Expected subscriptions to be back to original count")
+		}
+	})
+
+	t.Run("cleanup js publisher, cancel pending acks", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, err := nats.Connect(s.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		cbErr := make(chan error, 10)
+		js, err := nc.JetStream(nats.PublishAsyncErrHandler(func(js nats.JetStream, m *nats.Msg, err error) {
+			cbErr <- err
+		}))
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		// Create a stream with NoAck so that we can test that we cancel ack futures.
+		if _, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Subjects: []string{"FOO"}, NoAck: true}); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		numSubs := nc.NumSubscriptions()
+
+		var acks []nats.PubAckFuture
+		for i := 0; i < 10; i++ {
+			ack, err := js.PublishAsync("FOO", []byte("hello"))
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			acks = append(acks, ack)
+		}
+
+		asyncComplete := js.PublishAsyncComplete()
+		select {
+		case <-asyncComplete:
+			t.Fatalf("Should not complete, NoAck is set")
+		case <-time.After(200 * time.Millisecond):
+		}
+
+		if numSubs+1 != nc.NumSubscriptions() {
+			t.Fatalf("Expected an additional subscription after publish, got %d", nc.NumSubscriptions())
+		}
+
+		js.CleanupPublisher()
+
+		if numSubs != nc.NumSubscriptions() {
+			t.Fatalf("Expected subscriptions to be back to original count")
+		}
+
+		// check that PublishAsyncComplete channel is closed
+		select {
+		case <-asyncComplete:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Did not receive completion signal")
+		}
+
+		// check that all ack futures are canceled
+		for _, ack := range acks {
+			select {
+			case err := <-ack.Err():
+				if !errors.Is(err, nats.ErrJetStreamPublisherClosed) {
+					t.Fatalf("Expected JetStreamContextClosed error, got %v", err)
+				}
+			case <-ack.Ok():
+				t.Fatalf("Expected error on the ack future")
+			case <-time.After(200 * time.Millisecond):
+				t.Fatalf("Expected an error on the ack future")
+			}
+		}
+
+		// check that async error handler is called for each pending ack
+		for i := 0; i < 10; i++ {
+			select {
+			case err := <-cbErr:
+				if !errors.Is(err, nats.ErrJetStreamPublisherClosed) {
+					t.Fatalf("Expected JetStreamContextClosed error, got %v", err)
+				}
+			case <-time.After(200 * time.Millisecond):
+				t.Fatalf("Expected errors to be passed from the async handler")
+			}
+		}
+	})
+
+}
+
 func TestJetStreamPublishExpectZero(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer shutdownJSServerAndRemoveStorage(t, s)
@@ -9217,7 +9426,7 @@ func TestJetStreamClusterStreamLeaderChangeClientErr(t *testing.T) {
 						return err
 					}
 					if si.Cluster.Leader == "" {
-						return fmt.Errorf("No leader yet")
+						return errors.New("No leader yet")
 					}
 					return nil
 				})
diff --git a/test/json_test.go b/test/json_test.go
index 4ef6f42..cebbf31 100644
--- a/test/json_test.go
+++ b/test/json_test.go
@@ -22,6 +22,8 @@ import (
 	"github.com/nats-io/nats.go/encoders/builtin"
 )
 
+//lint:file-ignore SA1019 Ignore deprecation warnings for EncodedConn
+
 func NewJsonEncodedConn(tl TestLogger) *nats.EncodedConn {
 	ec, err := nats.NewEncodedConn(NewConnection(tl, TEST_PORT), nats.JSON_ENCODER)
 	if err != nil {
diff --git a/test/kv_test.go b/test/kv_test.go
index a1b5ca4..7688958 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -1113,7 +1113,7 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 		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")
+				return errors.New("Expected key to be gone")
 			}
 			if !errors.Is(err, nats.ErrKeyNotFound) {
 				return err
diff --git a/test/netchan_test.go b/test/netchan_test.go
index b7272a9..f21d329 100644
--- a/test/netchan_test.go
+++ b/test/netchan_test.go
@@ -20,6 +20,17 @@ import (
 	"github.com/nats-io/nats.go"
 )
 
+//lint:file-ignore SA1019 Ignore deprecation warnings for EncodedConn
+
+// NewEConn
+func NewEConn(t tLogger) *nats.EncodedConn {
+	ec, err := nats.NewEncodedConn(NewDefaultConnection(t), nats.DEFAULT_ENCODER)
+	if err != nil {
+		t.Fatalf("Failed to create an encoded connection: %v\n", err)
+	}
+	return ec
+}
+
 func TestBadChan(t *testing.T) {
 	s := RunDefaultServer()
 	defer s.Shutdown()
diff --git a/test/protobuf_test.go b/test/protobuf_test.go
index 08fdf67..d4bee8c 100644
--- a/test/protobuf_test.go
+++ b/test/protobuf_test.go
@@ -25,6 +25,8 @@ import (
 	pb "github.com/nats-io/nats.go/encoders/protobuf/testdata"
 )
 
+//lint:file-ignore SA1019 Ignore deprecation warnings for EncodedConn
+
 func NewProtoEncodedConn(tl TestLogger) *nats.EncodedConn {
 	ec, err := nats.NewEncodedConn(NewConnection(tl, TEST_PORT), protobuf.PROTOBUF_ENCODER)
 	if err != nil {
diff --git a/test/reconnect_test.go b/test/reconnect_test.go
index e543db7..9fc1b23 100644
--- a/test/reconnect_test.go
+++ b/test/reconnect_test.go
@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"net"
 	"net/url"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -178,19 +179,15 @@ func TestBasicReconnectFunctionality(t *testing.T) {
 		t.Fatalf("Should have connected ok: %v\n", err)
 	}
 	defer nc.Close()
-	ec, err := nats.NewEncodedConn(nc, nats.DEFAULT_ENCODER)
-	if err != nil {
-		t.Fatalf("Failed to create an encoded connection: %v\n", err)
-	}
 
 	testString := "bar"
-	ec.Subscribe("foo", func(s string) {
-		if s != testString {
+	nc.Subscribe("foo", func(m *nats.Msg) {
+		if string(m.Data) != testString {
 			t.Fatal("String doesn't match")
 		}
 		ch <- true
 	})
-	ec.Flush()
+	nc.Flush()
 
 	ts.Shutdown()
 	// server is stopped here...
@@ -199,14 +196,14 @@ func TestBasicReconnectFunctionality(t *testing.T) {
 		t.Fatalf("Did not get the disconnected callback on time\n")
 	}
 
-	if err := ec.Publish("foo", testString); err != nil {
+	if err := nc.Publish("foo", []byte("bar")); err != nil {
 		t.Fatalf("Failed to publish message: %v\n", err)
 	}
 
 	ts = startReconnectServer(t)
 	defer ts.Shutdown()
 
-	if err := ec.FlushTimeout(5 * time.Second); err != nil {
+	if err := nc.FlushTimeout(5 * time.Second); err != nil {
 		t.Fatalf("Error on Flush: %v", err)
 	}
 
@@ -215,7 +212,7 @@ func TestBasicReconnectFunctionality(t *testing.T) {
 	}
 
 	expectedReconnectCount := uint64(1)
-	reconnectCount := ec.Conn.Stats().Reconnects
+	reconnectCount := nc.Stats().Reconnects
 
 	if reconnectCount != expectedReconnectCount {
 		t.Fatalf("Reconnect count incorrect: %d vs %d\n",
@@ -241,23 +238,20 @@ func TestExtendedReconnectFunctionality(t *testing.T) {
 		t.Fatalf("Should have connected ok: %v", err)
 	}
 	defer nc.Close()
-	ec, err := nats.NewEncodedConn(nc, nats.DEFAULT_ENCODER)
-	if err != nil {
-		t.Fatalf("Failed to create an encoded connection: %v\n", err)
-	}
+
 	testString := "bar"
 	received := int32(0)
 
-	ec.Subscribe("foo", func(s string) {
+	nc.Subscribe("foo", func(*nats.Msg) {
 		atomic.AddInt32(&received, 1)
 	})
 
-	sub, _ := ec.Subscribe("foobar", func(s string) {
+	sub, _ := nc.Subscribe("foobar", func(*nats.Msg) {
 		atomic.AddInt32(&received, 1)
 	})
 
-	ec.Publish("foo", testString)
-	ec.Flush()
+	nc.Publish("foo", []byte(testString))
+	nc.Flush()
 
 	ts.Shutdown()
 	// server is stopped here..
@@ -268,18 +262,18 @@ func TestExtendedReconnectFunctionality(t *testing.T) {
 	}
 
 	// Sub while disconnected
-	ec.Subscribe("bar", func(s string) {
+	nc.Subscribe("bar", func(*nats.Msg) {
 		atomic.AddInt32(&received, 1)
 	})
 
 	// Unsub foobar while disconnected
 	sub.Unsubscribe()
 
-	if err = ec.Publish("foo", testString); err != nil {
+	if err = nc.Publish("foo", []byte(testString)); err != nil {
 		t.Fatalf("Received an error after disconnect: %v\n", err)
 	}
 
-	if err = ec.Publish("bar", testString); err != nil {
+	if err = nc.Publish("bar", []byte(testString)); err != nil {
 		t.Fatalf("Received an error after disconnect: %v\n", err)
 	}
 
@@ -292,19 +286,19 @@ func TestExtendedReconnectFunctionality(t *testing.T) {
 		t.Fatal("Did not receive a reconnect callback message")
 	}
 
-	if err = ec.Publish("foobar", testString); err != nil {
+	if err = nc.Publish("foobar", []byte(testString)); err != nil {
 		t.Fatalf("Received an error after server restarted: %v\n", err)
 	}
 
-	if err = ec.Publish("foo", testString); err != nil {
+	if err = nc.Publish("foo", []byte(testString)); err != nil {
 		t.Fatalf("Received an error after server restarted: %v\n", err)
 	}
 
 	ch := make(chan bool)
-	ec.Subscribe("done", func(b bool) {
+	nc.Subscribe("done", func(*nats.Msg) {
 		ch <- true
 	})
-	ec.Publish("done", true)
+	nc.Publish("done", nil)
 
 	if e := Wait(ch); e != nil {
 		t.Fatal("Did not receive our message")
@@ -337,11 +331,6 @@ func TestQueueSubsOnReconnect(t *testing.T) {
 	}
 	defer nc.Close()
 
-	ec, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-	if err != nil {
-		t.Fatalf("Failed to create an encoded connection: %v\n", err)
-	}
-
 	// To hold results.
 	results := make(map[int]int)
 	var mu sync.Mutex
@@ -364,25 +353,29 @@ func TestQueueSubsOnReconnect(t *testing.T) {
 	subj := "foo.bar"
 	qgroup := "workers"
 
-	cb := func(seqno int) {
+	cb := func(m *nats.Msg) {
 		mu.Lock()
 		defer mu.Unlock()
+		seqno, err := strconv.Atoi(string(m.Data))
+		if err != nil {
+			t.Fatalf("Received an invalid sequence number: %v\n", err)
+		}
 		results[seqno] = results[seqno] + 1
 	}
 
 	// Create Queue Subscribers
-	ec.QueueSubscribe(subj, qgroup, cb)
-	ec.QueueSubscribe(subj, qgroup, cb)
+	nc.QueueSubscribe(subj, qgroup, cb)
+	nc.QueueSubscribe(subj, qgroup, cb)
 
-	ec.Flush()
+	nc.Flush()
 
 	// Helper function to send messages and check results.
 	sendAndCheckMsgs := func(numToSend int) {
 		for i := 0; i < numToSend; i++ {
-			ec.Publish(subj, i)
+			nc.Publish(subj, []byte(fmt.Sprint(i)))
 		}
 		// Wait for processing.
-		ec.Flush()
+		nc.Flush()
 		time.Sleep(50 * time.Millisecond)
 
 		// Check Results
diff --git a/test/sub_test.go b/test/sub_test.go
index f0f83a8..a27e142 100644
--- a/test/sub_test.go
+++ b/test/sub_test.go
@@ -1120,7 +1120,7 @@ func TestAsyncSubscriptionPending(t *testing.T) {
 	}
 
 	// Test old way
-	q, _ := sub.QueuedMsgs()
+	q, _, _ := sub.Pending()
 	if q != total && q != total-1 {
 		t.Fatalf("Expected %d or %d, got %d", total, total-1, q)
 	}
@@ -1271,7 +1271,7 @@ func TestSyncSubscriptionPending(t *testing.T) {
 	nc.Flush()
 
 	// Test old way
-	q, _ := sub.QueuedMsgs()
+	q, _, _ := sub.Pending()
 	if q != total && q != total-1 {
 		t.Fatalf("Expected %d or %d, got %d", total, total-1, q)
 	}
@@ -1319,10 +1319,10 @@ func TestSetPendingLimits(t *testing.T) {
 	// Check for invalid values
 	invalid := func() error {
 		if err := sub.SetPendingLimits(0, 1); err == nil {
-			return fmt.Errorf("Setting limit with 0 should fail")
+			return errors.New("Setting limit with 0 should fail")
 		}
 		if err := sub.SetPendingLimits(1, 0); err == nil {
-			return fmt.Errorf("Setting limit with 0 should fail")
+			return errors.New("Setting limit with 0 should fail")
 		}
 		return nil
 	}
diff --git a/test/ws_test.go b/test/ws_test.go
index 15707d8..ee13b33 100644
--- a/test/ws_test.go
+++ b/test/ws_test.go
@@ -17,6 +17,7 @@ import (
 	"bytes"
 	"crypto/tls"
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"math/rand"
 	"net"
@@ -524,7 +525,7 @@ func TestWSStress(t *testing.T) {
 					return
 				}
 				if !bytes.Equal(m.Data[4:4+ps], mainPayload[:ps]) {
-					pushErr(fmt.Errorf("invalid content"))
+					pushErr(errors.New("invalid content"))
 					return
 				}
 				if atomic.AddInt64(&count, 1) == totalRecv {