After the week I've had, I deserve to make a commit like this lmao

metadapta.go

@@ -0,0 +1,203 @@
+package hopp
+
+import "io"
+import "fmt"
+import "net"
+import "sync"
+import "context"
+import "git.tebibyte.media/sashakoshka/hopp/tape"
+import "git.tebibyte.media/sashakoshka/go-util/sync"
+
+const int64Max = int64((^uint64(0)) >> 1)
+
+// Party represents a side of a connection.
+type Party bool; const (
+	ServerSide Party = false
+	ClientSide Party = true
+)
+
+type a struct {
+	underlying   net.Conn
+	party        Party
+	transID      int64
+	transLock    sync.RWMutex
+	sendLock     sync.Mutex
+	transMap     map[int64] *transA
+	transChan    chan *transA
+	err          error
+}
+
+// AdaptA returns a connection implementing METADAPT-A over a singular stream-
+// oriented transport such as TCP or UNIX domain stream sockets.
+func AdaptA(underlying net.Conn, party Party) Conn {
+	conn := &a {
+		underlying: underlying,
+		party:      party,
+		transMap:   make(map[int64] *transA),
+		transChan:  make(chan *transA),
+	}
+	go conn.receive()
+	return conn
+}
+
+func (this *a) Close() error {
+	return this.underlying.Close()
+}
+
+func (this *a) OpenTrans() (Trans, error) {
+	this.transLock.Lock()
+	defer this.transLock.Unlock()
+	id := this.transID
+	this.transID ++
+	trans := &transA {
+		id:       id,
+		incoming: usync.NewGate[incomingMessage](),
+	}
+	this.transMap[id] = trans
+	if this.transID == int64Max {
+		return nil, fmt.Errorf("could not open transaction: %w", ErrIntegerOverflow)
+	}
+	this.transID ++
+	return trans, nil
+}
+
+func (this *a) AcceptTrans(ctx context.Context) (Trans, error) {
+	select {
+	case trans := <- this.transChan:
+		return trans, nil
+	case <- ctx.Done():
+		return nil, ctx.Err()	
+	}
+}
+
+func (this *a) unlistTransactionSafe(id int64) {
+	this.transLock.Lock()
+	defer this.transLock.Unlock()
+	delete(this.transMap, id)
+}
+
+func (this *a) sendMessageSafe(trans int64, method uint16, data []byte) error {
+	this.sendLock.Lock()
+	defer this.sendLock.Lock()
+	
+	buffer := make([]byte, 8 + len(data))
+	tape.EncodeI64(buffer[:4], trans)
+	tape.EncodeI16(buffer[4:6], method)
+	length, ok := tape.U16CastSafe(len(data))
+	if !ok { return ErrPayloadTooLarge }
+	tape.EncodeI16(data[6:8], length)
+	copy(buffer[8:], data)
+	_, err := this.underlying.Write(buffer)
+	return err
+}
+
+func (this *a) receive() {
+	// TODO: multiplex receiving
+	// if a received transaction has a malformed ID, reject it here and
+	// cause the connection to fail
+	// at the end of this function, close all incoming channels
+	defer func() {
+		this.underlying.Close()
+		this.transLock.Lock()
+		defer this.transLock.Lock()
+		for _, trans := range this.transMap {
+			trans.Close()
+		}
+		clear(this.transMap)
+	}()
+	for {
+		transID, method, payload, err := decodeMessageA(this.underlying)
+		if err != nil {
+			this.err = fmt.Errorf("could not receive message: %w", err)
+			return
+		}
+
+		err = this.receiveMultiplex(transID, method, payload)
+		if err != nil {
+			this.err = fmt.Errorf("could not receive message: %w", err)
+			return
+		}
+	}
+}
+
+func (this *a) receiveMultiplex(transID int64, method uint16, payload []byte) error {
+	if transID == 0 || this.party == partyFromTransID(transID) {
+		return ErrMessageMalformed
+	}
+	
+	this.transLock.Lock()
+	defer this.transLock.Unlock()
+
+	trans, ok := this.transMap[transID]
+	if !ok {
+		trans = &transA {
+			parent:   this,
+			id:       transID,
+			incoming: usync.NewGate[incomingMessage](),
+		}
+		this.transChan <- trans
+	}
+
+	trans.incoming.Send(incomingMessage {
+		method:  method,
+		payload: payload,
+	})
+	return nil
+}
+
+type transA struct {
+	parent   *a
+	id       int64
+	incoming usync.Gate[incomingMessage]
+}
+
+func (this *transA) Close() error {
+	this.incoming.Close()
+	this.parent.unlistTransactionSafe(this.ID())
+	return nil
+}
+
+func (this *transA) ID() int64 {
+	return this.id
+}
+
+func (this *transA) Send(method uint16, data []byte) error {
+	return this.parent.sendMessageSafe(this.id, method, data)
+}
+
+func (this *transA) Receive() (method uint16, data []byte, err error) {
+	message, ok := <- this.incoming.Receive()
+	if !ok {
+		if this.parent.err == nil {
+			return 0, nil, fmt.Errorf("could not receive message: %w", io.EOF)
+		} else {
+			return 0, nil, this.parent.err
+		}
+	}
+	return message.method, message.payload, nil
+}
+
+type incomingMessage struct {
+	method  uint16
+	payload []byte
+}
+
+func decodeMessageA(reader io.Reader) (int64, uint16, []byte, error) {
+	headerBuffer := [8]byte { }
+	_, err := io.ReadFull(reader, headerBuffer[:])
+	if err != nil { return 0, 0, nil, err }
+	transID, err := tape.DecodeI64[int64](headerBuffer[:4])
+	if err != nil { return 0, 0, nil, err }
+	method, err := tape.DecodeI16[uint16](headerBuffer[4:6])
+	if err != nil { return 0, 0, nil, err }
+	length, err := tape.DecodeI16[uint16](headerBuffer[6:8])
+	if err != nil { return 0, 0, nil, err }
+	payloadBuffer := make([]byte, int(length))
+	_, err = io.ReadFull(reader, payloadBuffer)
+	if err != nil { return 0, 0, nil, err }
+	return transID, method, payloadBuffer, nil
+}
+
+func partyFromTransID(id int64) Party {
+	return id > 0
+}