Add ping and pong received callbacks

This change adds two optional callbacks to both `DialOptions` and
`AcceptOptions`. These callbacks are invoked synchronously when a ping
or pong frame is received, allowing advanced users to log or inspect
payloads for metrics or debugging. If the callback needs to perform more
complex work or reuse the payload outside the callback, it is
recommended to clone the byte slice and/or perform processing in a
separate goroutine.

Tests confirm that the ping/pong callbacks are invoked as expected.

References #246

Author: igolaizola

accept.go

@@ -5,6 +5,7 @@ package websocket
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha1"
 	"encoding/base64"
 	"errors"
@@ -62,6 +63,18 @@ type AcceptOptions struct {
 	// Defaults to 512 bytes for CompressionNoContextTakeover and 128 bytes
 	// for CompressionContextTakeover.
 	CompressionThreshold int
+
+	// OnPingReceived is an optional callback invoked synchronously when a ping frame is received.
+	//
+	// To avoid blocking, process the callback asynchronously using a goroutine.
+	// If you need to reuse the payload outside the callback, clone the byte slice.
+	// Any modifications to the payload within the callback will be sent in the subsequent pong frame.
+	OnPingReceived func(context.Context, []byte)
+
+	// OnPongReceived is an optional callback invoked synchronously when a pong frame is received.
+	//
+	// To avoid blocking, process the callback asynchronously using a goroutine.
+	OnPongReceived func(context.Context, []byte)
 }
 
 func (opts *AcceptOptions) cloneWithDefaults() *AcceptOptions {
@@ -156,6 +169,8 @@ func accept(w http.ResponseWriter, r *http.Request, opts *AcceptOptions) (_ *Con
 		client:         false,
 		copts:          copts,
 		flateThreshold: opts.CompressionThreshold,
+		onPingReceived: opts.OnPingReceived,
+		onPongReceived: opts.OnPongReceived,
 
 		br: brw.Reader,
 		bw: brw.Writer,

conn.go

@@ -83,9 +83,11 @@ type Conn struct {
 	closeMu sync.Mutex // Protects following.
 	closed  chan struct{}
 
-	pingCounter   atomic.Int64
-	activePingsMu sync.Mutex
-	activePings   map[string]chan<- struct{}
+	pingCounter    atomic.Int64
+	activePingsMu  sync.Mutex
+	activePings    map[string]chan<- struct{}
+	onPingReceived func(context.Context, []byte)
+	onPongReceived func(context.Context, []byte)
 }
 
 type connConfig struct {
@@ -94,6 +96,8 @@ type connConfig struct {
 	client         bool
 	copts          *compressionOptions
 	flateThreshold int
+	onPingReceived func(context.Context, []byte)
+	onPongReceived func(context.Context, []byte)
 
 	br *bufio.Reader
 	bw *bufio.Writer
@@ -114,8 +118,10 @@ func newConn(cfg connConfig) *Conn {
 		writeTimeout:    make(chan context.Context),
 		timeoutLoopDone: make(chan struct{}),
 
-		closed:      make(chan struct{}),
-		activePings: make(map[string]chan<- struct{}),
+		closed:         make(chan struct{}),
+		activePings:    make(map[string]chan<- struct{}),
+		onPingReceived: cfg.onPingReceived,
+		onPongReceived: cfg.onPongReceived,
 	}
 
 	c.readMu = newMu(c)

conn_test.go

@@ -97,6 +97,60 @@ func TestConn(t *testing.T) {
 		assert.Contains(t, err, "failed to wait for pong")
 	})
 
+	t.Run("pingPongReceived1", func(t *testing.T) {
+		var pingReceived, pongReceived bool
+		tt, c1, c2 := newConnTest(t,
+			&websocket.DialOptions{
+				OnPingReceived: func(ctx context.Context, payload []byte) {
+					pingReceived = true
+				},
+			}, &websocket.AcceptOptions{
+				OnPongReceived: func(ctx context.Context, payload []byte) {
+					pongReceived = true
+				},
+			},
+		)
+
+		c1.CloseRead(tt.ctx)
+		c2.CloseRead(tt.ctx)
+
+		ctx, cancel := context.WithTimeout(tt.ctx, time.Millisecond*100)
+		defer cancel()
+
+		err := c1.Ping(ctx)
+		assert.Success(t, err)
+
+		assert.Equal(t, "ping received", true, pingReceived)
+		assert.Equal(t, "pong received", true, pongReceived)
+	})
+
+	t.Run("pingPongReceived2", func(t *testing.T) {
+		var pingReceived, pongReceived bool
+		tt, c1, c2 := newConnTest(t,
+			&websocket.DialOptions{
+				OnPongReceived: func(ctx context.Context, payload []byte) {
+					pongReceived = true
+				},
+			}, &websocket.AcceptOptions{
+				OnPingReceived: func(ctx context.Context, payload []byte) {
+					pingReceived = true
+				},
+			},
+		)
+
+		c1.CloseRead(tt.ctx)
+		c2.CloseRead(tt.ctx)
+
+		ctx, cancel := context.WithTimeout(tt.ctx, time.Millisecond*100)
+		defer cancel()
+
+		err := c2.Ping(ctx)
+		assert.Success(t, err)
+
+		assert.Equal(t, "ping received", true, pingReceived)
+		assert.Equal(t, "pong received", true, pongReceived)
+	})
+
 	t.Run("concurrentWrite", func(t *testing.T) {
 		tt, c1, c2 := newConnTest(t, nil, nil)
 

dial.go

@@ -48,6 +48,18 @@ type DialOptions struct {
 	// Defaults to 512 bytes for CompressionNoContextTakeover and 128 bytes
 	// for CompressionContextTakeover.
 	CompressionThreshold int
+
+	// OnPingReceived is an optional callback invoked synchronously when a ping frame is received.
+	//
+	// To avoid blocking, process the callback asynchronously using a goroutine.
+	// If you need to reuse the payload outside the callback, clone the byte slice.
+	// Any modifications to the payload within the callback will be sent in the subsequent pong frame.
+	OnPingReceived func(context.Context, []byte)
+
+	// OnPongReceived is an optional callback invoked synchronously when a pong frame is received.
+	//
+	// To avoid blocking, process the callback asynchronously using a goroutine.
+	OnPongReceived func(context.Context, []byte)
 }
 
 func (opts *DialOptions) cloneWithDefaults(ctx context.Context) (context.Context, context.CancelFunc, *DialOptions) {
@@ -163,6 +175,8 @@ func dial(ctx context.Context, urls string, opts *DialOptions, rand io.Reader) (
 		client:         true,
 		copts:          copts,
 		flateThreshold: opts.CompressionThreshold,
+		onPingReceived: opts.OnPingReceived,
+		onPongReceived: opts.OnPongReceived,
 		br:             getBufioReader(rwc),
 		bw:             getBufioWriter(rwc),
 	}), resp, nil

read.go

@@ -312,8 +312,14 @@ func (c *Conn) handleControl(ctx context.Context, h header) (err error) {
 
 	switch h.opcode {
 	case opPing:
+		if c.onPingReceived != nil {
+			c.onPingReceived(ctx, b)
+		}
 		return c.writeControl(ctx, opPong, b)
 	case opPong:
+		if c.onPongReceived != nil {
+			c.onPongReceived(ctx, b)
+		}
 		c.activePingsMu.Lock()
 		pong, ok := c.activePings[string(b)]
 		c.activePingsMu.Unlock()