Repository: nats-io/nats.go
Pull request #974 (2022-05-10 10:06:08)

Add WS ProxyPath option
This is a PR to fix the issue described here: https://github.com/nats-io/nats.go/issues/859


Patch:
diff --git a/.words b/.words
index 24be7f6..63fba00 100644
--- a/.words
+++ b/.words
@@ -1,9 +1,26 @@
 1
-
+## ^^^ size estimate, just needs to be non-zero
+##
+## The .words file is used by gospel (v1.2+), which wraps the Hunspell libraries but populates the dictionary with identifiers from the Go source.
+## <https://github.com/kortschak/gospel>
+##
+## Comment lines are not actually parsed as comments, try to be careful ... but in practice they seem to work?
+##
+## We assume en_US hunspell dictionaries are installed and used.  The /AFFIXRULES are defined in en_US.aff (eg: /usr/share/hunspell/en_US.aff)
+##
+## words which are in the base dictionary can't have extra affix rules added to them, so we have to start with the affixed variant we want to add.
+##   thus creds rather than cred/S and so on
+## So we can't use receive/DRSZGBU, adding 'U', to allow unreceive and variants, we have to use unreceive as the stem.
+## We can't define our own affix or compound rules, to capture rfc\d{3,} or 0x[0-9A-Fa-f]{2}
+
+## People involved who are referenced in todo/fixmes
 derek
 dlc
 ivan
 
+## Legitimate spellings in non-US English dialects;
+## regular-if-rarer words just missing from the dictionary;
+## variants of words not covered by hunspell en_US rules.
 acknowledgement/SM
 arity
 deduplication/S
@@ -14,12 +31,31 @@ observable/S
 redelivery/S
 retransmitting
 retry/SB
-
+unmarshal/SDG
+
+# I think that retry, being added as a symbol, is precluding the re-addition here with affix rules,
+# so "retry/SB" above is ignored
+retries
+retryable
+
+## Things gospel doesn't pick up, but doesn't yet; I've filed <https://github.com/kortschak/gospel/issues/9>
+## Eg, plurals of non-collection types, or wire-format encodings in a struct field's tag
+AsyncSubscriptions
+ChanSubscriptions
+PubAckFutures
+SubOpts
+SyncSubscriptions
+no_wait
+
+## Conceptual nouns not actually in the source, describing state
 SlowConsumer
 
+## Symbols from elsewhere referred to in comments but not referenced in the code, so not currently surfaced by gospel as acceptable
 AppendInt
 ReadMIMEHeader
 
+## The rest
+
 clientProtoZero
 jetstream
 v1
@@ -47,6 +83,8 @@ tm
 todo
 unsub/S
 
+## The spelling tokenizer doesn't take "permessage-deflate" as allowing for ... "permessage-deflate",
+## which is an RFC7692 registered extension.  We have to explicitly list "permessage".
 permessage
 permessage-deflate
 urlA
diff --git a/.words.readme b/.words.readme
deleted file mode 100644
index 9d9f5cb..0000000
--- a/.words.readme
+++ /dev/null
@@ -1,25 +0,0 @@
-The .words file is used by gospel (v1.2+), which wraps the Hunspell libraries
-but populates the dictionary with identifiers from the Go source.
-
-<https://github.com/kortschak/gospel>
-
-Alas, no comments are allowed in the .words file and newer versions of gospel
-error out on seeing them.  This is really a hunspell restriction.
-
-We assume en_US hunspell dictionaries are installed and used.
-The /AFFIXRULES are defined in en_US.aff (eg: /usr/share/hunspell/en_US.aff)
-Invoke `hunspell -D` to see the actual locations.
-
-Words which are in the base dictionary can't have extra affix rules added to
-them, so we have to start with the affixed variant we want to add.
-Thus `creds` rather than `cred/S` and so on.
-
-So we can't use receive/DRSZGBU, adding 'U', to allow unreceive and variants,
-we have to use unreceive as the stem.
-
-We can't define our own affix or compound rules,
-to capture rfc\d{3,} or 0x[0-9A-Fa-f]{2}
-
-The spelling tokenizer doesn't take "permessage-deflate" as allowing for ...
-"permessage-deflate", which is an RFC7692 registered extension for websockets.
-We have to explicitly list "permessage".
diff --git a/README.md b/README.md
index 2e24aa1..60e0f22 100644
--- a/README.md
+++ b/README.md
@@ -29,7 +29,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.15.0
+go get github.com/nats-io/nats.go/@v1.14.0
 
 # For latest NATS Server, add /v2 at the end
 go get github.com/nats-io/nats-server/v2
diff --git a/js.go b/js.go
index 1259743..7001220 100644
--- a/js.go
+++ b/js.go
@@ -851,7 +851,7 @@ func RetryWait(dur time.Duration) PubOpt {
 	})
 }
 
-// RetryAttempts sets the retry number of attempts when ErrNoResponders is encountered.
+// RetryAttempts sets the retry number of attemopts when ErrNoResponders is encountered.
 func RetryAttempts(num int) PubOpt {
 	return pubOptFn(func(opts *pubOpts) error {
 		opts.rnum = num
@@ -2414,20 +2414,12 @@ func PullMaxWaiting(n int) SubOpt {
 	})
 }
 
-var (
-	// errNoMessages is an error that a Fetch request using no_wait can receive to signal
-	// that there are no more messages available.
-	errNoMessages = errors.New("nats: no messages")
-
-	// errRequestsPending is an error that represents a sub.Fetch requests that was using
-	// no_wait and expires time got discarded by the server.
-	errRequestsPending = errors.New("nats: requests pending")
-)
+var errNoMessages = errors.New("nats: no messages")
 
 // Returns if the given message is a user message or not, and if
 // `checkSts` is true, returns appropriate error based on the
 // content of the status (404, etc..)
-func checkMsg(msg *Msg, checkSts, isNoWait bool) (usrMsg bool, err error) {
+func checkMsg(msg *Msg, checkSts bool) (usrMsg bool, err error) {
 	// Assume user message
 	usrMsg = true
 
@@ -2456,17 +2448,11 @@ func checkMsg(msg *Msg, checkSts, isNoWait bool) (usrMsg bool, err error) {
 		// 404 indicates that there are no messages.
 		err = errNoMessages
 	case reqTimeoutSts:
-		// In case of a fetch request with no wait request and expires time,
-		// need to skip 408 errors and retry.
-		if isNoWait {
-			err = errRequestsPending
-		} else {
-			// Older servers may send a 408 when a request in the server was expired
-			// and interest is still found, which will be the case for our
-			// implementation. Regardless, ignore 408 errors until receiving at least
-			// one message when making requests without no_wait.
-			err = ErrTimeout
-		}
+		// Older servers may send a 408 when a request in the server was expired
+		// and interest is still found, which will be the case for our
+		// implementation. Regardless, ignore 408 errors until receiving at least
+		// one message.
+		err = ErrTimeout
 	default:
 		err = fmt.Errorf("nats: %s", msg.Header.Get(descrHdr))
 	}
@@ -2581,7 +2567,7 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 		// or status message, however, we don't care about values of status
 		// messages at this point in the Fetch() call, so checkMsg can't
 		// return an error.
-		if usrMsg, _ := checkMsg(msg, false, false); usrMsg {
+		if usrMsg, _ := checkMsg(msg, false); usrMsg {
 			msgs = append(msgs, msg)
 		}
 	}
@@ -2624,11 +2610,11 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 			if err == nil {
 				var usrMsg bool
 
-				usrMsg, err = checkMsg(msg, true, noWait)
+				usrMsg, err = checkMsg(msg, true)
 				if err == nil && usrMsg {
 					msgs = append(msgs, msg)
-				} else if noWait && (err == errNoMessages || err == errRequestsPending) && len(msgs) == 0 {
-					// If we have a 404/408 for our "no_wait" request and have
+				} else if noWait && (err == errNoMessages) && len(msgs) == 0 {
+					// If we have a 404 for our "no_wait" request and have
 					// not collected any message, then resend request to
 					// wait this time.
 					noWait = false
diff --git a/nats.go b/nats.go
index d3949c7..559cd26 100644
--- a/nats.go
+++ b/nats.go
@@ -48,7 +48,7 @@ import (
 
 // Default Constants
 const (
-	Version                   = "1.15.0"
+	Version                   = "1.14.0"
 	DefaultURL                = "nats://127.0.0.1:4222"
 	DefaultPort               = 4222
 	DefaultMaxReconnect       = 60
@@ -467,6 +467,10 @@ type Options struct {
 	// supports compression. If the server does too, then data will be compressed.
 	Compression bool
 
+	// For websocket connections, adds a path to connections url.
+	// This is useful when connecting to NATS behind a proxy.
+	ProxyPath string
+
 	// InboxPrefix allows the default _INBOX prefix to be customized
 	InboxPrefix string
 }
@@ -1147,6 +1151,15 @@ func Compression(enabled bool) Option {
 	}
 }
 
+// ProxyPath is an option for websocket connections that adds a path to connections url.
+// This is useful when connecting to NATS behind a proxy.
+func ProxyPath(path string) Option {
+	return func(o *Options) error {
+		o.ProxyPath = path
+		return nil
+	}
+}
+
 // CustomInboxPrefix configures the request + reply inbox prefix
 func CustomInboxPrefix(p string) Option {
 	return func(o *Options) error {
diff --git a/ws.go b/ws.go
index 4d24454..a76e582 100644
--- a/ws.go
+++ b/ws.go
@@ -570,6 +570,15 @@ func (nc *Conn) wsInitHandshake(u *url.URL) error {
 		scheme = "https"
 	}
 	ustr := fmt.Sprintf("%s://%s", scheme, u.Host)
+
+	if nc.Opts.ProxyPath != "" {
+		proxyPath := nc.Opts.ProxyPath
+		if !strings.HasPrefix(proxyPath, "/") {
+			proxyPath = "/" + proxyPath
+		}
+		ustr += proxyPath
+	}
+
 	u, err = url.Parse(ustr)
 	if err != nil {
 		return err


Test patch:
diff --git a/go_test.mod b/go_test.mod
index 68d2ac4..1673fdb 100644
--- a/go_test.mod
+++ b/go_test.mod
@@ -4,7 +4,7 @@ go 1.17
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/nats-io/nats-server/v2 v2.8.2
+	github.com/nats-io/nats-server/v2 v2.7.5-0.20220416010054-1b0a5d9e9b36
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 	google.golang.org/protobuf v1.23.0
@@ -14,7 +14,7 @@ require (
 	github.com/klauspost/compress v1.14.4 // indirect
 	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a // indirect
-	golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd // indirect
+	golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce // indirect
 	golang.org/x/sys v0.0.0-20220111092808-5a964db01320 // indirect
 	golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 // indirect
 )
diff --git a/go_test.sum b/go_test.sum
index 8b79f47..31be051 100644
--- a/go_test.sum
+++ b/go_test.sum
@@ -15,16 +15,16 @@ github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a h1:lem6QCvxR0Y28gth9P+wV2K/zYUUAkJ+55U8cpS0p5I=
 github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
-github.com/nats-io/nats-server/v2 v2.8.2 h1:5m1VytMEbZx0YINvKY+X2gXdLNwP43uLXnFRwz8j8KE=
-github.com/nats-io/nats-server/v2 v2.8.2/go.mod h1:vIdpKz3OG+DCg4q/xVPdXHoztEyKDWRtykQ4N7hd7C4=
+github.com/nats-io/nats-server/v2 v2.7.5-0.20220416010054-1b0a5d9e9b36 h1:jOdTqOcfwlsbTMz45x+bsVEPs7P8jPjMOrlu1o4de3k=
+github.com/nats-io/nats-server/v2 v2.7.5-0.20220416010054-1b0a5d9e9b36/go.mod h1:5vic7C58BFEVltiZhs7Kq81q2WcEPhJPsmNv1FOrdv0=
 github.com/nats-io/nats.go v1.14.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd h1:XcWmESyNjXJMLahc3mqVQJcgSTDxFxhETVlfk9uGc38=
-golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce h1:Roh6XWxHFKrPgC/EQhVubSAGQ6Ozk6IdxHSzt1mR0EI=
+golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
diff --git a/test/js_test.go b/test/js_test.go
index 01d6b8d..132bc2a 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -4262,103 +4262,97 @@ func testJetStreamMirror_Source(t *testing.T, nodes ...*jsServer) {
 		}
 	})
 
-	// Commenting out this test until we figure out what was the intent.
-	// Since v2.8.0, this test would fail with a "detected cycle" error,
-	// I guess because "m1" already sources "origin", so creating a
-	// stream with both as a source is bad.
-	/*
-		t.Run("create sourced stream from origin", func(t *testing.T) {
-			sources := make([]*nats.StreamSource, 0)
-			sources = append(sources, &nats.StreamSource{Name: "origin"})
-			sources = append(sources, &nats.StreamSource{Name: "m1"})
-			streamName := "s2"
-			_, err = js.AddStream(&nats.StreamConfig{
-				Name:     streamName,
-				Sources:  sources,
-				Storage:  nats.FileStorage,
-				Replicas: 1,
-			})
-			if err != nil {
-				t.Fatalf("Unexpected error creating stream: %v", err)
-			}
+	t.Run("create sourced stream from origin", func(t *testing.T) {
+		sources := make([]*nats.StreamSource, 0)
+		sources = append(sources, &nats.StreamSource{Name: "origin"})
+		sources = append(sources, &nats.StreamSource{Name: "m1"})
+		streamName := "s2"
+		_, err = js.AddStream(&nats.StreamConfig{
+			Name:     streamName,
+			Sources:  sources,
+			Storage:  nats.FileStorage,
+			Replicas: 1,
+		})
+		if err != nil {
+			t.Fatalf("Unexpected error creating stream: %v", err)
+		}
 
-			msgs := make([]*nats.RawStreamMsg, 0)
+		msgs := make([]*nats.RawStreamMsg, 0)
 
-			// Stored message sequences start at 1
-			startSequence := 1
-			expectedTotal := totalMsgs * 2
+		// Stored message sequences start at 1
+		startSequence := 1
+		expectedTotal := totalMsgs * 2
 
-		GetNextMsg:
-			for i := startSequence; i < expectedTotal+1; i++ {
-				var (
-					err     error
-					seq     = uint64(i)
-					msg     *nats.RawStreamMsg
-					timeout = time.Now().Add(5 * time.Second)
-				)
+	GetNextMsg:
+		for i := startSequence; i < expectedTotal+1; i++ {
+			var (
+				err     error
+				seq     = uint64(i)
+				msg     *nats.RawStreamMsg
+				timeout = time.Now().Add(5 * time.Second)
+			)
 
-			Retry:
-				for time.Now().Before(timeout) {
-					msg, err = js.GetMsg(streamName, seq)
-					if err != nil {
-						time.Sleep(100 * time.Millisecond)
-						continue Retry
-					}
-					msgs = append(msgs, msg)
-					continue GetNextMsg
-				}
+		Retry:
+			for time.Now().Before(timeout) {
+				msg, err = js.GetMsg(streamName, seq)
 				if err != nil {
-					t.Fatalf("Unexpected error fetching seq=%v: %v", seq, err)
+					time.Sleep(100 * time.Millisecond)
+					continue Retry
 				}
+				msgs = append(msgs, msg)
+				continue GetNextMsg
 			}
-
-			got := len(msgs)
-			if got < expectedTotal {
-				t.Errorf("Expected %v, got: %v", expectedTotal, got)
-			}
-
-			si, err := js.StreamInfo(streamName)
 			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			got = int(si.State.Msgs)
-			if got != expectedTotal {
-				t.Errorf("Expected %v, got: %v", expectedTotal, got)
+				t.Fatalf("Unexpected error fetching seq=%v: %v", seq, err)
 			}
+		}
+
+		got := len(msgs)
+		if got < expectedTotal {
+			t.Errorf("Expected %v, got: %v", expectedTotal, got)
+		}
+
+		si, err := js.StreamInfo(streamName)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		got = int(si.State.Msgs)
+		if got != expectedTotal {
+			t.Errorf("Expected %v, got: %v", expectedTotal, got)
+		}
 
-			got = len(si.Sources)
-			expected := 2
-			if got != expected {
-				t.Errorf("Expected %v, got: %v", expected, got)
+		got = len(si.Sources)
+		expected := 2
+		if got != expected {
+			t.Errorf("Expected %v, got: %v", expected, got)
+		}
+
+		t.Run("consume from sourced stream", func(t *testing.T) {
+			sub, err := js.SubscribeSync("origin", nats.BindStream(streamName))
+			if err != nil {
+				t.Fatal(err)
 			}
 
-			t.Run("consume from sourced stream", func(t *testing.T) {
-				sub, err := js.SubscribeSync("origin", nats.BindStream(streamName))
+			mmsgs := make([]*nats.Msg, 0)
+			for i := 0; i < totalMsgs; i++ {
+				msg, err := sub.NextMsg(2 * time.Second)
 				if err != nil {
-					t.Fatal(err)
+					t.Error(err)
 				}
-
-				mmsgs := make([]*nats.Msg, 0)
-				for i := 0; i < totalMsgs; i++ {
-					msg, err := sub.NextMsg(2 * time.Second)
-					if err != nil {
-						t.Error(err)
-					}
-					meta, err := msg.Metadata()
-					if err != nil {
-						t.Error(err)
-					}
-					if meta.Stream != streamName {
-						t.Errorf("Expected m1, got: %v", meta.Stream)
-					}
-					mmsgs = append(mmsgs, msg)
+				meta, err := msg.Metadata()
+				if err != nil {
+					t.Error(err)
 				}
-				if len(mmsgs) != totalMsgs {
-					t.Errorf("Expected to consume %v msgs, got: %v", totalMsgs, len(mmsgs))
+				if meta.Stream != streamName {
+					t.Errorf("Expected m1, got: %v", meta.Stream)
 				}
-			})
+				mmsgs = append(mmsgs, msg)
+			}
+			if len(mmsgs) != totalMsgs {
+				t.Errorf("Expected to consume %v msgs, got: %v", totalMsgs, len(mmsgs))
+			}
 		})
-	*/
+	})
 }
 
 func TestJetStream_ClusterMultipleSubscribe(t *testing.T) {
@@ -4388,15 +4382,6 @@ func TestJetStream_ClusterMultipleSubscribe(t *testing.T) {
 				}
 				withJSClusterAndStream(t, name, n, stream, testJetStream_ClusterMultiplePullSubscribe)
 			})
-
-			t.Run(fmt.Sprintf("psub n=%d r=%d multi fetch", n, r), func(t *testing.T) {
-				name := fmt.Sprintf("PFSUBN%d%d", n, r)
-				stream := &nats.StreamConfig{
-					Name:     name,
-					Replicas: n,
-				}
-				withJSClusterAndStream(t, name, n, stream, testJetStream_ClusterMultipleFetchPullSubscribe)
-			})
 		}
 	}
 }
@@ -4558,139 +4543,6 @@ func testJetStream_ClusterMultiplePullSubscribe(t *testing.T, subject string, sr
 	}
 }
 
-func testJetStream_ClusterMultipleFetchPullSubscribe(t *testing.T, subject string, srvs ...*jsServer) {
-	srv := srvs[0]
-	nc, js := jsClient(t, srv.Server)
-	defer nc.Close()
-
-	var wg sync.WaitGroup
-	ctx, done := context.WithTimeout(context.Background(), 5*time.Second)
-	defer done()
-
-	// Setup a number of subscriptions with different inboxes that will be
-	// fetching the messages in parallel.
-	nsubs := 4
-	subs := make([]*nats.Subscription, nsubs)
-	errCh := make(chan error, nsubs)
-	var queues sync.Map
-	for i := 0; i < nsubs; i++ {
-		wg.Add(1)
-		go func(n int) {
-			defer wg.Done()
-			var sub *nats.Subscription
-			var err error
-			sub, err = js.PullSubscribe(subject, "shared")
-			if err != nil {
-				errCh <- err
-			} else {
-				subs[n] = sub
-				queues.Store(sub.Subject, make([]*nats.Msg, 0))
-			}
-		}(i)
-	}
-
-	// Publishing of messages happens after the subscriptions are ready.
-	// The subscribers will be fetching messages while these are being
-	// produced so sometimes there are not going to be messages available.
-	wg.Wait()
-	var (
-		total     uint64 = 100
-		delivered uint64
-		batchSize = 2
-	)
-	go func() {
-		for i := 0; i < int(total); i++ {
-			js.Publish(subject, []byte(fmt.Sprintf("n:%v", i)))
-			time.Sleep(1 * time.Millisecond)
-		}
-	}()
-
-	ctx2, done2 := context.WithTimeout(ctx, 3*time.Second)
-	defer done2()
-
-	for _, psub := range subs {
-		if psub == nil {
-			continue
-		}
-		sub := psub
-		subject := sub.Subject
-		v, _ := queues.Load(sub.Subject)
-		queue := v.([]*nats.Msg)
-		go func() {
-			for {
-				select {
-				case <-ctx2.Done():
-					return
-				default:
-				}
-
-				if current := atomic.LoadUint64(&delivered); current >= total {
-					done2()
-					return
-				}
-
-				// Wait until all messages have been consumed.
-				for attempt := 0; attempt < 4; attempt++ {
-					recvd, err := sub.Fetch(batchSize, nats.MaxWait(1*time.Second))
-					if err != nil {
-						if err == nats.ErrConnectionClosed {
-							return
-						}
-						current := atomic.LoadUint64(&delivered)
-						if current >= total {
-							done2()
-							return
-						} else {
-							t.Logf("WARN: Timeout waiting for next message: %v", err)
-						}
-						continue
-					}
-					for _, msg := range recvd {
-						queue = append(queue, msg)
-						queues.Store(subject, queue)
-					}
-					atomic.AddUint64(&delivered, uint64(len(recvd)))
-					break
-				}
-			}
-		}()
-	}
-
-	// Wait until context is canceled after receiving all messages.
-	<-ctx2.Done()
-
-	if delivered < total {
-		t.Fatalf("Expected %v, got: %v", total, delivered)
-	}
-
-	select {
-	case <-ctx.Done():
-	case err := <-errCh:
-		if err != nil {
-			t.Fatalf("Unexpected error with multiple pull subscribers: %v", err)
-		}
-	}
-
-	var (
-		gotNoMessages bool
-		count         = 0
-	)
-	queues.Range(func(k, v interface{}) bool {
-		msgs := v.([]*nats.Msg)
-		count += len(msgs)
-
-		if len(msgs) == 0 {
-			gotNoMessages = true
-			return false
-		}
-		return true
-	})
-
-	if gotNoMessages {
-		t.Error("Expected all pull subscribers to receive some messages")
-	}
-}
-
 func TestJetStream_ClusterReconnect(t *testing.T) {
 	t.Skip("This test need to be revisited")
 	n := 3
@@ -5284,55 +5136,44 @@ func testJetStreamFetchOptions(t *testing.T, srvs ...*jsServer) {
 		}
 	})
 
-	t.Run("max waiting exceeded", func(t *testing.T) {
+	t.Run("max waiting timeout", func(t *testing.T) {
 		defer js.PurgeStream(subject)
 
-		_, err := js.AddConsumer(subject, &nats.ConsumerConfig{
-			Durable:    "max-waiting",
-			MaxWaiting: 2,
-			AckPolicy:  nats.AckExplicitPolicy,
-		})
+		expected := 10
+		sendMsgs(t, expected)
+
+		sub, err := js.PullSubscribe(subject, "max-waiting")
 		if err != nil {
 			t.Fatal(err)
 		}
+		defer sub.Unsubscribe()
 
+		// Poll more than the default max of waiting/inflight pull requests,
+		// so that We will get only 408 timeout errors.
+		errCh := make(chan error, 1024)
+		defer close(errCh)
 		var wg sync.WaitGroup
-		wg.Add(2)
-		for i := 0; i < 2; i++ {
+		for i := 0; i < 1024; i++ {
+			wg.Add(1)
+
 			go func() {
+				_, err := sub.Fetch(1, nats.MaxWait(500*time.Millisecond))
 				defer wg.Done()
-
-				sub, err := js.PullSubscribe(subject, "max-waiting")
 				if err != nil {
-					return
+					errCh <- err
 				}
-				sub.Fetch(1, nats.MaxWait(time.Second))
 			}()
 		}
+		wg.Wait()
 
-		// Give time to those 2 above to fill the MaxWaiting
-		checkFor(t, time.Second, 15*time.Millisecond, func() error {
-			ci, err := js.ConsumerInfo(subject, "max-waiting")
-			if err != nil {
-				return err
-			}
-			if n := ci.NumWaiting; n != 2 {
-				return fmt.Errorf("NumWaiting should be 2, was %v", n)
+		select {
+		case <-time.After(1 * time.Second):
+			t.Fatal("Expected RequestTimeout (408) error due to many inflight pulls")
+		case err := <-errCh:
+			if err != nil && (err.Error() != `nats: Request Timeout` && err != nats.ErrTimeout) {
+				t.Errorf("Expected request timeout fetching next message, got: %+v", err)
 			}
-			return nil
-		})
-
-		// Now this request should get a 409. Currently, we do not re-fetch
-		// on that error, so would be visible in the error returned by Fetch()
-		sub, err := js.PullSubscribe(subject, "max-waiting")
-		if err != nil {
-			t.Fatal(err)
 		}
-		_, err = sub.Fetch(1, nats.MaxWait(time.Second))
-		if err == nil || !strings.Contains(err.Error(), "MaxWaiting") {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		wg.Wait()
 	})
 
 	t.Run("no wait", func(t *testing.T) {
diff --git a/ws_test.go b/ws_test.go
index 84229b7..4dec114 100644
--- a/ws_test.go
+++ b/ws_test.go
@@ -16,11 +16,14 @@ package nats
 import (
 	"bytes"
 	"compress/flate"
+	"context"
 	"crypto/tls"
 	"encoding/binary"
 	"fmt"
 	"io"
 	"math/rand"
+	"net"
+	"net/http"
 	"reflect"
 	"runtime"
 	"strings"
@@ -1108,3 +1111,50 @@ func TestWSNoDeadlockOnAuthFailure(t *testing.T) {
 
 	tm.Stop()
 }
+
+func TestWSProxyPath(t *testing.T) {
+	const proxyPath = "proxy1"
+
+	// Listen to a random port
+	l, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatalf("Error in listen: %v", err)
+	}
+	defer l.Close()
+
+	proxyPort := l.Addr().(*net.TCPAddr).Port
+
+	ch := make(chan struct{}, 1)
+	proxySrv := &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.URL.Path == "/"+proxyPath {
+				ch <- struct{}{}
+			}
+		}),
+	}
+	defer proxySrv.Shutdown(context.Background())
+	go proxySrv.Serve(l)
+
+	for _, test := range []struct {
+		name string
+		path string
+	}{
+		{"without slash", proxyPath},
+		{"with slash", "/" + proxyPath},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			url := fmt.Sprintf("ws://127.0.0.1:%d", proxyPort)
+			nc, err := Connect(url, ProxyPath(test.path))
+			if err == nil {
+				nc.Close()
+				t.Fatal("Did not expect to connect")
+			}
+			select {
+			case <-ch:
+				// OK:
+			case <-time.After(time.Second):
+				t.Fatal("Proxy was not reached")
+			}
+		})
+	}
+}