hopp/metadapta.go

package hopp

import "io"
import "fmt"
import "net"
import "sync"
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
	done         chan struct { }
	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),
		done:       make(chan struct { }),
	}
	if party == ClientSide {
		conn.transID =  1
	} else {
		conn.transID = -1
	}
	go conn.receive()
	return conn
}

func (this *a) Close() error {
	close(this.done)
	return this.underlying.Close()
}

func (this *a) LocalAddr() net.Addr {
	return this.underlying.LocalAddr()
}

func (this *a) RemoteAddr() net.Addr {
	return this.underlying.RemoteAddr()
}

func (this *a) OpenTrans() (Trans, error) {
	this.transLock.Lock()
	defer this.transLock.Unlock()
	id := this.transID
	this.transID ++
	trans := &transA {
		parent:   this,
		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() (Trans, error) {
	select {
	case trans := <- this.transChan:
		return trans, nil
	case <- this.done:
		return nil, fmt.Errorf("could not accept transaction: %w", io.EOF)
	}
}

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

func (this *a) receive() {
	defer func() {
		this.underlying.Close()
		this.transLock.Lock()
		defer this.transLock.Unlock()
		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 { return ErrMessageMalformed }

	trans, err := func() (*transA, error) {
		this.transLock.Lock()
		defer this.transLock.Unlock()

		trans, ok := this.transMap[transID]
		if !ok {
			// it is forbidden for the other party to initiate a transaction
			// with an ID from this party
			if this.party == partyFromTransID(transID) {
				return nil, ErrMessageMalformed
			}
			trans = &transA {
				parent:   this,
				id:       transID,
				incoming: usync.NewGate[incomingMessage](),
			}
			this.transMap[transID] = trans
			this.transChan <- trans
		}
		return trans, nil
	}()
	if err != nil { return err }

	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 encodeMessageA(writer io.Writer, trans int64, method uint16, data []byte) error {
	buffer := make([]byte, 12 + len(data))
	tape.EncodeI64(buffer[:8], trans)
	tape.EncodeI16(buffer[8:10], method)
	length, ok := tape.U16CastSafe(len(data))
	if !ok { return ErrPayloadTooLarge }
	tape.EncodeI16(buffer[10:12], length)
	copy(buffer[12:], data)
	_, err := writer.Write(buffer)
	return err
}

func decodeMessageA(reader io.Reader) (int64, uint16, []byte, error) {
	headerBuffer := [12]byte { }
	_, err := io.ReadFull(reader, headerBuffer[:])
	if err != nil { return 0, 0, nil, err }
	transID, err := tape.DecodeI64[int64](headerBuffer[:8])
	if err != nil { return 0, 0, nil, err }
	method, err := tape.DecodeI16[uint16](headerBuffer[8:10])
	if err != nil { return 0, 0, nil, err }
	length, err := tape.DecodeI16[uint16](headerBuffer[10:12])
	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
}