hopp/metadapta.go

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()
	return encodeMessageA(this.underlying, trans, method, data)
}

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 encodeMessageA(writer io.Writer, trans int64, method uint16, data []byte) error {
	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 := writer.Write(buffer)
	return err
}

func partyFromTransID(id int64) Party {
	return id > 0
}