WIP METADAPT-A changes

connection.go

@@ -4,7 +4,7 @@ import "io"
 import "net"
 // import "time"
 
-const defaultSizeLimit = 1024 * 1024 // 1 megabyte
+const defaultSizeLimit int64 = 1024 * 1024 // 1 megabyte
 
 // Conn is a HOPP connection.
 type Conn interface {
@@ -25,7 +25,7 @@ type Conn interface {
 
 	// SetSizeLimit sets a limit (in bytes) for how large messages can be.
 	// By default, this limit is 1 megabyte.
-	SetSizeLimit(limit int)
+	SetSizeLimit(limit int64)
 }
 
 // Trans is a HOPP transaction.

metadapta.go

@@ -17,6 +17,7 @@ type Party bool; const (
 )
 
 type a struct {
+	sizeLimit    int
 	underlying   net.Conn
 	party        Party
 	transID      int64
@@ -87,6 +88,10 @@ func (this *a) AcceptTrans() (Trans, error) {
 	}
 }
 
+func (this *a) SetSizeLimit(limit int) {
+	this.sizeLimit = limit
+}
+
 func (this *a) unlistTransactionSafe(id int64) {
 	this.transLock.Lock()
 	defer this.transLock.Unlock()
@@ -110,13 +115,13 @@ func (this *a) receive() {
 		clear(this.transMap)
 	}()
 	for {
-		transID, method, payload, err := decodeMessageA(this.underlying)
+		transID, method, chunked, 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)
+		err = this.receiveMultiplex(transID, method, chunked, payload)
 		if err != nil {
 			this.err = fmt.Errorf("could not receive message: %w", err)
 			return
@@ -124,7 +129,7 @@ func (this *a) receive() {
 	}
 }
 
-func (this *a) receiveMultiplex(transID int64, method uint16, payload []byte) error {
+func (this *a) receiveMultiplex(transID int64, method uint16, chunked bool, payload []byte) error {
 	if transID == 0 { return ErrMessageMalformed }
 
 	trans, err := func() (*transA, error) {
@@ -152,15 +157,17 @@ func (this *a) receiveMultiplex(transID int64, method uint16, payload []byte) er
 
 	trans.incoming.Send(incomingMessage {
 		method:  method,
+		chunked: chunked,
 		payload: payload,
 	})
 	return nil
 }
 
 type transA struct {
-	parent   *a
+	parent        *a
-	id       int64
+	id            int64
-	incoming usync.Gate[incomingMessage]
+	incoming      usync.Gate[incomingMessage]
+	currentReader io.Reader
 }
 
 func (this *transA) Close() error {
@@ -183,26 +190,78 @@ func (this *transA) Send(method uint16, data []byte) error {
 }
 
 func (this *transA) Receive() (method uint16, data []byte, err error) {
-	receive := this.incoming.Receive()
+	method, reader, err := this.ReceiveReader()
+	if err != nil { return 0, nil, err }
+	data, err = io.ReadAll(reader)
+	if err != nil { return 0, nil, err }
+	return method, data, nil
+}
+
+func (this *transA) ReceiveReader() (uint16, io.Reader, error) {
+	// drain previous reader if necessary
+	if this.currentReader != nil {
+		io.Copy(io.Discard, this.currentReader)
+	}
+
+	// create new reader
+	reader := &readerA {
+		parent: this,
+	}
+	method, err := reader.pull()
+	if err != nil { return 0, nil, err}
+	this.currentReader = reader
+	return method, reader, nil
+}
+
+type readerA struct {
+	parent   *transA
+	leftover []byte
+	eof      bool
+}
+
+func (this *readerA) pull() (uint16, error) {
+	// if the previous message ended the chain, return an io.EOF
+	if this.eof {
+		return 0, io.EOF
+	}
+
+	// get a message from the transaction we are a part of
+	receive := this.parent.incoming.Receive()
 	if receive != nil {
 		if message, ok := <- receive; ok {
 			if message.method != closeMethod {
-				return message.method, message.payload, nil
+				this.leftover = append(this.leftover, message.payload...)
+				if !message.chunked {
+					this.eof = true
+				}
 			}
 		}
 	}
 
 	// close and return error on failure
-	this.Close()
+	this.eof = true
-	if this.parent.err == nil {
+	this.parent.Close()
-		return 0, nil, fmt.Errorf("could not receive message: %w", io.EOF)
+	if this.parent.parent.err == nil {
+		return 0, fmt.Errorf("could not receive message: %w", io.EOF)
 	} else {
-		return 0, nil, this.parent.err
+		return 0, this.parent.parent.err
 	}
 }
 
+func (this *readerA) Read(buffer []byte) (int, error) {
+	if len(this.leftover) == 0 {
+		if this.eof { return 0, io.EOF }
+		this.pull()
+	}
+
+	copied := copy(buffer, this.leftover)
+	this.leftover = this.leftover[copied:]
+	return copied, nil
+}
+
 type incomingMessage struct {
 	method  uint16
+	chunked bool
 	payload []byte
 }
 
@@ -218,22 +277,27 @@ func encodeMessageA(writer io.Writer, trans int64, method uint16, data []byte) e
 	return err
 }
 
-func decodeMessageA(reader io.Reader) (int64, uint16, []byte, error) {
+func decodeMessageA(reader io.Reader) (int64, uint16, bool, []byte, error) {
-	headerBuffer := [12]byte { }
+	headerBuffer := [18]byte { }
 	_, err := io.ReadFull(reader, headerBuffer[:])
-	if err != nil { return 0, 0, nil, err }
+	if err != nil { return 0, 0, false, nil, err }
 	transID, err := tape.DecodeI64[int64](headerBuffer[:8])
-	if err != nil { return 0, 0, nil, err }
+	if err != nil { return 0, 0, false, nil, err }
 	method, err := tape.DecodeI16[uint16](headerBuffer[8:10])
-	if err != nil { return 0, 0, nil, err }
+	if err != nil { return 0, 0, false, nil, err }
-	length, err := tape.DecodeI16[uint16](headerBuffer[10:12])
+	size, err := tape.DecodeI64[uint64](headerBuffer[10:18])
-	if err != nil { return 0, 0, nil, err }
+	if err != nil { return 0, 0, false, nil, err }
-	payloadBuffer := make([]byte, int(length))
+	chunked, size := splitCCBSize(size)
+	payloadBuffer := make([]byte, int(size))
 	_, err = io.ReadFull(reader, payloadBuffer)
-	if err != nil { return 0, 0, nil, err }
+	if err != nil { return 0, 0, false, nil, err }
-	return transID, method, payloadBuffer, nil
+	return transID, method, chunked, payloadBuffer, nil
 }
 
 func partyFromTransID(id int64) Party {
 	return id > 0
 }
+
+func splitCCBSize(size uint64) (bool, uint64) {
+	return size >> 63 > 1, size & 0x7FFFFFFFFFFFFFFF
+}

metadaptb.go

@@ -8,12 +8,12 @@ import "git.tebibyte.media/sashakoshka/hopp/tape"
 // B implements METADAPT-B over a multiplexed stream-oriented transport such as
 // QUIC.
 type b struct {
-	sizeLimit  int
+	sizeLimit  int64
 	underlying MultiConn
 }
 
-// AdaptB returns a connection implementing METADAPT-B over a singular stream-
+// AdaptB returns a connection implementing METADAPT-B over a multiplexed
-// oriented transport such as TCP or UNIX domain stream sockets.
+// stream-oriented transport such as QUIC.
 func AdaptB(underlying MultiConn) Conn {
 	return &b {
 		sizeLimit:  defaultSizeLimit,
@@ -45,36 +45,63 @@ func (this *b) AcceptTrans() (Trans, error) {
 	return this.newTrans(stream), nil
 }
 
-func (this *b) SetSizeLimit(limit int) {
+func (this *b) SetSizeLimit(limit int64) {
 	this.sizeLimit = limit
 }
 
-func (this *b) newTrans(underlying Stream) transB {
+func (this *b) newTrans(underlying Stream) *transB {
-	return transB {
+	return &transB {
 		sizeLimit: this.sizeLimit,
 		underlying: underlying,
 	}
 }
 
 type transB struct {
-	sizeLimit int
+	sizeLimit int64
 	underlying Stream
+	currentData io.Reader
 }
 
-func (trans transB) Close() error {
+func (this *transB) Close() error {
-	return trans.underlying.Close()
+	return this.underlying.Close()
 }
 
-func (trans transB) ID() int64 {
+func (this *transB) ID() int64 {
-	return trans.underlying.ID()
+	return this.underlying.ID()
 }
 
-func (trans transB) Send(method uint16, data []byte) error {
+func (this *transB) Send(method uint16, data []byte) error {
-	return encodeMessageB(trans.underlying, trans.sizeLimit, method, data)
+	return encodeMessageB(this.underlying, this.sizeLimit, method, data)
 }
 
-func (trans transB) Receive() (uint16, []byte, error) {
+func (this *transB) Receive() (uint16, []byte, error) {
-	return decodeMessageB(trans.underlying, trans.sizeLimit)
+	// get a reader for the next message
+	method, size, data, err := this.receiveReader()
+	if err != nil { return 0, nil, err }
+	// read the entire thing
+	payloadBuffer := make([]byte, int(size))
+	_, err = io.ReadFull(data, payloadBuffer)
+	if err != nil { return 0, nil, err }
+	// we have used up the reader by now so we can forget it exists
+	this.currentData = nil
+	return method, payloadBuffer, nil
+}
+
+func (this *transB) ReceiveReader() (uint16, io.Reader, error) {
+	method, _, data, err := this.receiveReader()
+	return method, data, err
+}
+
+func (this *transB) receiveReader() (uint16, int64, io.Reader, error) {
+	// decode the message
+	method, size, data, err := decodeMessageB(this.underlying, this.sizeLimit)
+	if err != nil { return 0, 0, nil, err }
+	// discard current reader if there is one
+	if this.currentData == nil {
+		io.Copy(io.Discard, this.currentData)
+	}
+	this.currentData = data
+	return method, size, data, nil
 }
 
 // MultiConn represens a multiplexed stream-oriented transport for use in
@@ -98,8 +125,8 @@ type Stream interface {
 	ID() int64
 }
 
-func encodeMessageB(writer io.Writer, sizeLimit int, method uint16, data []byte) error {
+func encodeMessageB(writer io.Writer, sizeLimit int64, method uint16, data []byte) error {
-	if len(data) > sizeLimit {
+	if len(data) > int(sizeLimit) {
 		return ErrPayloadTooLarge
 	}
 	buffer := make([]byte, 10 + len(data))
@@ -110,19 +137,27 @@ func encodeMessageB(writer io.Writer, sizeLimit int, method uint16, data []byte)
 	return err
 }
 
-func decodeMessageB(reader io.Reader, sizeLimit int) (uint16, []byte, error) {
+func decodeMessageB(
+	reader io.Reader,
+	sizeLimit int64,
+) (
+	method uint16,
+	size   int64,
+	data   io.Reader,
+	err    error,
+) {
 	headerBuffer := [10]byte { }
-	_, err := io.ReadFull(reader, headerBuffer[:])
+	_, err = io.ReadFull(reader, headerBuffer[:])
-	if err != nil { return 0, nil, err }
+	if err != nil { return 0, 0, nil, err }
-	method, err := tape.DecodeI16[uint16](headerBuffer[:2])
+	method, err = tape.DecodeI16[uint16](headerBuffer[:2])
-	if err != nil { return 0, nil, err }
+	if err != nil { return 0, 0, nil, err }
 	length, err := tape.DecodeI64[uint64](headerBuffer[2:10])
-	if err != nil { return 0, nil, err }
+	if err != nil { return 0, 0, nil, err }
 	if length > uint64(sizeLimit) {
-		return 0, nil, ErrPayloadTooLarge
+		return 0, 0, nil, ErrPayloadTooLarge
 	}
-	payloadBuffer := make([]byte, int(length))
+	return method, int64(length), &io.LimitedReader {
-	_, err = io.ReadFull(reader, payloadBuffer)
+		R: reader,
-	if err != nil { return 0, nil, err }
+		N: int64(length),
-	return method, payloadBuffer, nil
+	}, nil
 }