johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package channelconn

import (
	"io"
	"net"
	"os"
	"sync"
	"time"
)

const chunk = 16 << 10

type addr struct {
	network string
	address string
}

func (a addr) Network() string { return a.network }

func (a addr) String() string { return a.address }

type Conn struct {
	send func([]byte) error

	shut func() error

	in chan []byte

	done chan struct{}

	once sync.Once

	err error

	mu sync.Mutex

	buf []byte

	readDeadline time.Time

	writeDeadline time.Time
}

func New(send func([]byte) error, shut func() error) *Conn {
	if send == nil {
		send = func([]byte) error { return io.ErrClosedPipe }
	}
	if shut == nil {
		shut = func() error { return nil }
	}
	return &Conn{
		send: send,
		shut: shut,
		in:   make(chan []byte, 64),
		done: make(chan struct{}),
	}
}

func (c *Conn) Push(data []byte) bool {
	if len(data) == 0 {
		return true
	}
	select {
	case <-c.done:
		return false
	default:
	}
	copied := append([]byte(nil), data...)
	select {
	case c.in <- copied:
		return true
	case <-c.done:
		return false
	}
}

func (c *Conn) CloseWithError(err error) {
	if err == nil {
		err = io.EOF
	}
	c.once.Do(func() {
		c.mu.Lock()
		c.err = err
		c.mu.Unlock()
		close(c.done)
	})
}

func (c *Conn) Read(p []byte) (int, error) {
	c.mu.Lock()
	if len(c.buf) > 0 {
		n := copy(p, c.buf)
		c.buf = c.buf[n:]
		c.mu.Unlock()
		return n, nil
	}
	deadline := c.readDeadline
	c.mu.Unlock()
	var expire <-chan time.Time
	if !deadline.IsZero() {
		left := time.Until(deadline)
		if left <= 0 {
			return 0, os.ErrDeadlineExceeded
		}
		timer := time.NewTimer(left)
		defer timer.Stop()
		expire = timer.C
	}
	select {
	case data := <-c.in:
		n := copy(p, data)
		if n < len(data) {
			c.mu.Lock()
			c.buf = append(c.buf, data[n:]...)
			c.mu.Unlock()
		}
		return n, nil
	case <-c.done:
		select {
		case data := <-c.in:
			n := copy(p, data)
			if n < len(data) {
				c.mu.Lock()
				c.buf = append(c.buf, data[n:]...)
				c.mu.Unlock()
			}
			return n, nil
		default:
			return 0, c.closeError()
		}
	case <-expire:
		return 0, os.ErrDeadlineExceeded
	}
}

func (c *Conn) Write(p []byte) (int, error) {
	written := 0
	for written < len(p) {
		select {
		case <-c.done:
			return written, c.closeError()
		default:
		}
		if err := c.writeExpired(); err != nil {
			return written, err
		}
		n := min(len(p)-written, chunk)
		if err := c.send(p[written : written+n]); err != nil {
			return written, err
		}
		written += n
	}
	return written, nil
}

func (c *Conn) writeExpired() error {
	c.mu.Lock()
	deadline := c.writeDeadline
	c.mu.Unlock()
	if !deadline.IsZero() && !time.Now().Before(deadline) {
		return os.ErrDeadlineExceeded
	}
	return nil
}

func (c *Conn) Close() error {
	c.CloseWithError(io.EOF)
	return c.shut()
}

func (c *Conn) closeError() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.err == nil {
		return io.EOF
	}
	return c.err
}

func (c *Conn) LocalAddr() net.Addr { return addr{network: "webrtc", address: "local"} }

func (c *Conn) RemoteAddr() net.Addr { return addr{network: "webrtc", address: "remote"} }

func (c *Conn) SetDeadline(at time.Time) error {
	c.mu.Lock()
	c.readDeadline, c.writeDeadline = at, at
	c.mu.Unlock()
	return nil
}

func (c *Conn) SetReadDeadline(at time.Time) error {
	c.mu.Lock()
	c.readDeadline = at
	c.mu.Unlock()
	return nil
}

func (c *Conn) SetWriteDeadline(at time.Time) error {
	c.mu.Lock()
	c.writeDeadline = at
	c.mu.Unlock()
	return nil
}

var _ net.Conn = (*Conn)(nil)