From b453d3ee377e513da7fdf5f9241e0be088489717 Mon Sep 17 00:00:00 2001
From: Anmol Sethi <hi@nhooyr.io>
Date: Mon, 18 May 2020 02:47:53 -0400
Subject: [PATCH] Move all Wasm related code into ws_js.go
This way we don't pollute the directory tree.
---
accept_js.go | 20 ----
close.go | 205 +++++++++++++++++++++++++++++++++++
close_notjs.go | 211 ------------------------------------
compress.go | 180 +++++++++++++++++++++++++++++++
compress_notjs.go | 181 -------------------------------
conn.go | 264 +++++++++++++++++++++++++++++++++++++++++++++
conn_notjs.go | 265 ----------------------------------------------
ws_js.go | 134 +++++++++++++++++++++++
8 files changed, 783 insertions(+), 677 deletions(-)
delete mode 100644 accept_js.go
delete mode 100644 close_notjs.go
delete mode 100644 compress_notjs.go
delete mode 100644 conn_notjs.go
diff --git a/accept_js.go b/accept_js.go
deleted file mode 100644
index daad4b7..0000000
--- a/accept_js.go
+++ /dev/null
@@ -1,20 +0,0 @@
-package websocket
-
-import (
- "errors"
- "net/http"
-)
-
-// AcceptOptions represents Accept's options.
-type AcceptOptions struct {
- Subprotocols []string
- InsecureSkipVerify bool
- OriginPatterns []string
- CompressionMode CompressionMode
- CompressionThreshold int
-}
-
-// Accept is stubbed out for Wasm.
-func Accept(w http.ResponseWriter, r *http.Request, opts *AcceptOptions) (*Conn, error) {
- return nil, errors.New("unimplemented")
-}
diff --git a/close.go b/close.go
index 7cbc19e..d76dc2f 100644
--- a/close.go
+++ b/close.go
@@ -1,8 +1,16 @@
+// +build !js
+
package websocket
import (
+ "context"
+ "encoding/binary"
"errors"
"fmt"
+ "log"
+ "time"
+
+ "nhooyr.io/websocket/internal/errd"
)
// StatusCode represents a WebSocket status code.
@@ -74,3 +82,200 @@ func CloseStatus(err error) StatusCode {
}
return -1
}
+
+// Close performs the WebSocket close handshake with the given status code and reason.
+//
+// It will write a WebSocket close frame with a timeout of 5s and then wait 5s for
+// the peer to send a close frame.
+// All data messages received from the peer during the close handshake will be discarded.
+//
+// The connection can only be closed once. Additional calls to Close
+// are no-ops.
+//
+// The maximum length of reason must be 125 bytes. Avoid
+// sending a dynamic reason.
+//
+// Close will unblock all goroutines interacting with the connection once
+// complete.
+func (c *Conn) Close(code StatusCode, reason string) error {
+ return c.closeHandshake(code, reason)
+}
+
+func (c *Conn) closeHandshake(code StatusCode, reason string) (err error) {
+ defer errd.Wrap(&err, "failed to close WebSocket")
+
+ writeErr := c.writeClose(code, reason)
+ closeHandshakeErr := c.waitCloseHandshake()
+
+ if writeErr != nil {
+ return writeErr
+ }
+
+ if CloseStatus(closeHandshakeErr) == -1 {
+ return closeHandshakeErr
+ }
+
+ return nil
+}
+
+var errAlreadyWroteClose = errors.New("already wrote close")
+
+func (c *Conn) writeClose(code StatusCode, reason string) error {
+ c.closeMu.Lock()
+ wroteClose := c.wroteClose
+ c.wroteClose = true
+ c.closeMu.Unlock()
+ if wroteClose {
+ return errAlreadyWroteClose
+ }
+
+ ce := CloseError{
+ Code: code,
+ Reason: reason,
+ }
+
+ var p []byte
+ var marshalErr error
+ if ce.Code != StatusNoStatusRcvd {
+ p, marshalErr = ce.bytes()
+ if marshalErr != nil {
+ log.Printf("websocket: %v", marshalErr)
+ }
+ }
+
+ writeErr := c.writeControl(context.Background(), opClose, p)
+ if CloseStatus(writeErr) != -1 {
+ // Not a real error if it's due to a close frame being received.
+ writeErr = nil
+ }
+
+ // We do this after in case there was an error writing the close frame.
+ c.setCloseErr(fmt.Errorf("sent close frame: %w", ce))
+
+ if marshalErr != nil {
+ return marshalErr
+ }
+ return writeErr
+}
+
+func (c *Conn) waitCloseHandshake() error {
+ defer c.close(nil)
+
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+ defer cancel()
+
+ err := c.readMu.lock(ctx)
+ if err != nil {
+ return err
+ }
+ defer c.readMu.unlock()
+
+ if c.readCloseFrameErr != nil {
+ return c.readCloseFrameErr
+ }
+
+ for {
+ h, err := c.readLoop(ctx)
+ if err != nil {
+ return err
+ }
+
+ for i := int64(0); i < h.payloadLength; i++ {
+ _, err := c.br.ReadByte()
+ if err != nil {
+ return err
+ }
+ }
+ }
+}
+
+func parseClosePayload(p []byte) (CloseError, error) {
+ if len(p) == 0 {
+ return CloseError{
+ Code: StatusNoStatusRcvd,
+ }, nil
+ }
+
+ if len(p) < 2 {
+ return CloseError{}, fmt.Errorf("close payload %q too small, cannot even contain the 2 byte status code", p)
+ }
+
+ ce := CloseError{
+ Code: StatusCode(binary.BigEndian.Uint16(p)),
+ Reason: string(p[2:]),
+ }
+
+ if !validWireCloseCode(ce.Code) {
+ return CloseError{}, fmt.Errorf("invalid status code %v", ce.Code)
+ }
+
+ return ce, nil
+}
+
+// See http://www.iana.org/assignments/websocket/websocket.xhtml#close-code-number
+// and https://tools.ietf.org/html/rfc6455#section-7.4.1
+func validWireCloseCode(code StatusCode) bool {
+ switch code {
+ case statusReserved, StatusNoStatusRcvd, StatusAbnormalClosure, StatusTLSHandshake:
+ return false
+ }
+
+ if code >= StatusNormalClosure && code <= StatusBadGateway {
+ return true
+ }
+ if code >= 3000 && code <= 4999 {
+ return true
+ }
+
+ return false
+}
+
+func (ce CloseError) bytes() ([]byte, error) {
+ p, err := ce.bytesErr()
+ if err != nil {
+ err = fmt.Errorf("failed to marshal close frame: %w", err)
+ ce = CloseError{
+ Code: StatusInternalError,
+ }
+ p, _ = ce.bytesErr()
+ }
+ return p, err
+}
+
+const maxCloseReason = maxControlPayload - 2
+
+func (ce CloseError) bytesErr() ([]byte, error) {
+ if len(ce.Reason) > maxCloseReason {
+ return nil, fmt.Errorf("reason string max is %v but got %q with length %v", maxCloseReason, ce.Reason, len(ce.Reason))
+ }
+
+ if !validWireCloseCode(ce.Code) {
+ return nil, fmt.Errorf("status code %v cannot be set", ce.Code)
+ }
+
+ buf := make([]byte, 2+len(ce.Reason))
+ binary.BigEndian.PutUint16(buf, uint16(ce.Code))
+ copy(buf[2:], ce.Reason)
+ return buf, nil
+}
+
+func (c *Conn) setCloseErr(err error) {
+ c.closeMu.Lock()
+ c.setCloseErrLocked(err)
+ c.closeMu.Unlock()
+}
+
+func (c *Conn) setCloseErrLocked(err error) {
+ if c.closeErr == nil {
+ c.closeErr = fmt.Errorf("WebSocket closed: %w", err)
+ }
+}
+
+func (c *Conn) isClosed() bool {
+ select {
+ case <-c.closed:
+ return true
+ default:
+ return false
+ }
+}
diff --git a/close_notjs.go b/close_notjs.go
deleted file mode 100644
index 4251311..0000000
--- a/close_notjs.go
+++ /dev/null
@@ -1,211 +0,0 @@
-// +build !js
-
-package websocket
-
-import (
- "context"
- "encoding/binary"
- "errors"
- "fmt"
- "log"
- "time"
-
- "nhooyr.io/websocket/internal/errd"
-)
-
-// Close performs the WebSocket close handshake with the given status code and reason.
-//
-// It will write a WebSocket close frame with a timeout of 5s and then wait 5s for
-// the peer to send a close frame.
-// All data messages received from the peer during the close handshake will be discarded.
-//
-// The connection can only be closed once. Additional calls to Close
-// are no-ops.
-//
-// The maximum length of reason must be 125 bytes. Avoid
-// sending a dynamic reason.
-//
-// Close will unblock all goroutines interacting with the connection once
-// complete.
-func (c *Conn) Close(code StatusCode, reason string) error {
- return c.closeHandshake(code, reason)
-}
-
-func (c *Conn) closeHandshake(code StatusCode, reason string) (err error) {
- defer errd.Wrap(&err, "failed to close WebSocket")
-
- writeErr := c.writeClose(code, reason)
- closeHandshakeErr := c.waitCloseHandshake()
-
- if writeErr != nil {
- return writeErr
- }
-
- if CloseStatus(closeHandshakeErr) == -1 {
- return closeHandshakeErr
- }
-
- return nil
-}
-
-var errAlreadyWroteClose = errors.New("already wrote close")
-
-func (c *Conn) writeClose(code StatusCode, reason string) error {
- c.closeMu.Lock()
- wroteClose := c.wroteClose
- c.wroteClose = true
- c.closeMu.Unlock()
- if wroteClose {
- return errAlreadyWroteClose
- }
-
- ce := CloseError{
- Code: code,
- Reason: reason,
- }
-
- var p []byte
- var marshalErr error
- if ce.Code != StatusNoStatusRcvd {
- p, marshalErr = ce.bytes()
- if marshalErr != nil {
- log.Printf("websocket: %v", marshalErr)
- }
- }
-
- writeErr := c.writeControl(context.Background(), opClose, p)
- if CloseStatus(writeErr) != -1 {
- // Not a real error if it's due to a close frame being received.
- writeErr = nil
- }
-
- // We do this after in case there was an error writing the close frame.
- c.setCloseErr(fmt.Errorf("sent close frame: %w", ce))
-
- if marshalErr != nil {
- return marshalErr
- }
- return writeErr
-}
-
-func (c *Conn) waitCloseHandshake() error {
- defer c.close(nil)
-
- ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
- defer cancel()
-
- err := c.readMu.lock(ctx)
- if err != nil {
- return err
- }
- defer c.readMu.unlock()
-
- if c.readCloseFrameErr != nil {
- return c.readCloseFrameErr
- }
-
- for {
- h, err := c.readLoop(ctx)
- if err != nil {
- return err
- }
-
- for i := int64(0); i < h.payloadLength; i++ {
- _, err := c.br.ReadByte()
- if err != nil {
- return err
- }
- }
- }
-}
-
-func parseClosePayload(p []byte) (CloseError, error) {
- if len(p) == 0 {
- return CloseError{
- Code: StatusNoStatusRcvd,
- }, nil
- }
-
- if len(p) < 2 {
- return CloseError{}, fmt.Errorf("close payload %q too small, cannot even contain the 2 byte status code", p)
- }
-
- ce := CloseError{
- Code: StatusCode(binary.BigEndian.Uint16(p)),
- Reason: string(p[2:]),
- }
-
- if !validWireCloseCode(ce.Code) {
- return CloseError{}, fmt.Errorf("invalid status code %v", ce.Code)
- }
-
- return ce, nil
-}
-
-// See http://www.iana.org/assignments/websocket/websocket.xhtml#close-code-number
-// and https://tools.ietf.org/html/rfc6455#section-7.4.1
-func validWireCloseCode(code StatusCode) bool {
- switch code {
- case statusReserved, StatusNoStatusRcvd, StatusAbnormalClosure, StatusTLSHandshake:
- return false
- }
-
- if code >= StatusNormalClosure && code <= StatusBadGateway {
- return true
- }
- if code >= 3000 && code <= 4999 {
- return true
- }
-
- return false
-}
-
-func (ce CloseError) bytes() ([]byte, error) {
- p, err := ce.bytesErr()
- if err != nil {
- err = fmt.Errorf("failed to marshal close frame: %w", err)
- ce = CloseError{
- Code: StatusInternalError,
- }
- p, _ = ce.bytesErr()
- }
- return p, err
-}
-
-const maxCloseReason = maxControlPayload - 2
-
-func (ce CloseError) bytesErr() ([]byte, error) {
- if len(ce.Reason) > maxCloseReason {
- return nil, fmt.Errorf("reason string max is %v but got %q with length %v", maxCloseReason, ce.Reason, len(ce.Reason))
- }
-
- if !validWireCloseCode(ce.Code) {
- return nil, fmt.Errorf("status code %v cannot be set", ce.Code)
- }
-
- buf := make([]byte, 2+len(ce.Reason))
- binary.BigEndian.PutUint16(buf, uint16(ce.Code))
- copy(buf[2:], ce.Reason)
- return buf, nil
-}
-
-func (c *Conn) setCloseErr(err error) {
- c.closeMu.Lock()
- c.setCloseErrLocked(err)
- c.closeMu.Unlock()
-}
-
-func (c *Conn) setCloseErrLocked(err error) {
- if c.closeErr == nil {
- c.closeErr = fmt.Errorf("WebSocket closed: %w", err)
- }
-}
-
-func (c *Conn) isClosed() bool {
- select {
- case <-c.closed:
- return true
- default:
- return false
- }
-}
diff --git a/compress.go b/compress.go
index 80b46d1..63d961b 100644
--- a/compress.go
+++ b/compress.go
@@ -1,5 +1,15 @@
+// +build !js
+
package websocket
+import (
+ "io"
+ "net/http"
+ "sync"
+
+ "github.com/klauspost/compress/flate"
+)
+
// CompressionMode represents the modes available to the deflate extension.
// See https://tools.ietf.org/html/rfc7692
//
@@ -37,3 +47,173 @@ const (
// important than bandwidth.
CompressionDisabled
)
+
+func (m CompressionMode) opts() *compressionOptions {
+ return &compressionOptions{
+ clientNoContextTakeover: m == CompressionNoContextTakeover,
+ serverNoContextTakeover: m == CompressionNoContextTakeover,
+ }
+}
+
+type compressionOptions struct {
+ clientNoContextTakeover bool
+ serverNoContextTakeover bool
+}
+
+func (copts *compressionOptions) setHeader(h http.Header) {
+ s := "permessage-deflate"
+ if copts.clientNoContextTakeover {
+ s += "; client_no_context_takeover"
+ }
+ if copts.serverNoContextTakeover {
+ s += "; server_no_context_takeover"
+ }
+ h.Set("Sec-WebSocket-Extensions", s)
+}
+
+// These bytes are required to get flate.Reader to return.
+// They are removed when sending to avoid the overhead as
+// WebSocket framing tell's when the message has ended but then
+// we need to add them back otherwise flate.Reader keeps
+// trying to return more bytes.
+const deflateMessageTail = "\x00\x00\xff\xff"
+
+type trimLastFourBytesWriter struct {
+ w io.Writer
+ tail []byte
+}
+
+func (tw *trimLastFourBytesWriter) reset() {
+ if tw != nil && tw.tail != nil {
+ tw.tail = tw.tail[:0]
+ }
+}
+
+func (tw *trimLastFourBytesWriter) Write(p []byte) (int, error) {
+ if tw.tail == nil {
+ tw.tail = make([]byte, 0, 4)
+ }
+
+ extra := len(tw.tail) + len(p) - 4
+
+ if extra <= 0 {
+ tw.tail = append(tw.tail, p...)
+ return len(p), nil
+ }
+
+ // Now we need to write as many extra bytes as we can from the previous tail.
+ if extra > len(tw.tail) {
+ extra = len(tw.tail)
+ }
+ if extra > 0 {
+ _, err := tw.w.Write(tw.tail[:extra])
+ if err != nil {
+ return 0, err
+ }
+
+ // Shift remaining bytes in tail over.
+ n := copy(tw.tail, tw.tail[extra:])
+ tw.tail = tw.tail[:n]
+ }
+
+ // If p is less than or equal to 4 bytes,
+ // all of it is is part of the tail.
+ if len(p) <= 4 {
+ tw.tail = append(tw.tail, p...)
+ return len(p), nil
+ }
+
+ // Otherwise, only the last 4 bytes are.
+ tw.tail = append(tw.tail, p[len(p)-4:]...)
+
+ p = p[:len(p)-4]
+ n, err := tw.w.Write(p)
+ return n + 4, err
+}
+
+var flateReaderPool sync.Pool
+
+func getFlateReader(r io.Reader, dict []byte) io.Reader {
+ fr, ok := flateReaderPool.Get().(io.Reader)
+ if !ok {
+ return flate.NewReaderDict(r, dict)
+ }
+ fr.(flate.Resetter).Reset(r, dict)
+ return fr
+}
+
+func putFlateReader(fr io.Reader) {
+ flateReaderPool.Put(fr)
+}
+
+type slidingWindow struct {
+ buf []byte
+}
+
+var swPoolMu sync.RWMutex
+var swPool = map[int]*sync.Pool{}
+
+func slidingWindowPool(n int) *sync.Pool {
+ swPoolMu.RLock()
+ p, ok := swPool[n]
+ swPoolMu.RUnlock()
+ if ok {
+ return p
+ }
+
+ p = &sync.Pool{}
+
+ swPoolMu.Lock()
+ swPool[n] = p
+ swPoolMu.Unlock()
+
+ return p
+}
+
+func (sw *slidingWindow) init(n int) {
+ if sw.buf != nil {
+ return
+ }
+
+ if n == 0 {
+ n = 32768
+ }
+
+ p := slidingWindowPool(n)
+ buf, ok := p.Get().([]byte)
+ if ok {
+ sw.buf = buf[:0]
+ } else {
+ sw.buf = make([]byte, 0, n)
+ }
+}
+
+func (sw *slidingWindow) close() {
+ if sw.buf == nil {
+ return
+ }
+
+ swPoolMu.Lock()
+ swPool[cap(sw.buf)].Put(sw.buf)
+ swPoolMu.Unlock()
+ sw.buf = nil
+}
+
+func (sw *slidingWindow) write(p []byte) {
+ if len(p) >= cap(sw.buf) {
+ sw.buf = sw.buf[:cap(sw.buf)]
+ p = p[len(p)-cap(sw.buf):]
+ copy(sw.buf, p)
+ return
+ }
+
+ left := cap(sw.buf) - len(sw.buf)
+ if left < len(p) {
+ // We need to shift spaceNeeded bytes from the end to make room for p at the end.
+ spaceNeeded := len(p) - left
+ copy(sw.buf, sw.buf[spaceNeeded:])
+ sw.buf = sw.buf[:len(sw.buf)-spaceNeeded]
+ }
+
+ sw.buf = append(sw.buf, p...)
+}
diff --git a/compress_notjs.go b/compress_notjs.go
deleted file mode 100644
index 809a272..0000000
--- a/compress_notjs.go
+++ /dev/null
@@ -1,181 +0,0 @@
-// +build !js
-
-package websocket
-
-import (
- "io"
- "net/http"
- "sync"
-
- "github.com/klauspost/compress/flate"
-)
-
-func (m CompressionMode) opts() *compressionOptions {
- return &compressionOptions{
- clientNoContextTakeover: m == CompressionNoContextTakeover,
- serverNoContextTakeover: m == CompressionNoContextTakeover,
- }
-}
-
-type compressionOptions struct {
- clientNoContextTakeover bool
- serverNoContextTakeover bool
-}
-
-func (copts *compressionOptions) setHeader(h http.Header) {
- s := "permessage-deflate"
- if copts.clientNoContextTakeover {
- s += "; client_no_context_takeover"
- }
- if copts.serverNoContextTakeover {
- s += "; server_no_context_takeover"
- }
- h.Set("Sec-WebSocket-Extensions", s)
-}
-
-// These bytes are required to get flate.Reader to return.
-// They are removed when sending to avoid the overhead as
-// WebSocket framing tell's when the message has ended but then
-// we need to add them back otherwise flate.Reader keeps
-// trying to return more bytes.
-const deflateMessageTail = "\x00\x00\xff\xff"
-
-type trimLastFourBytesWriter struct {
- w io.Writer
- tail []byte
-}
-
-func (tw *trimLastFourBytesWriter) reset() {
- if tw != nil && tw.tail != nil {
- tw.tail = tw.tail[:0]
- }
-}
-
-func (tw *trimLastFourBytesWriter) Write(p []byte) (int, error) {
- if tw.tail == nil {
- tw.tail = make([]byte, 0, 4)
- }
-
- extra := len(tw.tail) + len(p) - 4
-
- if extra <= 0 {
- tw.tail = append(tw.tail, p...)
- return len(p), nil
- }
-
- // Now we need to write as many extra bytes as we can from the previous tail.
- if extra > len(tw.tail) {
- extra = len(tw.tail)
- }
- if extra > 0 {
- _, err := tw.w.Write(tw.tail[:extra])
- if err != nil {
- return 0, err
- }
-
- // Shift remaining bytes in tail over.
- n := copy(tw.tail, tw.tail[extra:])
- tw.tail = tw.tail[:n]
- }
-
- // If p is less than or equal to 4 bytes,
- // all of it is is part of the tail.
- if len(p) <= 4 {
- tw.tail = append(tw.tail, p...)
- return len(p), nil
- }
-
- // Otherwise, only the last 4 bytes are.
- tw.tail = append(tw.tail, p[len(p)-4:]...)
-
- p = p[:len(p)-4]
- n, err := tw.w.Write(p)
- return n + 4, err
-}
-
-var flateReaderPool sync.Pool
-
-func getFlateReader(r io.Reader, dict []byte) io.Reader {
- fr, ok := flateReaderPool.Get().(io.Reader)
- if !ok {
- return flate.NewReaderDict(r, dict)
- }
- fr.(flate.Resetter).Reset(r, dict)
- return fr
-}
-
-func putFlateReader(fr io.Reader) {
- flateReaderPool.Put(fr)
-}
-
-type slidingWindow struct {
- buf []byte
-}
-
-var swPoolMu sync.RWMutex
-var swPool = map[int]*sync.Pool{}
-
-func slidingWindowPool(n int) *sync.Pool {
- swPoolMu.RLock()
- p, ok := swPool[n]
- swPoolMu.RUnlock()
- if ok {
- return p
- }
-
- p = &sync.Pool{}
-
- swPoolMu.Lock()
- swPool[n] = p
- swPoolMu.Unlock()
-
- return p
-}
-
-func (sw *slidingWindow) init(n int) {
- if sw.buf != nil {
- return
- }
-
- if n == 0 {
- n = 32768
- }
-
- p := slidingWindowPool(n)
- buf, ok := p.Get().([]byte)
- if ok {
- sw.buf = buf[:0]
- } else {
- sw.buf = make([]byte, 0, n)
- }
-}
-
-func (sw *slidingWindow) close() {
- if sw.buf == nil {
- return
- }
-
- swPoolMu.Lock()
- swPool[cap(sw.buf)].Put(sw.buf)
- swPoolMu.Unlock()
- sw.buf = nil
-}
-
-func (sw *slidingWindow) write(p []byte) {
- if len(p) >= cap(sw.buf) {
- sw.buf = sw.buf[:cap(sw.buf)]
- p = p[len(p)-cap(sw.buf):]
- copy(sw.buf, p)
- return
- }
-
- left := cap(sw.buf) - len(sw.buf)
- if left < len(p) {
- // We need to shift spaceNeeded bytes from the end to make room for p at the end.
- spaceNeeded := len(p) - left
- copy(sw.buf, sw.buf[spaceNeeded:])
- sw.buf = sw.buf[:len(sw.buf)-spaceNeeded]
- }
-
- sw.buf = append(sw.buf, p...)
-}
diff --git a/conn.go b/conn.go
index a41808b..e208d11 100644
--- a/conn.go
+++ b/conn.go
@@ -1,5 +1,19 @@
+// +build !js
+
package websocket
+import (
+ "bufio"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "runtime"
+ "strconv"
+ "sync"
+ "sync/atomic"
+)
+
// MessageType represents the type of a WebSocket message.
// See https://tools.ietf.org/html/rfc6455#section-5.6
type MessageType int
@@ -11,3 +25,253 @@ const (
// MessageBinary is for binary messages like protobufs.
MessageBinary
)
+
+// Conn represents a WebSocket connection.
+// All methods may be called concurrently except for Reader and Read.
+//
+// You must always read from the connection. Otherwise control
+// frames will not be handled. See Reader and CloseRead.
+//
+// Be sure to call Close on the connection when you
+// are finished with it to release associated resources.
+//
+// On any error from any method, the connection is closed
+// with an appropriate reason.
+type Conn struct {
+ subprotocol string
+ rwc io.ReadWriteCloser
+ client bool
+ copts *compressionOptions
+ flateThreshold int
+ br *bufio.Reader
+ bw *bufio.Writer
+
+ readTimeout chan context.Context
+ writeTimeout chan context.Context
+
+ // Read state.
+ readMu *mu
+ readHeaderBuf [8]byte
+ readControlBuf [maxControlPayload]byte
+ msgReader *msgReader
+ readCloseFrameErr error
+
+ // Write state.
+ msgWriterState *msgWriterState
+ writeFrameMu *mu
+ writeBuf []byte
+ writeHeaderBuf [8]byte
+ writeHeader header
+
+ closed chan struct{}
+ closeMu sync.Mutex
+ closeErr error
+ wroteClose bool
+
+ pingCounter int32
+ activePingsMu sync.Mutex
+ activePings map[string]chan<- struct{}
+}
+
+type connConfig struct {
+ subprotocol string
+ rwc io.ReadWriteCloser
+ client bool
+ copts *compressionOptions
+ flateThreshold int
+
+ br *bufio.Reader
+ bw *bufio.Writer
+}
+
+func newConn(cfg connConfig) *Conn {
+ c := &Conn{
+ subprotocol: cfg.subprotocol,
+ rwc: cfg.rwc,
+ client: cfg.client,
+ copts: cfg.copts,
+ flateThreshold: cfg.flateThreshold,
+
+ br: cfg.br,
+ bw: cfg.bw,
+
+ readTimeout: make(chan context.Context),
+ writeTimeout: make(chan context.Context),
+
+ closed: make(chan struct{}),
+ activePings: make(map[string]chan<- struct{}),
+ }
+
+ c.readMu = newMu(c)
+ c.writeFrameMu = newMu(c)
+
+ c.msgReader = newMsgReader(c)
+
+ c.msgWriterState = newMsgWriterState(c)
+ if c.client {
+ c.writeBuf = extractBufioWriterBuf(c.bw, c.rwc)
+ }
+
+ if c.flate() && c.flateThreshold == 0 {
+ c.flateThreshold = 128
+ if !c.msgWriterState.flateContextTakeover() {
+ c.flateThreshold = 512
+ }
+ }
+
+ runtime.SetFinalizer(c, func(c *Conn) {
+ c.close(errors.New("connection garbage collected"))
+ })
+
+ go c.timeoutLoop()
+
+ return c
+}
+
+// Subprotocol returns the negotiated subprotocol.
+// An empty string means the default protocol.
+func (c *Conn) Subprotocol() string {
+ return c.subprotocol
+}
+
+func (c *Conn) close(err error) {
+ c.closeMu.Lock()
+ defer c.closeMu.Unlock()
+
+ if c.isClosed() {
+ return
+ }
+ c.setCloseErrLocked(err)
+ close(c.closed)
+ runtime.SetFinalizer(c, nil)
+
+ // Have to close after c.closed is closed to ensure any goroutine that wakes up
+ // from the connection being closed also sees that c.closed is closed and returns
+ // closeErr.
+ c.rwc.Close()
+
+ go func() {
+ c.msgWriterState.close()
+
+ c.msgReader.close()
+ }()
+}
+
+func (c *Conn) timeoutLoop() {
+ readCtx := context.Background()
+ writeCtx := context.Background()
+
+ for {
+ select {
+ case <-c.closed:
+ return
+
+ case writeCtx = <-c.writeTimeout:
+ case readCtx = <-c.readTimeout:
+
+ case <-readCtx.Done():
+ c.setCloseErr(fmt.Errorf("read timed out: %w", readCtx.Err()))
+ go c.writeError(StatusPolicyViolation, errors.New("timed out"))
+ case <-writeCtx.Done():
+ c.close(fmt.Errorf("write timed out: %w", writeCtx.Err()))
+ return
+ }
+ }
+}
+
+func (c *Conn) flate() bool {
+ return c.copts != nil
+}
+
+// Ping sends a ping to the peer and waits for a pong.
+// Use this to measure latency or ensure the peer is responsive.
+// Ping must be called concurrently with Reader as it does
+// not read from the connection but instead waits for a Reader call
+// to read the pong.
+//
+// TCP Keepalives should suffice for most use cases.
+func (c *Conn) Ping(ctx context.Context) error {
+ p := atomic.AddInt32(&c.pingCounter, 1)
+
+ err := c.ping(ctx, strconv.Itoa(int(p)))
+ if err != nil {
+ return fmt.Errorf("failed to ping: %w", err)
+ }
+ return nil
+}
+
+func (c *Conn) ping(ctx context.Context, p string) error {
+ pong := make(chan struct{})
+
+ c.activePingsMu.Lock()
+ c.activePings[p] = pong
+ c.activePingsMu.Unlock()
+
+ defer func() {
+ c.activePingsMu.Lock()
+ delete(c.activePings, p)
+ c.activePingsMu.Unlock()
+ }()
+
+ err := c.writeControl(ctx, opPing, []byte(p))
+ if err != nil {
+ return err
+ }
+
+ select {
+ case <-c.closed:
+ return c.closeErr
+ case <-ctx.Done():
+ err := fmt.Errorf("failed to wait for pong: %w", ctx.Err())
+ c.close(err)
+ return err
+ case <-pong:
+ return nil
+ }
+}
+
+type mu struct {
+ c *Conn
+ ch chan struct{}
+}
+
+func newMu(c *Conn) *mu {
+ return &mu{
+ c: c,
+ ch: make(chan struct{}, 1),
+ }
+}
+
+func (m *mu) forceLock() {
+ m.ch <- struct{}{}
+}
+
+func (m *mu) lock(ctx context.Context) error {
+ select {
+ case <-m.c.closed:
+ return m.c.closeErr
+ case <-ctx.Done():
+ err := fmt.Errorf("failed to acquire lock: %w", ctx.Err())
+ m.c.close(err)
+ return err
+ case m.ch <- struct{}{}:
+ // To make sure the connection is certainly alive.
+ // As it's possible the send on m.ch was selected
+ // over the receive on closed.
+ select {
+ case <-m.c.closed:
+ // Make sure to release.
+ m.unlock()
+ return m.c.closeErr
+ default:
+ }
+ return nil
+ }
+}
+
+func (m *mu) unlock() {
+ select {
+ case <-m.ch:
+ default:
+ }
+}
diff --git a/conn_notjs.go b/conn_notjs.go
deleted file mode 100644
index bb2eb22..0000000
--- a/conn_notjs.go
+++ /dev/null
@@ -1,265 +0,0 @@
-// +build !js
-
-package websocket
-
-import (
- "bufio"
- "context"
- "errors"
- "fmt"
- "io"
- "runtime"
- "strconv"
- "sync"
- "sync/atomic"
-)
-
-// Conn represents a WebSocket connection.
-// All methods may be called concurrently except for Reader and Read.
-//
-// You must always read from the connection. Otherwise control
-// frames will not be handled. See Reader and CloseRead.
-//
-// Be sure to call Close on the connection when you
-// are finished with it to release associated resources.
-//
-// On any error from any method, the connection is closed
-// with an appropriate reason.
-type Conn struct {
- subprotocol string
- rwc io.ReadWriteCloser
- client bool
- copts *compressionOptions
- flateThreshold int
- br *bufio.Reader
- bw *bufio.Writer
-
- readTimeout chan context.Context
- writeTimeout chan context.Context
-
- // Read state.
- readMu *mu
- readHeaderBuf [8]byte
- readControlBuf [maxControlPayload]byte
- msgReader *msgReader
- readCloseFrameErr error
-
- // Write state.
- msgWriterState *msgWriterState
- writeFrameMu *mu
- writeBuf []byte
- writeHeaderBuf [8]byte
- writeHeader header
-
- closed chan struct{}
- closeMu sync.Mutex
- closeErr error
- wroteClose bool
-
- pingCounter int32
- activePingsMu sync.Mutex
- activePings map[string]chan<- struct{}
-}
-
-type connConfig struct {
- subprotocol string
- rwc io.ReadWriteCloser
- client bool
- copts *compressionOptions
- flateThreshold int
-
- br *bufio.Reader
- bw *bufio.Writer
-}
-
-func newConn(cfg connConfig) *Conn {
- c := &Conn{
- subprotocol: cfg.subprotocol,
- rwc: cfg.rwc,
- client: cfg.client,
- copts: cfg.copts,
- flateThreshold: cfg.flateThreshold,
-
- br: cfg.br,
- bw: cfg.bw,
-
- readTimeout: make(chan context.Context),
- writeTimeout: make(chan context.Context),
-
- closed: make(chan struct{}),
- activePings: make(map[string]chan<- struct{}),
- }
-
- c.readMu = newMu(c)
- c.writeFrameMu = newMu(c)
-
- c.msgReader = newMsgReader(c)
-
- c.msgWriterState = newMsgWriterState(c)
- if c.client {
- c.writeBuf = extractBufioWriterBuf(c.bw, c.rwc)
- }
-
- if c.flate() && c.flateThreshold == 0 {
- c.flateThreshold = 128
- if !c.msgWriterState.flateContextTakeover() {
- c.flateThreshold = 512
- }
- }
-
- runtime.SetFinalizer(c, func(c *Conn) {
- c.close(errors.New("connection garbage collected"))
- })
-
- go c.timeoutLoop()
-
- return c
-}
-
-// Subprotocol returns the negotiated subprotocol.
-// An empty string means the default protocol.
-func (c *Conn) Subprotocol() string {
- return c.subprotocol
-}
-
-func (c *Conn) close(err error) {
- c.closeMu.Lock()
- defer c.closeMu.Unlock()
-
- if c.isClosed() {
- return
- }
- c.setCloseErrLocked(err)
- close(c.closed)
- runtime.SetFinalizer(c, nil)
-
- // Have to close after c.closed is closed to ensure any goroutine that wakes up
- // from the connection being closed also sees that c.closed is closed and returns
- // closeErr.
- c.rwc.Close()
-
- go func() {
- c.msgWriterState.close()
-
- c.msgReader.close()
- }()
-}
-
-func (c *Conn) timeoutLoop() {
- readCtx := context.Background()
- writeCtx := context.Background()
-
- for {
- select {
- case <-c.closed:
- return
-
- case writeCtx = <-c.writeTimeout:
- case readCtx = <-c.readTimeout:
-
- case <-readCtx.Done():
- c.setCloseErr(fmt.Errorf("read timed out: %w", readCtx.Err()))
- go c.writeError(StatusPolicyViolation, errors.New("timed out"))
- case <-writeCtx.Done():
- c.close(fmt.Errorf("write timed out: %w", writeCtx.Err()))
- return
- }
- }
-}
-
-func (c *Conn) flate() bool {
- return c.copts != nil
-}
-
-// Ping sends a ping to the peer and waits for a pong.
-// Use this to measure latency or ensure the peer is responsive.
-// Ping must be called concurrently with Reader as it does
-// not read from the connection but instead waits for a Reader call
-// to read the pong.
-//
-// TCP Keepalives should suffice for most use cases.
-func (c *Conn) Ping(ctx context.Context) error {
- p := atomic.AddInt32(&c.pingCounter, 1)
-
- err := c.ping(ctx, strconv.Itoa(int(p)))
- if err != nil {
- return fmt.Errorf("failed to ping: %w", err)
- }
- return nil
-}
-
-func (c *Conn) ping(ctx context.Context, p string) error {
- pong := make(chan struct{})
-
- c.activePingsMu.Lock()
- c.activePings[p] = pong
- c.activePingsMu.Unlock()
-
- defer func() {
- c.activePingsMu.Lock()
- delete(c.activePings, p)
- c.activePingsMu.Unlock()
- }()
-
- err := c.writeControl(ctx, opPing, []byte(p))
- if err != nil {
- return err
- }
-
- select {
- case <-c.closed:
- return c.closeErr
- case <-ctx.Done():
- err := fmt.Errorf("failed to wait for pong: %w", ctx.Err())
- c.close(err)
- return err
- case <-pong:
- return nil
- }
-}
-
-type mu struct {
- c *Conn
- ch chan struct{}
-}
-
-func newMu(c *Conn) *mu {
- return &mu{
- c: c,
- ch: make(chan struct{}, 1),
- }
-}
-
-func (m *mu) forceLock() {
- m.ch <- struct{}{}
-}
-
-func (m *mu) lock(ctx context.Context) error {
- select {
- case <-m.c.closed:
- return m.c.closeErr
- case <-ctx.Done():
- err := fmt.Errorf("failed to acquire lock: %w", ctx.Err())
- m.c.close(err)
- return err
- case m.ch <- struct{}{}:
- // To make sure the connection is certainly alive.
- // As it's possible the send on m.ch was selected
- // over the receive on closed.
- select {
- case <-m.c.closed:
- // Make sure to release.
- m.unlock()
- return m.c.closeErr
- default:
- }
- return nil
- }
-}
-
-func (m *mu) unlock() {
- select {
- case <-m.ch:
- default:
- }
-}
diff --git a/ws_js.go b/ws_js.go
index b87e32c..31e3c2f 100644
--- a/ws_js.go
+++ b/ws_js.go
@@ -377,3 +377,137 @@ func (c *Conn) isClosed() bool {
return false
}
}
+
+// AcceptOptions represents Accept's options.
+type AcceptOptions struct {
+ Subprotocols []string
+ InsecureSkipVerify bool
+ OriginPatterns []string
+ CompressionMode CompressionMode
+ CompressionThreshold int
+}
+
+// Accept is stubbed out for Wasm.
+func Accept(w http.ResponseWriter, r *http.Request, opts *AcceptOptions) (*Conn, error) {
+ return nil, errors.New("unimplemented")
+}
+
+// StatusCode represents a WebSocket status code.
+// https://tools.ietf.org/html/rfc6455#section-7.4
+type StatusCode int
+
+// https://www.iana.org/assignments/websocket/websocket.xhtml#close-code-number
+//
+// These are only the status codes defined by the protocol.
+//
+// You can define custom codes in the 3000-4999 range.
+// The 3000-3999 range is reserved for use by libraries, frameworks and applications.
+// The 4000-4999 range is reserved for private use.
+const (
+ StatusNormalClosure StatusCode = 1000
+ StatusGoingAway StatusCode = 1001
+ StatusProtocolError StatusCode = 1002
+ StatusUnsupportedData StatusCode = 1003
+
+ // 1004 is reserved and so unexported.
+ statusReserved StatusCode = 1004
+
+ // StatusNoStatusRcvd cannot be sent in a close message.
+ // It is reserved for when a close message is received without
+ // a status code.
+ StatusNoStatusRcvd StatusCode = 1005
+
+ // StatusAbnormalClosure is exported for use only with Wasm.
+ // In non Wasm Go, the returned error will indicate whether the
+ // connection was closed abnormally.
+ StatusAbnormalClosure StatusCode = 1006
+
+ StatusInvalidFramePayloadData StatusCode = 1007
+ StatusPolicyViolation StatusCode = 1008
+ StatusMessageTooBig StatusCode = 1009
+ StatusMandatoryExtension StatusCode = 1010
+ StatusInternalError StatusCode = 1011
+ StatusServiceRestart StatusCode = 1012
+ StatusTryAgainLater StatusCode = 1013
+ StatusBadGateway StatusCode = 1014
+
+ // StatusTLSHandshake is only exported for use with Wasm.
+ // In non Wasm Go, the returned error will indicate whether there was
+ // a TLS handshake failure.
+ StatusTLSHandshake StatusCode = 1015
+)
+
+// CloseError is returned when the connection is closed with a status and reason.
+//
+// Use Go 1.13's errors.As to check for this error.
+// Also see the CloseStatus helper.
+type CloseError struct {
+ Code StatusCode
+ Reason string
+}
+
+func (ce CloseError) Error() string {
+ return fmt.Sprintf("status = %v and reason = %q", ce.Code, ce.Reason)
+}
+
+// CloseStatus is a convenience wrapper around Go 1.13's errors.As to grab
+// the status code from a CloseError.
+//
+// -1 will be returned if the passed error is nil or not a CloseError.
+func CloseStatus(err error) StatusCode {
+ var ce CloseError
+ if errors.As(err, &ce) {
+ return ce.Code
+ }
+ return -1
+}
+
+// CompressionMode represents the modes available to the deflate extension.
+// See https://tools.ietf.org/html/rfc7692
+//
+// A compatibility layer is implemented for the older deflate-frame extension used
+// by safari. See https://tools.ietf.org/html/draft-tyoshino-hybi-websocket-perframe-deflate-06
+// It will work the same in every way except that we cannot signal to the peer we
+// want to use no context takeover on our side, we can only signal that they should.
+// It is however currently disabled due to Safari bugs. See https://github.com/nhooyr/websocket/issues/218
+type CompressionMode int
+
+const (
+ // CompressionNoContextTakeover grabs a new flate.Reader and flate.Writer as needed
+ // for every message. This applies to both server and client side.
+ //
+ // This means less efficient compression as the sliding window from previous messages
+ // will not be used but the memory overhead will be lower if the connections
+ // are long lived and seldom used.
+ //
+ // The message will only be compressed if greater than 512 bytes.
+ CompressionNoContextTakeover CompressionMode = iota
+
+ // CompressionContextTakeover uses a flate.Reader and flate.Writer per connection.
+ // This enables reusing the sliding window from previous messages.
+ // As most WebSocket protocols are repetitive, this can be very efficient.
+ // It carries an overhead of 8 kB for every connection compared to CompressionNoContextTakeover.
+ //
+ // If the peer negotiates NoContextTakeover on the client or server side, it will be
+ // used instead as this is required by the RFC.
+ CompressionContextTakeover
+
+ // CompressionDisabled disables the deflate extension.
+ //
+ // Use this if you are using a predominantly binary protocol with very
+ // little duplication in between messages or CPU and memory are more
+ // important than bandwidth.
+ CompressionDisabled
+)
+
+// MessageType represents the type of a WebSocket message.
+// See https://tools.ietf.org/html/rfc6455#section-5.6
+type MessageType int
+
+// MessageType constants.
+const (
+ // MessageText is for UTF-8 encoded text messages like JSON.
+ MessageText MessageType = iota + 1
+ // MessageBinary is for binary messages like protobufs.
+ MessageBinary
+)
--
GitLab