conn_browser.go

// +build js,wasm

package websocket

import (
	"bytes"
	"errors"
	"fmt"
	"net"
	"strings"
	"sync"
	"syscall/js"
	"time"
)

const (
	webSocketStateConnecting = 0
	webSocketStateOpen       = 1
	webSocketStateClosing    = 2
	webSocketStateClosed     = 3
)

var errConnectionClosed = errors.New("connection is closed")

// Conn implements net.Conn interface for WebSockets in js/wasm.
type Conn struct {
	js.Value
	messageHandler  *js.Func
	closeHandler    *js.Func
	errorHandler    *js.Func
	mut             sync.Mutex
	currDataMut     sync.RWMutex
	currData        bytes.Buffer
	closeOnce       sync.Once
	closeSignalOnce sync.Once
	closeSignal     chan struct{}
	dataSignal      chan struct{}
	localAddr       net.Addr
	remoteAddr      net.Addr
	firstErr        error // only read this _after_ observing that closeSignal has been closed.
	mux             sync.RWMutex
}

// NewConn creates a Conn given a regular js/wasm WebSocket Conn.
func NewConn(raw js.Value) *Conn {
	conn := &Conn{
		Value:       raw,
		closeSignal: make(chan struct{}),
		dataSignal:  make(chan struct{}, 1),
		localAddr:   NewAddr("0.0.0.0:0"),
		remoteAddr:  getRemoteAddr(raw),
	}
	// Force the JavaScript WebSockets API to use the ArrayBuffer type for
	// incoming messages instead of the Blob type. This is better for us because
	// ArrayBuffer can be converted to []byte synchronously but Blob cannot.
	conn.Set("binaryType", "arraybuffer")
	conn.setUpHandlers()
	return conn
}

func (c *Conn) Read(b []byte) (int, error) {
	c.mux.RLock()
	defer c.mux.RUnlock()
	select {
	case <-c.closeSignal:
		c.readAfterErr(b)
	default:
	}

	for {
		c.currDataMut.RLock()
		n, _ := c.currData.Read(b)
		c.currDataMut.RUnlock()

		if n != 0 {
			// Data was ready. Return the number of bytes read.
			return n, nil
		}

		// There is no data ready to be read. Wait for more data or for the
		// connection to be closed.
		select {
		case <-c.dataSignal:
		case <-c.closeSignal:
			return c.readAfterErr(b)
		}
	}
}

// readAfterError reads from c.currData. If there is no more data left it
// returns c.firstErr if non-nil and otherwise returns io.EOF.
func (c *Conn) readAfterErr(b []byte) (int, error) {
	if c.firstErr != nil {
		return 0, c.firstErr
	}
	c.currDataMut.RLock()
	n, err := c.currData.Read(b)
	c.currDataMut.RUnlock()
	return n, err
}

// checkOpen returns an error if the connection is not open. Otherwise, it
// returns nil.
func (c *Conn) checkOpen() error {
	state := c.Get("readyState").Int()
	switch state {
	case webSocketStateClosed, webSocketStateClosing:
		return errConnectionClosed
	}
	return nil
}

func (c *Conn) Write(b []byte) (n int, err error) {
	c.mux.Lock()
	defer c.mux.Unlock()
	defer func() {
		if e := recover(); e != nil {
			err = recoveredValueToError(e)
		}
	}()
	if err := c.checkOpen(); err != nil {
		return 0, err
	}
	uint8Array := js.Global().Get("Uint8Array").New(len(b))
	if js.CopyBytesToJS(uint8Array, b) != len(b) {
		panic("expected to copy all bytes")
	}
	c.Call("send", uint8Array.Get("buffer"))
	return len(b), nil
}

// Close closes the connection. Only the first call to Close will receive the
// close error, subsequent and concurrent calls will return nil.
// This method is thread-safe.
func (c *Conn) Close() error {
	c.mux.Lock()
	defer c.mux.Unlock()
	c.closeOnce.Do(func() {
		c.Call("close")
		c.signalClose(nil)
		c.releaseHandlers()
	})
	return nil
}

func (c *Conn) signalClose(err error) {
	c.closeSignalOnce.Do(func() {
		c.firstErr = err
		close(c.closeSignal)
	})
}

func (c *Conn) releaseHandlers() {
	c.mut.Lock()
	defer c.mut.Unlock()
	if c.messageHandler != nil {
		c.Call("removeEventListener", "message", *c.messageHandler)
		c.messageHandler.Release()
		c.messageHandler = nil
	}
	if c.closeHandler != nil {
		c.Call("removeEventListener", "close", *c.closeHandler)
		c.closeHandler.Release()
		c.closeHandler = nil
	}
	if c.errorHandler != nil {
		c.Call("removeEventListener", "error", *c.errorHandler)
		c.errorHandler.Release()
		c.errorHandler = nil
	}
}

func (c *Conn) LocalAddr() net.Addr {
	c.mux.Rlock()
	defer c.mux.RUnlock()
	return c.localAddr
}

func getRemoteAddr(val js.Value) net.Addr {
	rawURL := val.Get("url").String()
	withoutPrefix := strings.TrimPrefix(rawURL, "ws://")
	withoutSuffix := strings.TrimSuffix(withoutPrefix, "/")
	return NewAddr(withoutSuffix)
}

func (c *Conn) RemoteAddr() net.Addr {
	return c.remoteAddr
}

// TODO: Return os.ErrNoDeadline. For now we return nil because multiplexers
// don't handle the error correctly.
func (c *Conn) SetDeadline(t time.Time) error {
	c.mux.Rlock()
	defer c.mux.RUnlock()
	return nil
}

func (c *Conn) SetReadDeadline(t time.Time) error {
	c.mux.Rlock()
	defer c.mux.RUnlock()
	return nil
}

func (c *Conn) SetWriteDeadline(t time.Time) error {
	c.mux.Rlock()
	defer c.mux.RUnlock()
	return nil
}

func (c *Conn) setUpHandlers() {
	c.mut.Lock()
	defer c.mut.Unlock()
	if c.messageHandler != nil {
		// Message handlers already created. Nothing to do.
		return
	}
	messageHandler := js.FuncOf(func(this js.Value, args []js.Value) interface{} {
		arrayBuffer := args[0].Get("data")
		data := arrayBufferToBytes(arrayBuffer)
		c.currDataMut.Lock()
		if _, err := c.currData.Write(data); err != nil {
			c.currDataMut.Unlock()
			return err
		}
		c.currDataMut.Unlock()

		// Non-blocking signal
		select {
		case c.dataSignal <- struct{}{}:
		default:
		}

		return nil
	})
	c.messageHandler = &messageHandler
	c.Call("addEventListener", "message", messageHandler)

	closeHandler := js.FuncOf(func(this js.Value, args []js.Value) interface{} {
		go func() {
			c.signalClose(errorEventToError(args[0]))
			c.releaseHandlers()
		}()
		return nil
	})
	c.closeHandler = &closeHandler
	c.Call("addEventListener", "close", closeHandler)

	errorHandler := js.FuncOf(func(this js.Value, args []js.Value) interface{} {
		// Unfortunately, the "error" event doesn't appear to give us any useful
		// information. All we can do is close the connection.
		c.Close()
		return nil
	})
	c.errorHandler = &errorHandler
	c.Call("addEventListener", "error", errorHandler)
}

func (c *Conn) waitForOpen() error {
	openSignal := make(chan struct{})
	handler := js.FuncOf(func(this js.Value, args []js.Value) interface{} {
		close(openSignal)
		return nil
	})
	defer c.Call("removeEventListener", "open", handler)
	defer handler.Release()
	c.Call("addEventListener", "open", handler)
	select {
	case <-openSignal:
		return nil
	case <-c.closeSignal:
		// c.closeSignal means there was an error when trying to open the
		// connection.
		return c.firstErr
	}
}

// arrayBufferToBytes converts a JavaScript ArrayBuffer to a slice of bytes.
func arrayBufferToBytes(buffer js.Value) []byte {
	view := js.Global().Get("Uint8Array").New(buffer)
	dataLen := view.Length()
	data := make([]byte, dataLen)
	if js.CopyBytesToGo(data, view) != dataLen {
		panic("expected to copy all bytes")
	}
	return data
}

func errorEventToError(val js.Value) error {
	var typ string
	if gotType := val.Get("type"); gotType != js.Undefined() {
		typ = gotType.String()
	} else {
		typ = val.Type().String()
	}
	var reason string
	if gotReason := val.Get("reason"); gotReason != js.Undefined() && gotReason.String() != "" {
		reason = gotReason.String()
	} else {
		code := val.Get("code")
		if code != js.Undefined() {
			switch code := code.Int(); code {
			case 1006:
				reason = "code 1006: connection unexpectedly closed"
			default:
				reason = fmt.Sprintf("unexpected code: %d", code)
			}
		}
	}
	return fmt.Errorf("JavaScript error: (%s) %s", typ, reason)
}

func recoveredValueToError(e interface{}) error {
	switch e := e.(type) {
	case error:
		return e
	default:
		return fmt.Errorf("recovered from unexpected panic: %T %s", e, e)
	}
}