//go:build !js

package websocket

import (
	
	
	
	
	
	
	
	
	
)

// 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
)

// 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.
//
// This applies to context expirations as well unfortunately.
// See https://github.com/nhooyr/websocket/issues/242#issuecomment-633182220
type Conn struct {
	noCopy noCopy

	subprotocol    string
	rwc            io.ReadWriteCloser
	client         bool
	copts          *compressionOptions
	flateThreshold int
	br             *bufio.Reader
	bw             *bufio.Writer

	readTimeoutStop  atomic.Pointer[func() bool]
	writeTimeoutStop atomic.Pointer[func() bool]

	// Read state.
	readMu         *mu
	readHeaderBuf  [8]byte
	readControlBuf [maxControlPayload]byte
	msgReader      *msgReader

	// Write state.
	msgWriter      *msgWriter
	writeFrameMu   *mu
	writeBuf       []byte
	writeHeaderBuf [8]byte
	writeHeader    header

	// Close handshake state.
	closeStateMu     sync.RWMutex
	closeReceivedErr error
	closeSentErr     error

	// CloseRead state.
	closeReadMu   sync.Mutex
	closeReadCtx  context.Context
	closeReadDone chan struct{}

	closing atomic.Bool
	closeMu sync.Mutex // Protects following.
	closed  chan struct{}

	pingCounter    atomic.Int64
	activePingsMu  sync.Mutex
	activePings    map[string]chan<- struct{}
	onPingReceived func(context.Context, []byte) bool
	onPongReceived func(context.Context, []byte)
}

type connConfig struct {
	subprotocol    string
	rwc            io.ReadWriteCloser
	client         bool
	copts          *compressionOptions
	flateThreshold int
	onPingReceived func(context.Context, []byte) bool
	onPongReceived func(context.Context, []byte)

	br *bufio.Reader
	bw *bufio.Writer
}

func ( connConfig) *Conn {
	 := &Conn{
		subprotocol:    .subprotocol,
		rwc:            .rwc,
		client:         .client,
		copts:          .copts,
		flateThreshold: .flateThreshold,

		br: .br,
		bw: .bw,

		closed:         make(chan struct{}),
		activePings:    make(map[string]chan<- struct{}),
		onPingReceived: .onPingReceived,
		onPongReceived: .onPongReceived,
	}

	.readMu = newMu()
	.writeFrameMu = newMu()

	.msgReader = newMsgReader()

	.msgWriter = newMsgWriter()
	if .client {
		.writeBuf = extractBufioWriterBuf(.bw, .rwc)
	}

	if .flate() && .flateThreshold == 0 {
		.flateThreshold = 128
		if !.msgWriter.flateContextTakeover() {
			.flateThreshold = 512
		}
	}

	runtime.SetFinalizer(, func( *Conn) {
		.close()
	})

	return 
}

// Subprotocol returns the negotiated subprotocol.
// An empty string means the default protocol.
func ( *Conn) () string {
	return .subprotocol
}

func ( *Conn) () error {
	.closeMu.Lock()
	defer .closeMu.Unlock()

	if .isClosed() {
		return net.ErrClosed
	}
	runtime.SetFinalizer(, nil)
	close(.closed)

	// 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.
	 := .rwc.Close()
	// With the close of rwc, these become safe to close.
	.msgWriter.close()
	.msgReader.close()
	return 
}

func ( *Conn) ( context.Context) {
	 := context.AfterFunc(, func() {
		.clearWriteTimeout()
		.close()
	})
	swapTimeoutStop(&.writeTimeoutStop, &)
}

func ( *Conn) () {
	swapTimeoutStop(&.writeTimeoutStop, nil)
}

func ( *Conn) ( context.Context) {
	 := context.AfterFunc(, func() {
		.clearReadTimeout()
		.close()
	})
	swapTimeoutStop(&.readTimeoutStop, &)
}

func ( *Conn) () {
	swapTimeoutStop(&.readTimeoutStop, nil)
}

func ( *atomic.Pointer[func() bool],  *func() bool) {
	 := .Swap()
	if  != nil {
		(*)()
	}
}

func ( *Conn) () bool {
	return .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 ( *Conn) ( context.Context) error {
	 := .pingCounter.Add(1)

	 := .ping(, strconv.FormatInt(, 10))
	if  != nil {
		return fmt.Errorf("failed to ping: %w", )
	}
	return nil
}

func ( *Conn) ( context.Context,  string) error {
	 := make(chan struct{}, 1)

	.activePingsMu.Lock()
	.activePings[] = 
	.activePingsMu.Unlock()

	defer func() {
		.activePingsMu.Lock()
		delete(.activePings, )
		.activePingsMu.Unlock()
	}()

	 := .writeControl(, opPing, []byte())
	if  != nil {
		return 
	}

	select {
	case <-.closed:
		return net.ErrClosed
	case <-.Done():
		return fmt.Errorf("failed to wait for pong: %w", .Err())
	case <-:
		return nil
	}
}

type mu struct {
	c  *Conn
	ch chan struct{}
}

func ( *Conn) *mu {
	return &mu{
		c:  ,
		ch: make(chan struct{}, 1),
	}
}

func ( *mu) () {
	.ch <- struct{}{}
}

func ( *mu) () bool {
	select {
	case .ch <- struct{}{}:
		return true
	default:
		return false
	}
}

func ( *mu) ( context.Context) error {
	select {
	case <-.c.closed:
		return net.ErrClosed
	case <-.Done():
		return fmt.Errorf("failed to acquire lock: %w", .Err())
	case .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 <-.c.closed:
			// Make sure to release.
			.unlock()
			return net.ErrClosed
		default:
		}
		return nil
	}
}

func ( *mu) () {
	select {
	case <-.ch:
	default:
	}
}

type noCopy struct{}

func (*noCopy) () {}