internal/xsync/go_test.go

@@ -3,7 +3,7 @@ package xsync
 import (
 	"testing"
 
-	"nhooyr.io/websocket/internal/test/assert"
+	"github.com/coder/websocket/internal/test/assert"
 )
 
 func TestGoRecover(t *testing.T) {

internal/xsync/int64.go

-package xsync
-
-import (
-	"sync/atomic"
-)
-
-// Int64 represents an atomic int64.
-type Int64 struct {
-	// We do not use atomic.Load/StoreInt64 since it does not
-	// work on 32 bit computers but we need 64 bit integers.
-	i atomic.Value
-}
-
-// Load loads the int64.
-func (v *Int64) Load() int64 {
-	i, _ := v.i.Load().(int64)
-	return i
-}
-
-// Store stores the int64.
-func (v *Int64) Store(i int64) {
-	v.i.Store(i)
-}

main_test.go

+package websocket_test
+
+import (
+	"fmt"
+	"os"
+	"runtime"
+	"testing"
+)
+
+func goroutineStacks() []byte {
+	buf := make([]byte, 512)
+	for {
+		m := runtime.Stack(buf, true)
+		if m < len(buf) {
+			return buf[:m]
+		}
+		buf = make([]byte, len(buf)*2)
+	}
+}
+
+func TestMain(m *testing.M) {
+	code := m.Run()
+	if runtime.GOOS != "js" && runtime.NumGoroutine() != 1 ||
+		runtime.GOOS == "js" && runtime.NumGoroutine() != 2 {
+		fmt.Fprintf(os.Stderr, "goroutine leak detected, expected 1 but got %d goroutines\n", runtime.NumGoroutine())
+		fmt.Fprintf(os.Stderr, "%s\n", goroutineStacks())
+		os.Exit(1)
+	}
+	os.Exit(code)
+}

make.sh

-#!/bin/sh
-set -eu
-cd -- "$(dirname "$0")"
-
-./ci/fmt.sh
-./ci/lint.sh
-./ci/test.sh

mask.go

+package websocket
+
+import (
+	"encoding/binary"
+	"math/bits"
+)
+
+// maskGo applies the WebSocket masking algorithm to p
+// with the given key.
+// See https://tools.ietf.org/html/rfc6455#section-5.3
+//
+// The returned value is the correctly rotated key to
+// to continue to mask/unmask the message.
+//
+// It is optimized for LittleEndian and expects the key
+// to be in little endian.
+//
+// See https://github.com/golang/go/issues/31586
+func maskGo(b []byte, key uint32) uint32 {
+	if len(b) >= 8 {
+		key64 := uint64(key)<<32 | uint64(key)
+
+		// At some point in the future we can clean these unrolled loops up.
+		// See https://github.com/golang/go/issues/31586#issuecomment-487436401
+
+		// Then we xor until b is less than 128 bytes.
+		for len(b) >= 128 {
+			v := binary.LittleEndian.Uint64(b)
+			binary.LittleEndian.PutUint64(b, v^key64)
+			v = binary.LittleEndian.Uint64(b[8:16])
+			binary.LittleEndian.PutUint64(b[8:16], v^key64)
+			v = binary.LittleEndian.Uint64(b[16:24])
+			binary.LittleEndian.PutUint64(b[16:24], v^key64)
+			v = binary.LittleEndian.Uint64(b[24:32])
+			binary.LittleEndian.PutUint64(b[24:32], v^key64)
+			v = binary.LittleEndian.Uint64(b[32:40])
+			binary.LittleEndian.PutUint64(b[32:40], v^key64)
+			v = binary.LittleEndian.Uint64(b[40:48])
+			binary.LittleEndian.PutUint64(b[40:48], v^key64)
+			v = binary.LittleEndian.Uint64(b[48:56])
+			binary.LittleEndian.PutUint64(b[48:56], v^key64)
+			v = binary.LittleEndian.Uint64(b[56:64])
+			binary.LittleEndian.PutUint64(b[56:64], v^key64)
+			v = binary.LittleEndian.Uint64(b[64:72])
+			binary.LittleEndian.PutUint64(b[64:72], v^key64)
+			v = binary.LittleEndian.Uint64(b[72:80])
+			binary.LittleEndian.PutUint64(b[72:80], v^key64)
+			v = binary.LittleEndian.Uint64(b[80:88])
+			binary.LittleEndian.PutUint64(b[80:88], v^key64)
+			v = binary.LittleEndian.Uint64(b[88:96])
+			binary.LittleEndian.PutUint64(b[88:96], v^key64)
+			v = binary.LittleEndian.Uint64(b[96:104])
+			binary.LittleEndian.PutUint64(b[96:104], v^key64)
+			v = binary.LittleEndian.Uint64(b[104:112])
+			binary.LittleEndian.PutUint64(b[104:112], v^key64)
+			v = binary.LittleEndian.Uint64(b[112:120])
+			binary.LittleEndian.PutUint64(b[112:120], v^key64)
+			v = binary.LittleEndian.Uint64(b[120:128])
+			binary.LittleEndian.PutUint64(b[120:128], v^key64)
+			b = b[128:]
+		}
+
+		// Then we xor until b is less than 64 bytes.
+		for len(b) >= 64 {
+			v := binary.LittleEndian.Uint64(b)
+			binary.LittleEndian.PutUint64(b, v^key64)
+			v = binary.LittleEndian.Uint64(b[8:16])
+			binary.LittleEndian.PutUint64(b[8:16], v^key64)
+			v = binary.LittleEndian.Uint64(b[16:24])
+			binary.LittleEndian.PutUint64(b[16:24], v^key64)
+			v = binary.LittleEndian.Uint64(b[24:32])
+			binary.LittleEndian.PutUint64(b[24:32], v^key64)
+			v = binary.LittleEndian.Uint64(b[32:40])
+			binary.LittleEndian.PutUint64(b[32:40], v^key64)
+			v = binary.LittleEndian.Uint64(b[40:48])
+			binary.LittleEndian.PutUint64(b[40:48], v^key64)
+			v = binary.LittleEndian.Uint64(b[48:56])
+			binary.LittleEndian.PutUint64(b[48:56], v^key64)
+			v = binary.LittleEndian.Uint64(b[56:64])
+			binary.LittleEndian.PutUint64(b[56:64], v^key64)
+			b = b[64:]
+		}
+
+		// Then we xor until b is less than 32 bytes.
+		for len(b) >= 32 {
+			v := binary.LittleEndian.Uint64(b)
+			binary.LittleEndian.PutUint64(b, v^key64)
+			v = binary.LittleEndian.Uint64(b[8:16])
+			binary.LittleEndian.PutUint64(b[8:16], v^key64)
+			v = binary.LittleEndian.Uint64(b[16:24])
+			binary.LittleEndian.PutUint64(b[16:24], v^key64)
+			v = binary.LittleEndian.Uint64(b[24:32])
+			binary.LittleEndian.PutUint64(b[24:32], v^key64)
+			b = b[32:]
+		}
+
+		// Then we xor until b is less than 16 bytes.
+		for len(b) >= 16 {
+			v := binary.LittleEndian.Uint64(b)
+			binary.LittleEndian.PutUint64(b, v^key64)
+			v = binary.LittleEndian.Uint64(b[8:16])
+			binary.LittleEndian.PutUint64(b[8:16], v^key64)
+			b = b[16:]
+		}
+
+		// Then we xor until b is less than 8 bytes.
+		for len(b) >= 8 {
+			v := binary.LittleEndian.Uint64(b)
+			binary.LittleEndian.PutUint64(b, v^key64)
+			b = b[8:]
+		}
+	}
+
+	// Then we xor until b is less than 4 bytes.
+	for len(b) >= 4 {
+		v := binary.LittleEndian.Uint32(b)
+		binary.LittleEndian.PutUint32(b, v^key)
+		b = b[4:]
+	}
+
+	// xor remaining bytes.
+	for i := range b {
+		b[i] ^= byte(key)
+		key = bits.RotateLeft32(key, -8)
+	}
+
+	return key
+}

mask_amd64.s

+#include "textflag.h"
+
+// func maskAsm(b *byte, len int, key uint32)
+TEXT ·maskAsm(SB), NOSPLIT, $0-28
+	// AX = b
+	// CX = len (left length)
+	// SI = key (uint32)
+	// DI = uint64(SI) | uint64(SI)<<32
+	MOVQ b+0(FP), AX
+	MOVQ len+8(FP), CX
+	MOVL key+16(FP), SI
+
+	// calculate the DI
+	// DI = SI<<32 | SI
+	MOVL SI, DI
+	MOVQ DI, DX
+	SHLQ $32, DI
+	ORQ  DX, DI
+
+	CMPQ  CX, $15
+	JLE   less_than_16
+	CMPQ  CX, $63
+	JLE   less_than_64
+	CMPQ  CX, $128
+	JLE   sse
+	TESTQ $31, AX
+	JNZ   unaligned
+
+unaligned_loop_1byte:
+	XORB  SI, (AX)
+	INCQ  AX
+	DECQ  CX
+	ROLL  $24, SI
+	TESTQ $7, AX
+	JNZ   unaligned_loop_1byte
+
+	// calculate DI again since SI was modified
+	// DI = SI<<32 | SI
+	MOVL SI, DI
+	MOVQ DI, DX
+	SHLQ $32, DI
+	ORQ  DX, DI
+
+	TESTQ $31, AX
+	JZ    sse
+
+unaligned:
+	TESTQ $7, AX               // AND $7 & len, if not zero jump to loop_1b.
+	JNZ   unaligned_loop_1byte
+
+unaligned_loop:
+	// we don't need to check the CX since we know it's above 128
+	XORQ  DI, (AX)
+	ADDQ  $8, AX
+	SUBQ  $8, CX
+	TESTQ $31, AX
+	JNZ   unaligned_loop
+	JMP   sse
+
+sse:
+	CMPQ       CX, $0x40
+	JL         less_than_64
+	MOVQ       DI, X0
+	PUNPCKLQDQ X0, X0
+
+sse_loop:
+	MOVOU 0*16(AX), X1
+	MOVOU 1*16(AX), X2
+	MOVOU 2*16(AX), X3
+	MOVOU 3*16(AX), X4
+	PXOR  X0, X1
+	PXOR  X0, X2
+	PXOR  X0, X3
+	PXOR  X0, X4
+	MOVOU X1, 0*16(AX)
+	MOVOU X2, 1*16(AX)
+	MOVOU X3, 2*16(AX)
+	MOVOU X4, 3*16(AX)
+	ADDQ  $0x40, AX
+	SUBQ  $0x40, CX
+	CMPQ  CX, $0x40
+	JAE   sse_loop
+
+less_than_64:
+	TESTQ $32, CX
+	JZ    less_than_32
+	XORQ  DI, (AX)
+	XORQ  DI, 8(AX)
+	XORQ  DI, 16(AX)
+	XORQ  DI, 24(AX)
+	ADDQ  $32, AX
+
+less_than_32:
+	TESTQ $16, CX
+	JZ    less_than_16
+	XORQ  DI, (AX)
+	XORQ  DI, 8(AX)
+	ADDQ  $16, AX
+
+less_than_16:
+	TESTQ $8, CX
+	JZ    less_than_8
+	XORQ  DI, (AX)
+	ADDQ  $8, AX
+
+less_than_8:
+	TESTQ $4, CX
+	JZ    less_than_4
+	XORL  SI, (AX)
+	ADDQ  $4, AX
+
+less_than_4:
+	TESTQ $2, CX
+	JZ    less_than_2
+	XORW  SI, (AX)
+	ROLL  $16, SI
+	ADDQ  $2, AX
+
+less_than_2:
+	TESTQ $1, CX
+	JZ    done
+	XORB  SI, (AX)
+	ROLL  $24, SI
+
+done:
+	MOVL SI, ret+24(FP)
+	RET

mask_arm64.s

+#include "textflag.h"
+
+// func maskAsm(b *byte, len int, key uint32)
+TEXT ·maskAsm(SB), NOSPLIT, $0-28
+	// R0 = b
+	// R1 = len
+	// R3 = key (uint32)
+	// R2 = uint64(key)<<32 | uint64(key)
+	MOVD  b_ptr+0(FP), R0
+	MOVD  b_len+8(FP), R1
+	MOVWU key+16(FP), R3
+	MOVD  R3, R2
+	ORR   R2<<32, R2, R2
+	VDUP  R2, V0.D2
+	CMP   $64, R1
+	BLT   less_than_64
+
+loop_64:
+	VLD1   (R0), [V1.B16, V2.B16, V3.B16, V4.B16]
+	VEOR   V1.B16, V0.B16, V1.B16
+	VEOR   V2.B16, V0.B16, V2.B16
+	VEOR   V3.B16, V0.B16, V3.B16
+	VEOR   V4.B16, V0.B16, V4.B16
+	VST1.P [V1.B16, V2.B16, V3.B16, V4.B16], 64(R0)
+	SUBS   $64, R1
+	CMP    $64, R1
+	BGE    loop_64
+
+less_than_64:
+	CBZ    R1, end
+	TBZ    $5, R1, less_than_32
+	VLD1   (R0), [V1.B16, V2.B16]
+	VEOR   V1.B16, V0.B16, V1.B16
+	VEOR   V2.B16, V0.B16, V2.B16
+	VST1.P [V1.B16, V2.B16], 32(R0)
+
+less_than_32:
+	TBZ   $4, R1, less_than_16
+	LDP   (R0), (R11, R12)
+	EOR   R11, R2, R11
+	EOR   R12, R2, R12
+	STP.P (R11, R12), 16(R0)
+
+less_than_16:
+	TBZ    $3, R1, less_than_8
+	MOVD   (R0), R11
+	EOR    R2, R11, R11
+	MOVD.P R11, 8(R0)
+
+less_than_8:
+	TBZ     $2, R1, less_than_4
+	MOVWU   (R0), R11
+	EORW    R2, R11, R11
+	MOVWU.P R11, 4(R0)
+
+less_than_4:
+	TBZ     $1, R1, less_than_2
+	MOVHU   (R0), R11
+	EORW    R3, R11, R11
+	MOVHU.P R11, 2(R0)
+	RORW    $16, R3
+
+less_than_2:
+	TBZ     $0, R1, end
+	MOVBU   (R0), R11
+	EORW    R3, R11, R11
+	MOVBU.P R11, 1(R0)
+	RORW    $8, R3
+
+end:
+	MOVWU R3, ret+24(FP)
+	RET

mask_asm.go

+//go:build amd64 || arm64
+
+package websocket
+
+func mask(b []byte, key uint32) uint32 {
+	// TODO: Will enable in v1.9.0.
+	return maskGo(b, key)
+	/*
+		if len(b) > 0 {
+			return maskAsm(&b[0], len(b), key)
+		}
+		return key
+	*/
+}
+
+// @nhooyr: I am not confident that the amd64 or the arm64 implementations of this
+// function are perfect. There are almost certainly missing optimizations or
+// opportunities for simplification. I'm confident there are no bugs though.
+// For example, the arm64 implementation doesn't align memory like the amd64.
+// Or the amd64 implementation could use AVX512 instead of just AVX2.
+// The AVX2 code I had to disable anyway as it wasn't performing as expected.
+// See https://github.com/nhooyr/websocket/pull/326#issuecomment-1771138049
+//
+//go:noescape
+//lint:ignore U1000 disabled till v1.9.0
+func maskAsm(b *byte, len int, key uint32) uint32

mask_asm_test.go

+//go:build amd64 || arm64
+
+package websocket
+
+import "testing"
+
+func TestMaskASM(t *testing.T) {
+	t.Parallel()
+
+	testMask(t, "maskASM", mask)
+}

mask_go.go

+//go:build !amd64 && !arm64 && !js
+
+package websocket
+
+func mask(b []byte, key uint32) uint32 {
+	return maskGo(b, key)
+}

mask_test.go

+package websocket
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/binary"
+	"math/big"
+	"math/bits"
+	"testing"
+
+	"github.com/coder/websocket/internal/test/assert"
+)
+
+func basicMask(b []byte, key uint32) uint32 {
+	for i := range b {
+		b[i] ^= byte(key)
+		key = bits.RotateLeft32(key, -8)
+	}
+	return key
+}
+
+func basicMask2(b []byte, key uint32) uint32 {
+	keyb := binary.LittleEndian.AppendUint32(nil, key)
+	pos := 0
+	for i := range b {
+		b[i] ^= keyb[pos&3]
+		pos++
+	}
+	return bits.RotateLeft32(key, (pos&3)*-8)
+}
+
+func TestMask(t *testing.T) {
+	t.Parallel()
+
+	testMask(t, "basicMask", basicMask)
+	testMask(t, "maskGo", maskGo)
+	testMask(t, "basicMask2", basicMask2)
+}
+
+func testMask(t *testing.T, name string, fn func(b []byte, key uint32) uint32) {
+	t.Run(name, func(t *testing.T) {
+		t.Parallel()
+		for i := 0; i < 9999; i++ {
+			keyb := make([]byte, 4)
+			_, err := rand.Read(keyb)
+			assert.Success(t, err)
+			key := binary.LittleEndian.Uint32(keyb)
+
+			n, err := rand.Int(rand.Reader, big.NewInt(1<<16))
+			assert.Success(t, err)
+
+			b := make([]byte, 1+n.Int64())
+			_, err = rand.Read(b)
+			assert.Success(t, err)
+
+			b2 := make([]byte, len(b))
+			copy(b2, b)
+			b3 := make([]byte, len(b))
+			copy(b3, b)
+
+			key2 := basicMask(b2, key)
+			key3 := fn(b3, key)
+
+			if key2 != key3 {
+				t.Errorf("expected key %X but got %X", key2, key3)
+			}
+			if !bytes.Equal(b2, b3) {
+				t.Error("bad bytes")
+				return
+			}
+		}
+	})
+}

netconn.go

@@ -33,8 +33,13 @@ import (
 // where only the reading/writing goroutines are interrupted but the connection
 // is kept alive.
 //
-// The Addr methods will return a mock net.Addr that returns "websocket" for Network
-// and "websocket/unknown-addr" for String.
+// The Addr methods will return the real addresses for connections obtained
+// from websocket.Accept. But for connections obtained from websocket.Dial, a mock net.Addr
+// will be returned that gives "websocket" for Network() and "websocket/unknown-addr" for
+// String(). This is because websocket.Dial only exposes a io.ReadWriteCloser instead of the
+// full net.Conn to us.
+//
+// When running as WASM, the Addr methods will always return the mock address described above.
 //
 // A received StatusNormalClosure or StatusGoingAway close frame will be translated to
 // io.EOF when reading.
@@ -63,7 +68,7 @@ func NetConn(ctx context.Context, c *Conn, msgType MessageType) net.Conn {
 		defer nc.writeMu.unlock()
 
 		// Prevents future writes from writing until the deadline is reset.
-		atomic.StoreInt64(&nc.writeExpired, 1)
+		nc.writeExpired.Store(1)
 	})
 	if !nc.writeTimer.Stop() {
 		<-nc.writeTimer.C
@@ -79,7 +84,7 @@ func NetConn(ctx context.Context, c *Conn, msgType MessageType) net.Conn {
 		defer nc.readMu.unlock()
 
 		// Prevents future reads from reading until the deadline is reset.
-		atomic.StoreInt64(&nc.readExpired, 1)
+		nc.readExpired.Store(1)
 	})
 	if !nc.readTimer.Stop() {
 		<-nc.readTimer.C
@@ -94,13 +99,13 @@ type netConn struct {
 
 	writeTimer   *time.Timer
 	writeMu      *mu
-	writeExpired int64
+	writeExpired atomic.Int64
 	writeCtx     context.Context
 	writeCancel  context.CancelFunc
 
 	readTimer   *time.Timer
 	readMu      *mu
-	readExpired int64
+	readExpired atomic.Int64
 	readCtx     context.Context
 	readCancel  context.CancelFunc
 	readEOFed   bool
@@ -121,7 +126,7 @@ func (nc *netConn) Write(p []byte) (int, error) {
 	nc.writeMu.forceLock()
 	defer nc.writeMu.unlock()
 
-	if atomic.LoadInt64(&nc.writeExpired) == 1 {
+	if nc.writeExpired.Load() == 1 {
 		return 0, fmt.Errorf("failed to write: %w", context.DeadlineExceeded)
 	}
 
@@ -136,7 +141,20 @@ func (nc *netConn) Read(p []byte) (int, error) {
 	nc.readMu.forceLock()
 	defer nc.readMu.unlock()
 
-	if atomic.LoadInt64(&nc.readExpired) == 1 {
+	for {
+		n, err := nc.read(p)
+		if err != nil {
+			return n, err
+		}
+		if n == 0 {
+			continue
+		}
+		return n, nil
+	}
+}
+
+func (nc *netConn) read(p []byte) (int, error) {
+	if nc.readExpired.Load() == 1 {
 		return 0, fmt.Errorf("failed to read: %w", context.DeadlineExceeded)
 	}
 
@@ -181,14 +199,6 @@ func (a websocketAddr) String() string {
 	return "websocket/unknown-addr"
 }
 
-func (nc *netConn) RemoteAddr() net.Addr {
-	return websocketAddr{}
-}
-
-func (nc *netConn) LocalAddr() net.Addr {
-	return websocketAddr{}
-}
-
 func (nc *netConn) SetDeadline(t time.Time) error {
 	nc.SetWriteDeadline(t)
 	nc.SetReadDeadline(t)
@@ -196,21 +206,29 @@ func (nc *netConn) SetDeadline(t time.Time) error {
 }
 
 func (nc *netConn) SetWriteDeadline(t time.Time) error {
-	atomic.StoreInt64(&nc.writeExpired, 0)
+	nc.writeExpired.Store(0)
 	if t.IsZero() {
 		nc.writeTimer.Stop()
 	} else {
-		nc.writeTimer.Reset(t.Sub(time.Now()))
+		dur := time.Until(t)
+		if dur <= 0 {
+			dur = 1
+		}
+		nc.writeTimer.Reset(dur)
 	}
 	return nil
 }
 
 func (nc *netConn) SetReadDeadline(t time.Time) error {
-	atomic.StoreInt64(&nc.readExpired, 0)
+	nc.readExpired.Store(0)
 	if t.IsZero() {
 		nc.readTimer.Stop()
 	} else {
-		nc.readTimer.Reset(t.Sub(time.Now()))
+		dur := time.Until(t)
+		if dur <= 0 {
+			dur = 1
+		}
+		nc.readTimer.Reset(dur)
 	}
 	return nil
 }

netconn_js.go

+package websocket
+
+import "net"
+
+func (nc *netConn) RemoteAddr() net.Addr {
+	return websocketAddr{}
+}
+
+func (nc *netConn) LocalAddr() net.Addr {
+	return websocketAddr{}
+}

netconn_notjs.go

+//go:build !js
+// +build !js
+
+package websocket
+
+import "net"
+
+func (nc *netConn) RemoteAddr() net.Addr {
+	if unc, ok := nc.c.rwc.(net.Conn); ok {
+		return unc.RemoteAddr()
+	}
+	return websocketAddr{}
+}
+
+func (nc *netConn) LocalAddr() net.Addr {
+	if unc, ok := nc.c.rwc.(net.Conn); ok {
+		return unc.LocalAddr()
+	}
+	return websocketAddr{}
+}

ws_js_test.go

@@ -7,9 +7,9 @@ import (
 	"testing"
 	"time"
 
-	"nhooyr.io/websocket"
-	"nhooyr.io/websocket/internal/test/assert"
-	"nhooyr.io/websocket/internal/test/wstest"
+	"github.com/coder/websocket"
+	"github.com/coder/websocket/internal/test/assert"
+	"github.com/coder/websocket/internal/test/wstest"
 )
 
 func TestWasm(t *testing.T) {

wsjson/wsjson.go

 // Package wsjson provides helpers for reading and writing JSON messages.
-package wsjson // import "nhooyr.io/websocket/wsjson"
+package wsjson // import "github.com/coder/websocket/wsjson"
 
 import (
 	"context"
 	"encoding/json"
 	"fmt"
 
-	"nhooyr.io/websocket"
-	"nhooyr.io/websocket/internal/bpool"
-	"nhooyr.io/websocket/internal/errd"
+	"github.com/coder/websocket"
+	"github.com/coder/websocket/internal/bpool"
+	"github.com/coder/websocket/internal/errd"
+	"github.com/coder/websocket/internal/util"
 )
 
 // Read reads a JSON message from c into v.
@@ -51,17 +52,17 @@ func Write(ctx context.Context, c *websocket.Conn, v interface{}) error {
 func write(ctx context.Context, c *websocket.Conn, v interface{}) (err error) {
 	defer errd.Wrap(&err, "failed to write JSON message")
 
-	w, err := c.Writer(ctx, websocket.MessageText)
-	if err != nil {
-		return err
-	}
-
 	// json.Marshal cannot reuse buffers between calls as it has to return
 	// a copy of the byte slice but Encoder does as it directly writes to w.
-	err = json.NewEncoder(w).Encode(v)
+	err = json.NewEncoder(util.WriterFunc(func(p []byte) (int, error) {
+		err := c.Write(ctx, websocket.MessageText, p)
+		if err != nil {
+			return 0, err
+		}
+		return len(p), nil
+	})).Encode(v)
 	if err != nil {
 		return fmt.Errorf("failed to marshal JSON: %w", err)
 	}
-
-	return w.Close()
+	return nil
 }

wsjson/wsjson_test.go

+package wsjson_test
+
+import (
+	"encoding/json"
+	"io"
+	"strconv"
+	"testing"
+
+	"github.com/coder/websocket/internal/test/xrand"
+)
+
+func BenchmarkJSON(b *testing.B) {
+	sizes := []int{
+		8,
+		16,
+		32,
+		128,
+		256,
+		512,
+		1024,
+		2048,
+		4096,
+		8192,
+		16384,
+	}
+
+	b.Run("json.Encoder", func(b *testing.B) {
+		for _, size := range sizes {
+			b.Run(strconv.Itoa(size), func(b *testing.B) {
+				msg := xrand.String(size)
+				b.SetBytes(int64(size))
+				b.ReportAllocs()
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					json.NewEncoder(io.Discard).Encode(msg)
+				}
+			})
+		}
+	})
+	b.Run("json.Marshal", func(b *testing.B) {
+		for _, size := range sizes {
+			b.Run(strconv.Itoa(size), func(b *testing.B) {
+				msg := xrand.String(size)
+				b.SetBytes(int64(size))
+				b.ReportAllocs()
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					json.Marshal(msg)
+				}
+			})
+		}
+	})
+}