johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

//go:build js

package wsconn

import (
	"context"
	"errors"
	"fmt"
	"net"
	"os"
	"sync"
	"syscall/js"
	"time"
)

const (
	socketConnecting = 0
	socketOpen       = 1
	closeNormal      = 1000
	frameOverhead    = 64
)

var socketHandlers = []string{"onopen", "onmessage", "onerror", "onclose"}

type Conn struct {
	ws     js.Value
	ready  chan struct{}
	opened chan struct{}
	done   chan struct{}

	mu       sync.Mutex
	err      error
	isOpen   bool
	queue    [][]byte
	buffered int
	funcs    []js.Func
	released bool
}

func DialConfig(ctx context.Context, target string, cfg Config) (*Conn, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	ctor := js.Global().Get("WebSocket")
	if ctor.Type() != js.TypeFunction {
		return nil, errors.New("wsconn: websocket dial: this js runtime has no WebSocket")
	}
	ws, err := newSocket(ctor, target)
	if err != nil {
		return nil, fmt.Errorf("wsconn: websocket dial: %w", err)
	}
	c := &Conn{
		ws:     ws,
		ready:  make(chan struct{}, 1),
		opened: make(chan struct{}),
		done:   make(chan struct{}),
	}
	ws.Set("binaryType", "arraybuffer")
	c.handle("onopen", c.onOpen)
	c.handle("onmessage", c.onMessage)
	c.handle("onerror", c.onError)
	c.handle("onclose", c.onClose)
	timeout := cfg.handshakeTimeout()
	timer := time.NewTimer(timeout)
	defer timer.Stop()
	select {
	case <-c.opened:
		return c, nil
	case <-c.done:
		failure := c.failure()
		_ = c.Close()
		return nil, fmt.Errorf("wsconn: websocket dial: %w", failure)
	case <-ctx.Done():
		_ = c.Close()
		return nil, ctx.Err()
	case <-timer.C:
		_ = c.Close()
		return nil, fmt.Errorf("wsconn: websocket dial: handshake timeout after %s: %w", timeout, os.ErrDeadlineExceeded)
	}
}

func newSocket(ctor js.Value, target string) (ws js.Value, err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("%v", r)
		}
	}()
	return ctor.New(target), nil
}

func (c *Conn) handle(name string, fn func(js.Value)) {
	f := js.FuncOf(func(_ js.Value, args []js.Value) any {
		event := js.Undefined()
		if len(args) > 0 {
			event = args[0]
		}
		fn(event)
		return nil
	})
	c.funcs = append(c.funcs, f)
	c.ws.Set(name, f)
}

func (c *Conn) onOpen(js.Value) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.isOpen {
		return
	}
	c.isOpen = true
	close(c.opened)
}

func (c *Conn) onError(js.Value) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.endLocked(&CloseError{Code: CloseAbnormal, Reason: "error event"})
}

func (c *Conn) onClose(event js.Value) {
	closeErr := &CloseError{Code: CloseAbnormal}
	if event.Type() == js.TypeObject {
		if code := event.Get("code"); code.Type() == js.TypeNumber {
			closeErr.Code = code.Int()
		}
		if reason := event.Get("reason"); reason.Type() == js.TypeString {
			closeErr.Reason = reason.String()
		}
		closeErr.Clean = event.Get("wasClean").Truthy()
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	c.endLocked(closeErr)
}

func (c *Conn) onMessage(event js.Value) {
	frame, err := frameOf(event)
	if err == nil {
		err = c.buffer(frame)
	}
	if err != nil {
		c.fail(err)
	}
}

func (c *Conn) buffer(frame []byte) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.err != nil {
		return nil
	}
	size := len(frame) + frameOverhead
	if c.buffered+size > MaxBufferedBytes {
		return fmt.Errorf("wsconn: websocket read buffer is over %d bytes in %d frames", MaxBufferedBytes, len(c.queue))
	}
	c.queue = append(c.queue, frame)
	c.buffered += size
	select {
	case c.ready <- struct{}{}:
	default:
	}
	return nil
}

func frameOf(event js.Value) ([]byte, error) {
	if event.Type() != js.TypeObject {
		return nil, errors.New("wsconn: websocket message without an event")
	}
	data := event.Get("data")
	switch data.Type() {
	case js.TypeString:
		frame := []byte(data.String())
		if len(frame) > MaxFrameBytes {
			return nil, frameTooBig(len(frame))
		}
		return frame, nil
	case js.TypeObject:
		size := data.Get("byteLength")
		if size.Type() != js.TypeNumber {
			return nil, errors.New("wsconn: websocket message is neither text nor an ArrayBuffer")
		}
		n := size.Int()
		if n > MaxFrameBytes {
			return nil, frameTooBig(n)
		}
		frame := make([]byte, n)
		if n > 0 {
			js.CopyBytesToGo(frame, js.Global().Get("Uint8Array").New(data))
		}
		return frame, nil
	default:
		return nil, errors.New("wsconn: websocket message is neither text nor an ArrayBuffer")
	}
}

func frameTooBig(n int) error {
	return fmt.Errorf("wsconn: websocket frame of %d bytes is over the %d byte limit", n, MaxFrameBytes)
}

func (c *Conn) endLocked(err error) bool {
	if c.err != nil {
		return false
	}
	c.err = err
	close(c.done)
	return true
}

func (c *Conn) fail(err error) {
	c.mu.Lock()
	ended := c.endLocked(err)
	c.mu.Unlock()
	if ended {
		c.closeSocket()
	}
}

func (c *Conn) closeSocket() {
	defer func() { _ = recover() }()
	switch c.ws.Get("readyState").Int() {
	case socketConnecting, socketOpen:
		c.ws.Call("close", closeNormal)
	}
}

func (c *Conn) failure() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.err
}

func (c *Conn) next() ([]byte, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.queue) == 0 {
		return nil, false
	}
	frame := c.queue[0]
	c.queue[0] = nil
	c.queue = c.queue[1:]
	if len(c.queue) == 0 {
		c.queue = nil
	}
	c.buffered -= len(frame) + frameOverhead
	return frame, true
}

func (c *Conn) Read(ctx context.Context) ([]byte, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if frame, ok := c.next(); ok {
		return frame, nil
	}
	timer := time.NewTimer(ReadTimeout)
	defer timer.Stop()
	for {
		select {
		case <-c.ready:
			if frame, ok := c.next(); ok {
				return frame, nil
			}
		case <-c.done:
			if frame, ok := c.next(); ok {
				return frame, nil
			}
			return nil, c.failure()
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-timer.C:
			c.fail(fmt.Errorf("wsconn: websocket read: no frame in %s: %w", ReadTimeout, os.ErrDeadlineExceeded))
			return nil, c.failure()
		}
	}
}

func (c *Conn) Close() error {
	c.mu.Lock()
	c.endLocked(fmt.Errorf("wsconn: %w", net.ErrClosed))
	if c.released {
		c.mu.Unlock()
		return nil
	}
	c.released = true
	funcs := c.funcs
	c.funcs = nil
	c.mu.Unlock()
	for _, name := range socketHandlers {
		c.ws.Set(name, js.Null())
	}
	c.closeSocket()
	for _, f := range funcs {
		f.Release()
	}
	return nil
}