Upgrade NATS client

For compatibility with newer NATS streaming version https://github.com/openfaas/faas-netes/pull/819 Signed-off-by: Alex Ellis (OpenFaaS Ltd) <alexellis2@gmail.com>
2025-06-21 00:06:38 +00:00 · 2021-07-26 16:55:47 +01:00
parent 33c07e1ae8
commit 06a51373e2
390 changed files with 26182 additions and 16357 deletions
--- a/gateway/vendor/github.com/nats-io/nats.go/.travis.yml
+++ b/gateway/vendor/github.com/nats-io/nats.go/.travis.yml
@ -1,23 +1,19 @@
 language: go
 go:
- 1.14.x
- 1.13.x
-env:
- GO111MODULE=off
+- 1.16.x
+- 1.15.x
 go_import_path: github.com/nats-io/nats.go
 install:
 - go get -t ./...
- go get github.com/nats-io/nats-server
 - go get github.com/mattn/goveralls
 - go get github.com/wadey/gocovmerge
 - go get -u honnef.co/go/tools/cmd/staticcheck
 - go get -u github.com/client9/misspell/cmd/misspell
 before_script:
 - $(exit $(go fmt ./... | wc -l))
- go vet ./...
+- go vet -modfile=go_test.mod ./...
 - find . -type f -name "*.go" | xargs misspell -error -locale US
 - staticcheck ./...
 script:
- go test -i -race ./...
- go test -v -run=TestNoRace -p=1 ./...
- if [[ "$TRAVIS_GO_VERSION" =~ 1.14 ]]; then ./scripts/cov.sh TRAVIS; else go test -race -v -p=1 ./... --failfast; fi
+- go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.16 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast; fi
--- a/gateway/vendor/github.com/nats-io/nats.go/MAINTAINERS.md
+++ b/gateway/vendor/github.com/nats-io/nats.go/MAINTAINERS.md
@ -2,9 +2,7 @@

 Maintainership is on a per project basis.

-### Core-maintainers
+### Maintainers
  - Derek Collison <derek@nats.io> [@derekcollison](https://github.com/derekcollison)
  - Ivan Kozlovic <ivan@nats.io> [@kozlovic](https://github.com/kozlovic)
-  
-### Maintainers
-  - Waldemar Quevedo <wally@nats.io> [@wallyqs](https://github.com/wallyqs)
+  - Waldemar Quevedo <wally@nats.io> [@wallyqs](https://github.com/wallyqs)
--- a/gateway/vendor/github.com/nats-io/nats.go/README.md
+++ b/gateway/vendor/github.com/nats-io/nats.go/README.md
@ -3,7 +3,8 @@ A [Go](http://golang.org) client for the [NATS messaging system](https://nats.io

 [![License Apache 2](https://img.shields.io/badge/License-Apache2-blue.svg)](https://www.apache.org/licenses/LICENSE-2.0)
 [![FOSSA Status](https://app.fossa.io/api/projects/git%2Bgithub.com%2Fnats-io%2Fgo-nats.svg?type=shield)](https://app.fossa.io/projects/git%2Bgithub.com%2Fnats-io%2Fgo-nats?ref=badge_shield)
-[![Go Report Card](https://goreportcard.com/badge/github.com/nats-io/nats.go)](https://goreportcard.com/report/github.com/nats-io/nats.go) [![Build Status](https://travis-ci.org/nats-io/nats.go.svg?branch=master)](http://travis-ci.org/nats-io/nats.go) [![GoDoc](https://godoc.org/github.com/nats-io/nats.go?status.svg)](http://godoc.org/github.com/nats-io/nats.go) [![Coverage Status](https://coveralls.io/repos/nats-io/nats.go/badge.svg?branch=master)](https://coveralls.io/r/nats-io/nats.go?branch=master)
+[![Go Report Card](https://goreportcard.com/badge/github.com/nats-io/nats.go)](https://goreportcard.com/report/github.com/nats-io/nats.go) [![Build Status](https://travis-ci.com/nats-io/nats.go.svg?branch=master)](http://travis-ci.com/nats-io/nats.go) [![GoDoc](https://img.shields.io/badge/GoDoc-reference-007d9c)](https://pkg.go.dev/github.com/nats-io/nats.go)
+ [![Coverage Status](https://coveralls.io/repos/nats-io/nats.go/badge.svg?branch=master)](https://coveralls.io/r/nats-io/nats.go?branch=master)

 ## Installation

@ -20,7 +21,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.10.0
+go get github.com/nats-io/nats.go/@v1.11.0

 # For latest NATS Server, add /v2 at the end
 go get github.com/nats-io/nats-server/v2
@ -32,7 +33,7 @@ go get github.com/nats-io/nats-server/v2
 ## Basic Usage

 ```go
-import nats "github.com/nats-io/nats.go"
+import "github.com/nats-io/nats.go"

 // Connect to a server
 nc, _ := nats.Connect(nats.DefaultURL)
@ -81,6 +82,85 @@ nc.Drain()
 nc.Close()
 ```

+## JetStream Basic Usage
+
+```go
+import "github.com/nats-io/nats.go"
+
+// Connect to NATS
+nc, _ := nats.Connect(nats.DefaultURL)
+
+// Create JetStream Context
+js, _ := nc.JetStream(nats.PublishAsyncMaxPending(256))
+
+// Simple Stream Publisher
+js.Publish("ORDERS.scratch", []byte("hello"))
+
+// Simple Async Stream Publisher
+for i := 0; i < 500; i++ {
+	js.PublishAsync("ORDERS.scratch", []byte("hello"))
+}
+select {
+case <-js.PublishAsyncComplete():
+case <-time.After(5 * time.Second):
+	fmt.Println("Did not resolve in time")
+}
+
+// Simple Async Ephemeral Consumer
+js.Subscribe("ORDERS.*", func(m *nats.Msg) {
+	fmt.Printf("Received a JetStream message: %s\n", string(m.Data))
+})
+
+// Simple Sync Durable Consumer (optional SubOpts at the end)
+sub, err := js.SubscribeSync("ORDERS.*", nats.Durable("MONITOR"), nats.MaxDeliver(3))
+m, err := sub.NextMsg(timeout)
+
+// Simple Pull Consumer
+sub, err := js.PullSubscribe("ORDERS.*", "MONITOR")
+msgs, err := sub.Fetch(10)
+
+// Unsubscribe
+sub.Unsubscribe()
+
+// Drain
+sub.Drain()
+```
+
+## JetStream Basic Management
+
+```go
+import "github.com/nats-io/nats.go"
+
+// Connect to NATS
+nc, _ := nats.Connect(nats.DefaultURL)
+
+// Create JetStream Context
+js, _ := nc.JetStream()
+
+// Create a Stream
+js.AddStream(&nats.StreamConfig{
+	Name:     "ORDERS",
+	Subjects: []string{"ORDERS.*"},
+})
+
+// Update a Stream
+js.UpdateStream(&nats.StreamConfig{
+	Name:     "ORDERS",
+	MaxBytes: 8,
+})
+
+// Create a Consumer
+js.AddConsumer("ORDERS", &nats.ConsumerConfig{
+	Durable: "MONITOR",
+})
+
+// Delete Consumer
+js.DeleteConsumer("ORDERS", "MONITOR")
+
+// Delete Stream
+js.DeleteStream("ORDERS")
+```
+
 ## Encoded Connections

 ```go
@ -277,13 +357,27 @@ nc.Publish("foo.bar.baz", []byte("Hello World"))
 nc.QueueSubscribe("foo", "job_workers", func(_ *Msg) {
  received += 1;
 })
-
 ```

 ## Advanced Usage

 ```go

+// Normally, the library will return an error when trying to connect and
+// there is no server running. The RetryOnFailedConnect option will set
+// the connection in reconnecting state if it failed to connect right away.
+nc, err := nats.Connect(nats.DefaultURL,
+    nats.RetryOnFailedConnect(true),
+    nats.MaxReconnects(10),
+    nats.ReconnectWait(time.Second),
+    nats.ReconnectHandler(func(_ *nats.Conn) {
+        // Note that this will be invoked for the first asynchronous connect.
+    }))
+if err != nil {
+    // Should not return an error even if it can't connect, but you still
+    // need to check in case there are some configuration errors.
+}
+
 // Flush connection to server, returns when all messages have been processed.
 nc.Flush()
 fmt.Println("All clear!")
--- a/gateway/vendor/github.com/nats-io/nats.go/context.go
+++ b/gateway/vendor/github.com/nats-io/nats.go/context.go
@ -11,9 +11,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-// +build go1.7
-
-// A Go client for the NATS messaging system (https://nats.io).
 package nats

 import (
@ -21,9 +18,33 @@ import (
 	"reflect"
 )

+// RequestMsgWithContext takes a context, a subject and payload
+// in bytes and request expecting a single response.
+func (nc *Conn) RequestMsgWithContext(ctx context.Context, msg *Msg) (*Msg, error) {
+	var hdr []byte
+	var err error
+
+	if len(msg.Header) > 0 {
+		if !nc.info.Headers {
+			return nil, ErrHeadersNotSupported
+		}
+
+		hdr, err = msg.headerBytes()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return nc.requestWithContext(ctx, msg.Subject, hdr, msg.Data)
+}
+
 // RequestWithContext takes a context, a subject and payload
 // in bytes and request expecting a single response.
 func (nc *Conn) RequestWithContext(ctx context.Context, subj string, data []byte) (*Msg, error) {
+	return nc.requestWithContext(ctx, subj, nil, data)
+}
+
+func (nc *Conn) requestWithContext(ctx context.Context, subj string, hdr, data []byte) (*Msg, error) {
 	if ctx == nil {
 		return nil, ErrInvalidContext
 	}
@ -36,49 +57,52 @@ func (nc *Conn) RequestWithContext(ctx context.Context, subj string, data []byte
 		return nil, ctx.Err()
 	}

-	nc.mu.Lock()
+	var m *Msg
+	var err error
+
 	// If user wants the old style.
-	if nc.Opts.UseOldRequestStyle {
-		nc.mu.Unlock()
-		return nc.oldRequestWithContext(ctx, subj, data)
-	}
-
-	mch, token, err := nc.createNewRequestAndSend(subj, data)
-	if err != nil {
-		return nil, err
-	}
-
-	var ok bool
-	var msg *Msg
-
-	select {
-	case msg, ok = <-mch:
-		if !ok {
-			return nil, ErrConnectionClosed
+	if nc.useOldRequestStyle() {
+		m, err = nc.oldRequestWithContext(ctx, subj, hdr, data)
+	} else {
+		mch, token, err := nc.createNewRequestAndSend(subj, hdr, data)
+		if err != nil {
+			return nil, err
 		}
-	case <-ctx.Done():
-		nc.mu.Lock()
-		delete(nc.respMap, token)
-		nc.mu.Unlock()
-		return nil, ctx.Err()
-	}

-	return msg, nil
+		var ok bool
+
+		select {
+		case m, ok = <-mch:
+			if !ok {
+				return nil, ErrConnectionClosed
+			}
+		case <-ctx.Done():
+			nc.mu.Lock()
+			delete(nc.respMap, token)
+			nc.mu.Unlock()
+			return nil, ctx.Err()
+		}
+	}
+	// Check for no responder status.
+	if err == nil && len(m.Data) == 0 && m.Header.Get(statusHdr) == noResponders {
+		m, err = nil, ErrNoResponders
+	}
+	return m, err
 }

 // oldRequestWithContext utilizes inbox and subscription per request.
-func (nc *Conn) oldRequestWithContext(ctx context.Context, subj string, data []byte) (*Msg, error) {
+func (nc *Conn) oldRequestWithContext(ctx context.Context, subj string, hdr, data []byte) (*Msg, error) {
 	inbox := NewInbox()
 	ch := make(chan *Msg, RequestChanLen)

-	s, err := nc.subscribe(inbox, _EMPTY_, nil, ch, true)
+	s, err := nc.subscribe(inbox, _EMPTY_, nil, ch, true, nil)
 	if err != nil {
 		return nil, err
 	}
 	s.AutoUnsubscribe(1)
 	defer s.Unsubscribe()

-	err = nc.PublishRequest(subj, inbox, data)
+	err = nc.publish(subj, inbox, hdr, data)
 	if err != nil {
 		return nil, err
 	}
--- a/gateway/vendor/github.com/nats-io/nats.go/dependencies.md
+++ b/gateway/vendor/github.com/nats-io/nats.go/dependencies.md
@ -0,0 +1,13 @@
+# External Dependencies
+
+This file lists the dependencies used in this repository.
+
+| Dependency | License |
+|-|-|
+| Go | BSD 3-Clause "New" or "Revised" License |
+| github.com/nats-io/nats.go | Apache License 2.0 |
+| github.com/golang/protobuf v1.4.2 | BSD 3-Clause "New" or "Revised" License |
+| github.com/nats-io/nats-server/v2 v2.1.8-0.20201115145023-f61fa8529a0f | Apache License 2.0 |
+| github.com/nats-io/nkeys v0.2.0 | Apache License 2.0 |
+| github.com/nats-io/nuid v1.0.1 | Apache License 2.0 |
+| google.golang.org/protobuf v1.23.0 | BSD 3-Clause License |
--- a/gateway/vendor/github.com/nats-io/nats.go/enc.go
+++ b/gateway/vendor/github.com/nats-io/nats.go/enc.go
@ -93,7 +93,7 @@ func (c *EncodedConn) Publish(subject string, v interface{}) error {
 	if err != nil {
 		return err
 	}
-	return c.Conn.publish(subject, _EMPTY_, b)
+	return c.Conn.publish(subject, _EMPTY_, nil, b)
 }

 // PublishRequest will perform a Publish() expecting a response on the
@ -104,7 +104,7 @@ func (c *EncodedConn) PublishRequest(subject, reply string, v interface{}) error
 	if err != nil {
 		return err
 	}
-	return c.Conn.publish(subject, reply, b)
+	return c.Conn.publish(subject, reply, nil, b)
 }

 // Request will create an Inbox and perform a Request() call
@ -130,7 +130,7 @@ func (c *EncodedConn) Request(subject string, v interface{}, vPtr interface{}, t

 // Handler is a specific callback used for Subscribe. It is generalized to
 // an interface{}, but we will discover its format and arguments at runtime
-// and perform the correct callback, including de-marshaling JSON strings
+// and perform the correct callback, including de-marshaling encoded data
 // back into the appropriate struct based on the signature of the Handler.
 //
 // Handlers are expected to have one of four signatures.
@ -234,7 +234,7 @@ func (c *EncodedConn) subscribe(subject, queue string, cb Handler) (*Subscriptio
 		cbValue.Call(oV)
 	}

-	return c.Conn.subscribe(subject, queue, natsCB, nil, false)
+	return c.Conn.subscribe(subject, queue, natsCB, nil, false, nil)
 }

 // FlushTimeout allows a Flush operation to have an associated timeout.
--- a/gateway/vendor/github.com/nats-io/nats.go/go.mod
+++ b/gateway/vendor/github.com/nats-io/nats.go/go.mod
@ -1,7 +1,8 @@
 module github.com/nats-io/nats.go

+go 1.16
+
 require (
-	github.com/nats-io/jwt v0.3.2
-	github.com/nats-io/nkeys v0.1.4
+	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 )
--- a/gateway/vendor/github.com/nats-io/nats.go/go.sum
+++ b/gateway/vendor/github.com/nats-io/nats.go/go.sum
@ -1,15 +1,11 @@
-github.com/nats-io/jwt v0.3.2 h1:+RB5hMpXUUA2dfxuhBTEkMOrYmM+gKIZYS1KjSostMI=
-github.com/nats-io/jwt v0.3.2/go.mod h1:/euKqTS1ZD+zzjYrY7pseZrTtWQSjujC7xjPc8wL6eU=
-github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
-github.com/nats-io/nkeys v0.1.4 h1:aEsHIssIk6ETN5m2/MD8Y4B2X7FfXrBAUdkyRvbVYzA=
-github.com/nats-io/nkeys v0.1.4/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
+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-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59 h1:3zb4D3T4G8jdExgVU/95+vQXfpEPiMdCaZgmGVxjNHM=
-golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-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/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b h1:wSOdpTq0/eI46Ez/LkDwIsAKA71YP2SRKBODiRWM0as=
+golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
--- a/gateway/vendor/github.com/nats-io/nats.go/go_test.mod
+++ b/gateway/vendor/github.com/nats-io/nats.go/go_test.mod
@ -0,0 +1,11 @@
+module github.com/nats-io/nats.go
+
+go 1.15
+
+require (
+	github.com/golang/protobuf v1.4.2
+	github.com/nats-io/nats-server/v2 v2.2.7-0.20210618192106-93a3720475a4
+	github.com/nats-io/nkeys v0.3.0
+	github.com/nats-io/nuid v1.0.1
+	google.golang.org/protobuf v1.23.0
+)
--- a/gateway/vendor/github.com/nats-io/nats.go/go_test.sum
+++ b/gateway/vendor/github.com/nats-io/nats.go/go_test.sum
@ -0,0 +1,59 @@
+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=
+github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
+github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
+github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
+github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+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.11.12 h1:famVnQVu7QwryBN4jNseQdUKES71ZAOnB6UQQJPZvqk=
+github.com/klauspost/compress v1.11.12/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
+github.com/minio/highwayhash v1.0.1 h1:dZ6IIu8Z14VlC0VpfKofAhCy74wu/Qb5gcn52yWoz/0=
+github.com/minio/highwayhash v1.0.1/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.0.2 h1:ejVCLO8gu6/4bOKIHQpmB5UhhUJfAQw55yvLWpfmKjI=
+github.com/nats-io/jwt/v2 v2.0.2/go.mod h1:VRP+deawSXyhNjXmxPCHskrR6Mq50BqpEI5SEcNiGlY=
+github.com/nats-io/nats-server/v2 v2.2.6 h1:FPK9wWx9pagxcw14s8W9rlfzfyHm61uNLnJyybZbn48=
+github.com/nats-io/nats-server/v2 v2.2.6/go.mod h1:sEnFaxqe09cDmfMgACxZbziXnhQFhwk+aKkZjBBRYrI=
+github.com/nats-io/nats-server/v2 v2.2.7-0.20210615172038-0069f752b61b h1:hy5rgG4Hur55cWBKxD/VbkjaRYYAxo5Ayk9AxGJcHTs=
+github.com/nats-io/nats-server/v2 v2.2.7-0.20210615172038-0069f752b61b/go.mod h1:hBgcnXvNESvh65J1nMtxaHHsaUxSmteZXCH1JLTuvfg=
+github.com/nats-io/nats-server/v2 v2.2.7-0.20210618192106-93a3720475a4 h1:8QM5O7j1a9SdEPzzpQj7daRu4fi/sxfXRxfcKGa5Dr0=
+github.com/nats-io/nats-server/v2 v2.2.7-0.20210618192106-93a3720475a4/go.mod h1:hBgcnXvNESvh65J1nMtxaHHsaUxSmteZXCH1JLTuvfg=
+github.com/nats-io/nats.go v1.11.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
+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-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.0.0-20210314154223-e6e6c4f2bb5b h1:wSOdpTq0/eI46Ez/LkDwIsAKA71YP2SRKBODiRWM0as=
+golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
+golang.org/x/crypto v0.0.0-20210505212654-3497b51f5e64 h1:QuAh/1Gwc0d+u9walMU1NqzhRemNegsv5esp2ALQIY4=
+golang.org/x/crypto v0.0.0-20210505212654-3497b51f5e64/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-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.0.0-20201119102817-f84b799fce68 h1:nxC68pudNYkKU6jWhgrqdreuFiOQWj1Fs7T3VrH4Pjw=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1 h1:NusfzzA6yGQ+ua51ck7E3omNUX/JuqbFSaRGqU8CcLI=
+golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
+google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
+google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
+google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
+google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
+google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
+google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
--- a/gateway/vendor/github.com/nats-io/nats.go/js.go
+++ b/gateway/vendor/github.com/nats-io/nats.go/js.go
--- a/gateway/vendor/github.com/nats-io/nats.go/jsm.go
+++ b/gateway/vendor/github.com/nats-io/nats.go/jsm.go
--- a/gateway/vendor/github.com/nats-io/nats.go/nats.go
+++ b/gateway/vendor/github.com/nats-io/nats.go/nats.go
--- a/gateway/vendor/github.com/nats-io/nats.go/netchan.go
+++ b/gateway/vendor/github.com/nats-io/nats.go/netchan.go
@ -107,5 +107,5 @@ func (c *EncodedConn) bindRecvChan(subject, queue string, channel interface{}) (
 		chVal.Send(oPtr)
 	}

-	return c.Conn.subscribe(subject, queue, cb, nil, false)
+	return c.Conn.subscribe(subject, queue, cb, nil, false, nil)
 }
--- a/gateway/vendor/github.com/nats-io/nats.go/parser.go
+++ b/gateway/vendor/github.com/nats-io/nats.go/parser.go
@ -1,4 +1,4 @@
-// Copyright 2012-2018 The NATS Authors
+// Copyright 2012-2020 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
@ -21,6 +21,7 @@ type msgArg struct {
 	subject []byte
 	reply   []byte
 	sid     int64
+	hdr     int
 	size    int
 }

@ -30,6 +31,7 @@ type parseState struct {
 	state   int
 	as      int
 	drop    int
+	hdr     int
 	ma      msgArg
 	argBuf  []byte
 	msgBuf  []byte
@ -54,6 +56,7 @@ const (
 	MSG_ARG
 	MSG_PAYLOAD
 	MSG_END
+	OP_H
 	OP_P
 	OP_PI
 	OP_PIN
@ -83,6 +86,12 @@ func (nc *Conn) parse(buf []byte) error {
 			switch b {
 			case 'M', 'm':
 				nc.ps.state = OP_M
+				nc.ps.hdr = -1
+				nc.ps.ma.hdr = -1
+			case 'H', 'h':
+				nc.ps.state = OP_H
+				nc.ps.hdr = 0
+				nc.ps.ma.hdr = 0
 			case 'P', 'p':
 				nc.ps.state = OP_P
 			case '+':
@ -94,6 +103,13 @@ func (nc *Conn) parse(buf []byte) error {
 			default:
 				goto parseErr
 			}
+		case OP_H:
+			switch b {
+			case 'M', 'm':
+				nc.ps.state = OP_M
+			default:
+				goto parseErr
+			}
 		case OP_M:
 			switch b {
 			case 'S', 's':
@ -140,8 +156,7 @@ func (nc *Conn) parse(buf []byte) error {
 				nc.ps.drop, nc.ps.as, nc.ps.state = 0, i+1, MSG_PAYLOAD

 				// jump ahead with the index. If this overruns
-				// what is left we fall out and process split
-				// buffer.
+				// what is left we fall out and process a split buffer.
 				i = nc.ps.as + nc.ps.ma.size - 1
 			default:
 				if nc.ps.argBuf != nil {
@ -415,6 +430,11 @@ func (nc *Conn) cloneMsgArg() {
 const argsLenMax = 4

 func (nc *Conn) processMsgArgs(arg []byte) error {
+	// Use separate function for header based messages.
+	if nc.ps.hdr >= 0 {
+		return nc.processHeaderMsgArgs(arg)
+	}
+
 	// Unroll splitArgs to avoid runtime/heap issues
 	a := [argsLenMax][]byte{}
 	args := a[:0]
@ -459,6 +479,57 @@ func (nc *Conn) processMsgArgs(arg []byte) error {
 	return nil
 }

+// processHeaderMsgArgs is for a header based message.
+func (nc *Conn) processHeaderMsgArgs(arg []byte) error {
+	// Unroll splitArgs to avoid runtime/heap issues
+	a := [argsLenMax][]byte{}
+	args := a[:0]
+	start := -1
+	for i, b := range arg {
+		switch b {
+		case ' ', '\t', '\r', '\n':
+			if start >= 0 {
+				args = append(args, arg[start:i])
+				start = -1
+			}
+		default:
+			if start < 0 {
+				start = i
+			}
+		}
+	}
+	if start >= 0 {
+		args = append(args, arg[start:])
+	}
+
+	switch len(args) {
+	case 4:
+		nc.ps.ma.subject = args[0]
+		nc.ps.ma.sid = parseInt64(args[1])
+		nc.ps.ma.reply = nil
+		nc.ps.ma.hdr = int(parseInt64(args[2]))
+		nc.ps.ma.size = int(parseInt64(args[3]))
+	case 5:
+		nc.ps.ma.subject = args[0]
+		nc.ps.ma.sid = parseInt64(args[1])
+		nc.ps.ma.reply = args[2]
+		nc.ps.ma.hdr = int(parseInt64(args[3]))
+		nc.ps.ma.size = int(parseInt64(args[4]))
+	default:
+		return fmt.Errorf("nats: processHeaderMsgArgs Parse Error: '%s'", arg)
+	}
+	if nc.ps.ma.sid < 0 {
+		return fmt.Errorf("nats: processHeaderMsgArgs Bad or Missing Sid: '%s'", arg)
+	}
+	if nc.ps.ma.hdr < 0 || nc.ps.ma.hdr > nc.ps.ma.size {
+		return fmt.Errorf("nats: processHeaderMsgArgs Bad or Missing Header Size: '%s'", arg)
+	}
+	if nc.ps.ma.size < 0 {
+		return fmt.Errorf("nats: processHeaderMsgArgs Bad or Missing Size: '%s'", arg)
+	}
+	return nil
+}
+
 // Ascii numbers 0-9
 const (
 	ascii_0 = 48
--- a/gateway/vendor/github.com/nats-io/nats.go/ws.go
+++ b/gateway/vendor/github.com/nats-io/nats.go/ws.go
@ -0,0 +1,745 @@
+// Copyright 2021 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 nats
+
+import (
+	"bufio"
+	"bytes"
+	"compress/flate"
+	"crypto/rand"
+	"crypto/sha1"
+	"encoding/base64"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	mrand "math/rand"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+	"unicode/utf8"
+)
+
+type wsOpCode int
+
+const (
+	// From https://tools.ietf.org/html/rfc6455#section-5.2
+	wsTextMessage   = wsOpCode(1)
+	wsBinaryMessage = wsOpCode(2)
+	wsCloseMessage  = wsOpCode(8)
+	wsPingMessage   = wsOpCode(9)
+	wsPongMessage   = wsOpCode(10)
+
+	wsFinalBit = 1 << 7
+	wsRsv1Bit  = 1 << 6 // Used for compression, from https://tools.ietf.org/html/rfc7692#section-6
+	wsRsv2Bit  = 1 << 5
+	wsRsv3Bit  = 1 << 4
+
+	wsMaskBit = 1 << 7
+
+	wsContinuationFrame     = 0
+	wsMaxFrameHeaderSize    = 14
+	wsMaxControlPayloadSize = 125
+
+	// From https://tools.ietf.org/html/rfc6455#section-11.7
+	wsCloseStatusNormalClosure      = 1000
+	wsCloseStatusNoStatusReceived   = 1005
+	wsCloseStatusAbnormalClosure    = 1006
+	wsCloseStatusInvalidPayloadData = 1007
+
+	wsScheme    = "ws"
+	wsSchemeTLS = "wss"
+
+	wsPMCExtension      = "permessage-deflate" // per-message compression
+	wsPMCSrvNoCtx       = "server_no_context_takeover"
+	wsPMCCliNoCtx       = "client_no_context_takeover"
+	wsPMCReqHeaderValue = wsPMCExtension + "; " + wsPMCSrvNoCtx + "; " + wsPMCCliNoCtx
+)
+
+// From https://tools.ietf.org/html/rfc6455#section-1.3
+var wsGUID = []byte("258EAFA5-E914-47DA-95CA-C5AB0DC85B11")
+
+var compressFinalBlock = []byte{0x00, 0x00, 0xff, 0xff, 0x01, 0x00, 0x00, 0xff, 0xff}
+
+type websocketReader struct {
+	r       io.Reader
+	pending [][]byte
+	ib      []byte
+	ff      bool
+	fc      bool
+	dc      *wsDecompressor
+	nc      *Conn
+}
+
+type wsDecompressor struct {
+	flate io.ReadCloser
+	bufs  [][]byte
+	off   int
+}
+
+type websocketWriter struct {
+	w          io.Writer
+	compress   bool
+	compressor *flate.Writer
+	ctrlFrames [][]byte // pending frames that should be sent at the next Write()
+	cm         []byte   // close message that needs to be sent when everything else has been sent
+	cmDone     bool     // a close message has been added or sent (never going back to false)
+	noMoreSend bool     // if true, even if there is a Write() call, we should not send anything
+}
+
+func (d *wsDecompressor) Read(dst []byte) (int, error) {
+	if len(dst) == 0 {
+		return 0, nil
+	}
+	if len(d.bufs) == 0 {
+		return 0, io.EOF
+	}
+	copied := 0
+	rem := len(dst)
+	for buf := d.bufs[0]; buf != nil && rem > 0; {
+		n := len(buf[d.off:])
+		if n > rem {
+			n = rem
+		}
+		copy(dst[copied:], buf[d.off:d.off+n])
+		copied += n
+		rem -= n
+		d.off += n
+		buf = d.nextBuf()
+	}
+	return copied, nil
+}
+
+func (d *wsDecompressor) nextBuf() []byte {
+	// We still have remaining data in the first buffer
+	if d.off != len(d.bufs[0]) {
+		return d.bufs[0]
+	}
+	// We read the full first buffer. Reset offset.
+	d.off = 0
+	// We were at the last buffer, so we are done.
+	if len(d.bufs) == 1 {
+		d.bufs = nil
+		return nil
+	}
+	// Here we move to the next buffer.
+	d.bufs = d.bufs[1:]
+	return d.bufs[0]
+}
+
+func (d *wsDecompressor) ReadByte() (byte, error) {
+	if len(d.bufs) == 0 {
+		return 0, io.EOF
+	}
+	b := d.bufs[0][d.off]
+	d.off++
+	d.nextBuf()
+	return b, nil
+}
+
+func (d *wsDecompressor) addBuf(b []byte) {
+	d.bufs = append(d.bufs, b)
+}
+
+func (d *wsDecompressor) decompress() ([]byte, error) {
+	d.off = 0
+	// As per https://tools.ietf.org/html/rfc7692#section-7.2.2
+	// add 0x00, 0x00, 0xff, 0xff and then a final block so that flate reader
+	// does not report unexpected EOF.
+	d.bufs = append(d.bufs, compressFinalBlock)
+	// Create or reset the decompressor with his object (wsDecompressor)
+	// that provides Read() and ReadByte() APIs that will consume from
+	// the compressed buffers (d.bufs).
+	if d.flate == nil {
+		d.flate = flate.NewReader(d)
+	} else {
+		d.flate.(flate.Resetter).Reset(d, nil)
+	}
+	// TODO: When Go 1.15 support is dropped, replace with io.ReadAll()
+	b, err := ioutil.ReadAll(d.flate)
+	// Now reset the compressed buffers list
+	d.bufs = nil
+	return b, err
+}
+
+func wsNewReader(r io.Reader) *websocketReader {
+	return &websocketReader{r: r, ff: true}
+}
+
+func (r *websocketReader) Read(p []byte) (int, error) {
+	var err error
+	var buf []byte
+
+	if l := len(r.ib); l > 0 {
+		buf = r.ib
+		r.ib = nil
+	} else {
+		if len(r.pending) > 0 {
+			return r.drainPending(p), nil
+		}
+
+		// Get some data from the underlying reader.
+		n, err := r.r.Read(p)
+		if err != nil {
+			return 0, err
+		}
+		buf = p[:n]
+	}
+
+	// Now parse this and decode frames. We will possibly read more to
+	// ensure that we get a full frame.
+	var (
+		tmpBuf []byte
+		pos    int
+		max    = len(buf)
+		rem    = 0
+	)
+	for pos < max {
+		b0 := buf[pos]
+		frameType := wsOpCode(b0 & 0xF)
+		final := b0&wsFinalBit != 0
+		compressed := b0&wsRsv1Bit != 0
+		pos++
+
+		tmpBuf, pos, err = wsGet(r.r, buf, pos, 1)
+		if err != nil {
+			return 0, err
+		}
+		b1 := tmpBuf[0]
+
+		// Store size in case it is < 125
+		rem = int(b1 & 0x7F)
+
+		switch frameType {
+		case wsPingMessage, wsPongMessage, wsCloseMessage:
+			if rem > wsMaxControlPayloadSize {
+				return 0, fmt.Errorf(
+					fmt.Sprintf("control frame length bigger than maximum allowed of %v bytes",
+						wsMaxControlPayloadSize))
+			}
+			if compressed {
+				return 0, errors.New("control frame should not be compressed")
+			}
+			if !final {
+				return 0, errors.New("control frame does not have final bit set")
+			}
+		case wsTextMessage, wsBinaryMessage:
+			if !r.ff {
+				return 0, errors.New("new message started before final frame for previous message was received")
+			}
+			r.ff = final
+			r.fc = compressed
+		case wsContinuationFrame:
+			// Compressed bit must be only set in the first frame
+			if r.ff || compressed {
+				return 0, errors.New("invalid continuation frame")
+			}
+			r.ff = final
+		default:
+			return 0, fmt.Errorf("unknown opcode %v", frameType)
+		}
+
+		// If the encoded size is <= 125, then `rem` is simply the remainder size of the
+		// frame. If it is 126, then the actual size is encoded as a uint16. For larger
+		// frames, `rem` will initially be 127 and the actual size is encoded as a uint64.
+		switch rem {
+		case 126:
+			tmpBuf, pos, err = wsGet(r.r, buf, pos, 2)
+			if err != nil {
+				return 0, err
+			}
+			rem = int(binary.BigEndian.Uint16(tmpBuf))
+		case 127:
+			tmpBuf, pos, err = wsGet(r.r, buf, pos, 8)
+			if err != nil {
+				return 0, err
+			}
+			rem = int(binary.BigEndian.Uint64(tmpBuf))
+		}
+
+		// Handle control messages in place...
+		if wsIsControlFrame(frameType) {
+			pos, err = r.handleControlFrame(frameType, buf, pos, rem)
+			if err != nil {
+				return 0, err
+			}
+			rem = 0
+			continue
+		}
+
+		var b []byte
+		// This ensures that we get the full payload for this frame.
+		b, pos, err = wsGet(r.r, buf, pos, rem)
+		if err != nil {
+			return 0, err
+		}
+		// We read the full frame.
+		rem = 0
+		addToPending := true
+		if r.fc {
+			// Don't add to pending if we are not dealing with the final frame.
+			addToPending = r.ff
+			// Add the compressed payload buffer to the list.
+			r.addCBuf(b)
+			// Decompress only when this is the final frame.
+			if r.ff {
+				b, err = r.dc.decompress()
+				if err != nil {
+					return 0, err
+				}
+				r.fc = false
+			}
+		}
+		// Add to the pending list if dealing with uncompressed frames or
+		// after we have received the full compressed message and decompressed it.
+		if addToPending {
+			r.pending = append(r.pending, b)
+		}
+	}
+	// In case of compression, there may be nothing to drain
+	if len(r.pending) > 0 {
+		return r.drainPending(p), nil
+	}
+	return 0, nil
+}
+
+func (r *websocketReader) addCBuf(b []byte) {
+	if r.dc == nil {
+		r.dc = &wsDecompressor{}
+	}
+	// Add a copy of the incoming buffer to the list of compressed buffers.
+	r.dc.addBuf(append([]byte(nil), b...))
+}
+
+func (r *websocketReader) drainPending(p []byte) int {
+	var n int
+	var max = len(p)
+
+	for i, buf := range r.pending {
+		if n+len(buf) <= max {
+			copy(p[n:], buf)
+			n += len(buf)
+		} else {
+			// Is there room left?
+			if n < max {
+				// Write the partial and update this slice.
+				rem := max - n
+				copy(p[n:], buf[:rem])
+				n += rem
+				r.pending[i] = buf[rem:]
+			}
+			// These are the remaining slices that will need to be used at
+			// the next Read() call.
+			r.pending = r.pending[i:]
+			return n
+		}
+	}
+	r.pending = r.pending[:0]
+	return n
+}
+
+func wsGet(r io.Reader, buf []byte, pos, needed int) ([]byte, int, error) {
+	avail := len(buf) - pos
+	if avail >= needed {
+		return buf[pos : pos+needed], pos + needed, nil
+	}
+	b := make([]byte, needed)
+	start := copy(b, buf[pos:])
+	for start != needed {
+		n, err := r.Read(b[start:cap(b)])
+		start += n
+		if err != nil {
+			return b, start, err
+		}
+	}
+	return b, pos + avail, nil
+}
+
+func (r *websocketReader) handleControlFrame(frameType wsOpCode, buf []byte, pos, rem int) (int, error) {
+	var payload []byte
+	var err error
+
+	statusPos := pos
+	if rem > 0 {
+		payload, pos, err = wsGet(r.r, buf, pos, rem)
+		if err != nil {
+			return pos, err
+		}
+	}
+	switch frameType {
+	case wsCloseMessage:
+		status := wsCloseStatusNoStatusReceived
+		body := ""
+		// If there is a payload, it should contain 2 unsigned bytes
+		// that represent the status code and then optional payload.
+		if len(payload) >= 2 {
+			status = int(binary.BigEndian.Uint16(buf[statusPos : statusPos+2]))
+			body = string(buf[statusPos+2 : statusPos+len(payload)])
+			if body != "" && !utf8.ValidString(body) {
+				// https://tools.ietf.org/html/rfc6455#section-5.5.1
+				// If body is present, it must be a valid utf8
+				status = wsCloseStatusInvalidPayloadData
+				body = "invalid utf8 body in close frame"
+			}
+		}
+		r.nc.wsEnqueueCloseMsg(status, body)
+		// Return io.EOF so that readLoop will close the connection as ClientClosed
+		// after processing pending buffers.
+		return pos, io.EOF
+	case wsPingMessage:
+		r.nc.wsEnqueueControlMsg(wsPongMessage, payload)
+	case wsPongMessage:
+		// Nothing to do..
+	}
+	return pos, nil
+}
+
+func (w *websocketWriter) Write(p []byte) (int, error) {
+	if w.noMoreSend {
+		return 0, nil
+	}
+	var total int
+	var n int
+	var err error
+	// If there are control frames, they can be sent now. Actually spec says
+	// that they should be sent ASAP, so we will send before any application data.
+	if len(w.ctrlFrames) > 0 {
+		n, err = w.writeCtrlFrames()
+		if err != nil {
+			return n, err
+		}
+		total += n
+	}
+	// Do the following only if there is something to send.
+	// We will end with checking for need to send close message.
+	if len(p) > 0 {
+		if w.compress {
+			buf := &bytes.Buffer{}
+			if w.compressor == nil {
+				w.compressor, _ = flate.NewWriter(buf, flate.BestSpeed)
+			} else {
+				w.compressor.Reset(buf)
+			}
+			w.compressor.Write(p)
+			w.compressor.Close()
+			b := buf.Bytes()
+			p = b[:len(b)-4]
+		}
+		fh, key := wsCreateFrameHeader(w.compress, wsBinaryMessage, len(p))
+		wsMaskBuf(key, p)
+		n, err = w.w.Write(fh)
+		total += n
+		if err == nil {
+			n, err = w.w.Write(p)
+			total += n
+		}
+	}
+	if err == nil && w.cm != nil {
+		n, err = w.writeCloseMsg()
+		total += n
+	}
+	return total, err
+}
+
+func (w *websocketWriter) writeCtrlFrames() (int, error) {
+	var (
+		n     int
+		total int
+		i     int
+		err   error
+	)
+	for ; i < len(w.ctrlFrames); i++ {
+		buf := w.ctrlFrames[i]
+		n, err = w.w.Write(buf)
+		total += n
+		if err != nil {
+			break
+		}
+	}
+	if i != len(w.ctrlFrames) {
+		w.ctrlFrames = w.ctrlFrames[i+1:]
+	} else {
+		w.ctrlFrames = w.ctrlFrames[:0]
+	}
+	return total, err
+}
+
+func (w *websocketWriter) writeCloseMsg() (int, error) {
+	n, err := w.w.Write(w.cm)
+	w.cm, w.noMoreSend = nil, true
+	return n, err
+}
+
+func wsMaskBuf(key, buf []byte) {
+	for i := 0; i < len(buf); i++ {
+		buf[i] ^= key[i&3]
+	}
+}
+
+// Create the frame header.
+// Encodes the frame type and optional compression flag, and the size of the payload.
+func wsCreateFrameHeader(compressed bool, frameType wsOpCode, l int) ([]byte, []byte) {
+	fh := make([]byte, wsMaxFrameHeaderSize)
+	n, key := wsFillFrameHeader(fh, compressed, frameType, l)
+	return fh[:n], key
+}
+
+func wsFillFrameHeader(fh []byte, compressed bool, frameType wsOpCode, l int) (int, []byte) {
+	var n int
+	b := byte(frameType)
+	b |= wsFinalBit
+	if compressed {
+		b |= wsRsv1Bit
+	}
+	b1 := byte(wsMaskBit)
+	switch {
+	case l <= 125:
+		n = 2
+		fh[0] = b
+		fh[1] = b1 | byte(l)
+	case l < 65536:
+		n = 4
+		fh[0] = b
+		fh[1] = b1 | 126
+		binary.BigEndian.PutUint16(fh[2:], uint16(l))
+	default:
+		n = 10
+		fh[0] = b
+		fh[1] = b1 | 127
+		binary.BigEndian.PutUint64(fh[2:], uint64(l))
+	}
+	var key []byte
+	var keyBuf [4]byte
+	if _, err := io.ReadFull(rand.Reader, keyBuf[:4]); err != nil {
+		kv := mrand.Int31()
+		binary.LittleEndian.PutUint32(keyBuf[:4], uint32(kv))
+	}
+	copy(fh[n:], keyBuf[:4])
+	key = fh[n : n+4]
+	n += 4
+	return n, key
+}
+
+func (nc *Conn) wsInitHandshake(u *url.URL) error {
+	compress := nc.Opts.Compression
+	tlsRequired := u.Scheme == wsSchemeTLS || nc.Opts.Secure || nc.Opts.TLSConfig != nil
+	// Do TLS here as needed.
+	if tlsRequired {
+		if err := nc.makeTLSConn(); err != nil {
+			return err
+		}
+	} else {
+		nc.bindToNewConn()
+	}
+
+	var err error
+
+	// For http request, we need the passed URL to contain either http or https scheme.
+	scheme := "http"
+	if tlsRequired {
+		scheme = "https"
+	}
+	ustr := fmt.Sprintf("%s://%s", scheme, u.Host)
+	u, err = url.Parse(ustr)
+	if err != nil {
+		return err
+	}
+	req := &http.Request{
+		Method:     "GET",
+		URL:        u,
+		Proto:      "HTTP/1.1",
+		ProtoMajor: 1,
+		ProtoMinor: 1,
+		Header:     make(http.Header),
+		Host:       u.Host,
+	}
+	wsKey, err := wsMakeChallengeKey()
+	if err != nil {
+		return err
+	}
+
+	req.Header["Upgrade"] = []string{"websocket"}
+	req.Header["Connection"] = []string{"Upgrade"}
+	req.Header["Sec-WebSocket-Key"] = []string{wsKey}
+	req.Header["Sec-WebSocket-Version"] = []string{"13"}
+	if compress {
+		req.Header.Add("Sec-WebSocket-Extensions", wsPMCReqHeaderValue)
+	}
+	if err := req.Write(nc.conn); err != nil {
+		return err
+	}
+
+	var resp *http.Response
+
+	br := bufio.NewReaderSize(nc.conn, 4096)
+	nc.conn.SetReadDeadline(time.Now().Add(nc.Opts.Timeout))
+	resp, err = http.ReadResponse(br, req)
+	if err == nil &&
+		(resp.StatusCode != 101 ||
+			!strings.EqualFold(resp.Header.Get("Upgrade"), "websocket") ||
+			!strings.EqualFold(resp.Header.Get("Connection"), "upgrade") ||
+			resp.Header.Get("Sec-Websocket-Accept") != wsAcceptKey(wsKey)) {
+
+		err = fmt.Errorf("invalid websocket connection")
+	}
+	// Check compression extension...
+	if err == nil && compress {
+		// Check that not only permessage-deflate extension is present, but that
+		// we also have server and client no context take over.
+		srvCompress, noCtxTakeover := wsPMCExtensionSupport(resp.Header)
+
+		// If server does not support compression, then simply disable it in our side.
+		if !srvCompress {
+			compress = false
+		} else if !noCtxTakeover {
+			err = fmt.Errorf("compression negotiation error")
+		}
+	}
+	if resp != nil {
+		resp.Body.Close()
+	}
+	nc.conn.SetReadDeadline(time.Time{})
+	if err != nil {
+		return err
+	}
+
+	wsr := wsNewReader(nc.br.r)
+	wsr.nc = nc
+	// We have to slurp whatever is in the bufio reader and copy to br.r
+	if n := br.Buffered(); n != 0 {
+		wsr.ib, _ = br.Peek(n)
+	}
+	nc.br.r = wsr
+	nc.bw.w = &websocketWriter{w: nc.bw.w, compress: compress}
+	nc.ws = true
+	return nil
+}
+
+func (nc *Conn) wsClose() {
+	nc.mu.Lock()
+	defer nc.mu.Unlock()
+	if !nc.ws {
+		return
+	}
+	nc.wsEnqueueCloseMsgLocked(wsCloseStatusNormalClosure, _EMPTY_)
+}
+
+func (nc *Conn) wsEnqueueCloseMsg(status int, payload string) {
+	// In some low-level unit tests it will happen...
+	if nc == nil {
+		return
+	}
+	nc.mu.Lock()
+	nc.wsEnqueueCloseMsgLocked(status, payload)
+	nc.mu.Unlock()
+}
+
+func (nc *Conn) wsEnqueueCloseMsgLocked(status int, payload string) {
+	wr, ok := nc.bw.w.(*websocketWriter)
+	if !ok || wr.cmDone {
+		return
+	}
+	statusAndPayloadLen := 2 + len(payload)
+	frame := make([]byte, 2+4+statusAndPayloadLen)
+	n, key := wsFillFrameHeader(frame, false, wsCloseMessage, statusAndPayloadLen)
+	// Set the status
+	binary.BigEndian.PutUint16(frame[n:], uint16(status))
+	// If there is a payload, copy
+	if len(payload) > 0 {
+		copy(frame[n+2:], payload)
+	}
+	// Mask status + payload
+	wsMaskBuf(key, frame[n:n+statusAndPayloadLen])
+	wr.cm = frame
+	wr.cmDone = true
+	nc.bw.flush()
+}
+
+func (nc *Conn) wsEnqueueControlMsg(frameType wsOpCode, payload []byte) {
+	// In some low-level unit tests it will happen...
+	if nc == nil {
+		return
+	}
+	fh, key := wsCreateFrameHeader(false, frameType, len(payload))
+	nc.mu.Lock()
+	wr, ok := nc.bw.w.(*websocketWriter)
+	if !ok {
+		nc.mu.Unlock()
+		return
+	}
+	wr.ctrlFrames = append(wr.ctrlFrames, fh)
+	if len(payload) > 0 {
+		wsMaskBuf(key, payload)
+		wr.ctrlFrames = append(wr.ctrlFrames, payload)
+	}
+	nc.bw.flush()
+	nc.mu.Unlock()
+}
+
+func wsPMCExtensionSupport(header http.Header) (bool, bool) {
+	for _, extensionList := range header["Sec-Websocket-Extensions"] {
+		extensions := strings.Split(extensionList, ",")
+		for _, extension := range extensions {
+			extension = strings.Trim(extension, " \t")
+			params := strings.Split(extension, ";")
+			for i, p := range params {
+				p = strings.Trim(p, " \t")
+				if strings.EqualFold(p, wsPMCExtension) {
+					var snc bool
+					var cnc bool
+					for j := i + 1; j < len(params); j++ {
+						p = params[j]
+						p = strings.Trim(p, " \t")
+						if strings.EqualFold(p, wsPMCSrvNoCtx) {
+							snc = true
+						} else if strings.EqualFold(p, wsPMCCliNoCtx) {
+							cnc = true
+						}
+						if snc && cnc {
+							return true, true
+						}
+					}
+					return true, false
+				}
+			}
+		}
+	}
+	return false, false
+}
+
+func wsMakeChallengeKey() (string, error) {
+	p := make([]byte, 16)
+	if _, err := io.ReadFull(rand.Reader, p); err != nil {
+		return "", err
+	}
+	return base64.StdEncoding.EncodeToString(p), nil
+}
+
+func wsAcceptKey(key string) string {
+	h := sha1.New()
+	h.Write([]byte(key))
+	h.Write(wsGUID)
+	return base64.StdEncoding.EncodeToString(h.Sum(nil))
+}
+
+// Returns true if the op code corresponds to a control frame.
+func wsIsControlFrame(frameType wsOpCode) bool {
+	return frameType >= wsCloseMessage
+}
+
+func isWebsocketScheme(u *url.URL) bool {
+	return u.Scheme == wsScheme || u.Scheme == wsSchemeTLS
+}