Apr 12, 2024 · Apr 9, 2024 · Apr 11, 2024 · Apr 9, 2024
diff --git a/conn.go b/conn.go
 closeMu sync.Mutex
 closing bool

 pingCounterint32
 pingCounteratomic.Int32
 activePingsMu sync.Mutex
 activePings   map[string]chan<- struct{}
 }
 //
 // TCP Keepalives should suffice for most use cases.
 func (c *Conn) Ping(ctx context.Context) error {
 p :=atomic.AddInt32(&c.pingCounter,1)
 p := c.pingCounter.Add(1)

 err := c.ping(ctx, strconv.Itoa(int(p)))
 if err != nil {
diff --git a/internal/xsync/int64.go b/internal/xsync/int64.go
diff --git a/netconn.go b/netconn.go
 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
 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
 }

 type netConn struct {
 // These must be first to be aligned on 32 bit platforms.
 // https://github.com/nhooyr/websocket/pull/438
 readExpired  int64
 writeExpired int64

 c       *Conn
 msgType MessageType

 writeTimer  *time.Timer
 writeMu     *mu
 writeCtx    context.Context
 writeCancel context.CancelFunc

 readTimer  *time.Timer
 readMu     *mu
 readCtx    context.Context
 readCancel context.CancelFunc
 readEOFed  bool
 reader     io.Reader
 writeTimer   *time.Timer
 writeMu      *mu
 writeExpired atomic.Int64
 writeCtx     context.Context
 writeCancel  context.CancelFunc

 readTimer   *time.Timer
 readMu      *mu
 readExpired atomic.Int64
 readCtx     context.Context
 readCancel  context.CancelFunc
 readEOFed   bool
 reader      io.Reader
 }

 var _ net.Conn = &netConn{}
 nc.writeMu.forceLock()
 defer nc.writeMu.unlock()

 ifatomic.LoadInt64(&nc.writeExpired) == 1 {
 if nc.writeExpired.Load() == 1 {
 return 0, fmt.Errorf("failed to write: %w", context.DeadlineExceeded)
 }

 }

 func (nc *netConn) read(p []byte) (int, error) {
 ifatomic.LoadInt64(&nc.readExpired) == 1 {
 if nc.readExpired.Load() == 1 {
 return 0, fmt.Errorf("failed to read: %w", context.DeadlineExceeded)
 }

 }

 func (nc *netConn) SetWriteDeadline(t time.Time) error {
 atomic.StoreInt64(&nc.writeExpired,0)
 nc.writeExpired.Store(0)
 if t.IsZero() {
 nc.writeTimer.Stop()
 } else {
 }

 func (nc *netConn) SetReadDeadline(t time.Time) error {
 atomic.StoreInt64(&nc.readExpired,0)
 nc.readExpired.Store(0)
 if t.IsZero() {
 nc.readTimer.Stop()
 } else {
diff --git a/read.go b/read.go
 "io"
 "net"
 "strings"
 "sync/atomic"
 "time"

 "nhooyr.io/websocket/internal/errd"
 "nhooyr.io/websocket/internal/util"
 "nhooyr.io/websocket/internal/xsync"
 )

 // Reader reads from the connection until there is a WebSocket
 type limitReader struct {
 c     *Conn
 r     io.Reader
 limitxsync.Int64
 limitatomic.Int64
 n     int64
 }

diff --git a/ws_js.go b/ws_js.go
 "runtime"
 "strings"
 "sync"
 "sync/atomic"
 "syscall/js"

 "nhooyr.io/websocket/internal/bpool"
 "nhooyr.io/websocket/internal/wsjs"
 "nhooyr.io/websocket/internal/xsync"
 )

 // opcode represents a WebSocket opcode.
 ws     wsjs.WebSocket

 // read limit for a message in bytes.
 msgReadLimitxsync.Int64
 msgReadLimitatomic.Int64

 closeReadMu  sync.Mutex
 closeReadCtx context.Context
Original file line number	Diff line number	Diff line change
Expand Up		@@ -77,7 +77,7 @@ type Conn struct {
		closeMu sync.Mutex
		closing bool

		pingCounterint32
		pingCounteratomic.Int32
Jacalz marked this conversation as resolved. OutdatedShow resolvedHide resolved
		activePingsMu sync.Mutex
		activePings map[string]chan<- struct{}
		}
Expand DownExpand Up		@@ -200,7 +200,7 @@ func (c *Conn) flate() bool {
		//
		// TCP Keepalives should suffice for most use cases.
		func (c *Conn) Ping(ctx context.Context) error {
		p :=atomic.AddInt32(&c.pingCounter,1)
		p := c.pingCounter.Add(1)

		err := c.ping(ctx, strconv.Itoa(int(p)))
		if err != nil {
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -68,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
Expand All		@@ -84,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
Expand All		@@ -94,25 +94,22 @@ func NetConn(ctx context.Context, c *Conn, msgType MessageType) net.Conn {
		}

		type netConn struct {
		// These must be first to be aligned on 32 bit platforms.
		// https://github.com/nhooyr/websocket/pull/438
		readExpired int64
		writeExpired int64

		c *Conn
		msgType MessageType

		writeTimer *time.Timer
		writeMu *mu
		writeCtx context.Context
		writeCancel context.CancelFunc

		readTimer *time.Timer
		readMu *mu
		readCtx context.Context
		readCancel context.CancelFunc
		readEOFed bool
		reader io.Reader
		writeTimer *time.Timer
		writeMu *mu
		writeExpired atomic.Int64
		writeCtx context.Context
		writeCancel context.CancelFunc

		readTimer *time.Timer
		readMu *mu
		readExpired atomic.Int64
		readCtx context.Context
		readCancel context.CancelFunc
		readEOFed bool
		reader io.Reader
		}

		var _ net.Conn = &netConn{}
Expand All		@@ -129,7 +126,7 @@ func (nc *netConn) Write(p []byte) (int, error) {
		nc.writeMu.forceLock()
		defer nc.writeMu.unlock()

		ifatomic.LoadInt64(&nc.writeExpired) == 1 {
		if nc.writeExpired.Load() == 1 {
		return 0, fmt.Errorf("failed to write: %w", context.DeadlineExceeded)
		}

Expand DownExpand Up		@@ -157,7 +154,7 @@ func (nc *netConn) Read(p []byte) (int, error) {
		}

		func (nc *netConn) read(p []byte) (int, error) {
		ifatomic.LoadInt64(&nc.readExpired) == 1 {
		if nc.readExpired.Load() == 1 {
		return 0, fmt.Errorf("failed to read: %w", context.DeadlineExceeded)
		}

Expand DownExpand Up		@@ -209,7 +206,7 @@ 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 {
Expand All		@@ -223,7 +220,7 @@ func (nc *netConn) SetWriteDeadline(t time.Time) error {
		}

		func (nc *netConn) SetReadDeadline(t time.Time) error {
		atomic.StoreInt64(&nc.readExpired,0)
		nc.readExpired.Store(0)
		if t.IsZero() {
		nc.readTimer.Stop()
		} else {
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -11,11 +11,11 @@ import (
		"io"
		"net"
		"strings"
		"sync/atomic"
		"time"

		"nhooyr.io/websocket/internal/errd"
		"nhooyr.io/websocket/internal/util"
		"nhooyr.io/websocket/internal/xsync"
		)

		// Reader reads from the connection until there is a WebSocket
Expand DownExpand Up		@@ -465,7 +465,7 @@ func (mr *msgReader) read(p []byte) (int, error) {
		type limitReader struct {
		c *Conn
		r io.Reader
		limitxsync.Int64
		limitatomic.Int64
		n int64
		}

Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -12,11 +12,11 @@ import (
		"runtime"
		"strings"
		"sync"
		"sync/atomic"
		"syscall/js"

		"nhooyr.io/websocket/internal/bpool"
		"nhooyr.io/websocket/internal/wsjs"
		"nhooyr.io/websocket/internal/xsync"
		)

		// opcode represents a WebSocket opcode.
Expand DownExpand Up		@@ -45,7 +45,7 @@ type Conn struct {
		ws wsjs.WebSocket

		// read limit for a message in bytes.
		msgReadLimitxsync.Int64
		msgReadLimitatomic.Int64

		closeReadMu sync.Mutex
		closeReadCtx context.Context
Expand Down