<commit_before>// +build js,wasm

package websocket

import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"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
	closeSignal    chan struct{}
	dataSignal     chan struct{}
	localAddr      net.Addr
	remoteAddr     net.Addr
	firstErr       error
}

// 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) {
	if err := c.checkOpen(); err != nil {
		return c.readAfterErr(b)
	}

	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:
			continue
		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) {
	c.currDataMut.RLock()
	n, err := c.currData.Read(b)
	c.currDataMut.RUnlock()
	if n == 0 {
		if c.firstErr != nil {
			return 0, c.firstErr
		} else {
			return 0, io.EOF
		}
	}
	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) {
	if err := c.checkOpen(); err != nil {
		return 0, err
	}
	uint8Array := js.Global().Get("Uint8Array").New(len(b))
	for i := 0; i < len(b); i++ {
		uint8Array.SetIndex(i, b[i])
	}
	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.signalClose()
	c.Call("close")
	c.releaseHandlers()
	return nil
}

func (c *Conn) signalClose() {
	c.closeOnce.Do(func() {
		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()<commit_msg>Simplify iteration over Uint8Array<commit_after>// +build js,wasm

package websocket

import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"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
	closeSignal    chan struct{}
	dataSignal     chan struct{}
	localAddr      net.Addr
	remoteAddr     net.Addr
	firstErr       error
}

// 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) {
	if err := c.checkOpen(); err != nil {
		return c.readAfterErr(b)
	}

	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:
			continue
		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) {
	c.currDataMut.RLock()
	n, err := c.currData.Read(b)
	c.currDataMut.RUnlock()
	if n == 0 {
		if c.firstErr != nil {
			return 0, c.firstErr
		} else {
			return 0, io.EOF
		}
	}
	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) {
	if err := c.checkOpen(); err != nil {
		return 0, err
	}
	uint8Array := js.Global().Get("Uint8Array").New(len(b))
	for i, bi := range b {
		uint8Array.SetIndex(i, bi)
	}
	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.signalClose()
	c.Call("close")
	c.releaseHandlers()
	return nil
}

func (c *Conn) signalClose() {
	c.closeOnce.Do(func() {
		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()
{'fraction_non_alphanumeric': 0.1260753485612578, 'fraction_numerical': 0.0059329575793533075, 'mean_word_length': 3.283989834815756, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4905868', 'n_tokens_mistral': 2490, 'n_tokens_neox': 2358, 'n_words': 727}