129 lines
3.2 KiB
Go
129 lines
3.2 KiB
Go
package hopp
|
|
|
|
import "io"
|
|
import "net"
|
|
import "context"
|
|
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
|
|
underlying MultiConn
|
|
}
|
|
|
|
// AdaptB returns a connection implementing METADAPT-B over a singular stream-
|
|
// oriented transport such as TCP or UNIX domain stream sockets.
|
|
func AdaptB(underlying MultiConn) Conn {
|
|
return &b {
|
|
sizeLimit: defaultSizeLimit,
|
|
underlying: underlying,
|
|
}
|
|
}
|
|
|
|
func (this *b) Close() error {
|
|
return this.underlying.Close()
|
|
}
|
|
|
|
func (this *b) LocalAddr() net.Addr {
|
|
return this.underlying.LocalAddr()
|
|
}
|
|
|
|
func (this *b) RemoteAddr() net.Addr {
|
|
return this.underlying.RemoteAddr()
|
|
}
|
|
|
|
func (this *b) OpenTrans() (Trans, error) {
|
|
stream, err := this.underlying.OpenStream()
|
|
if err != nil { return nil, err }
|
|
return this.newTrans(stream), nil
|
|
}
|
|
|
|
func (this *b) AcceptTrans() (Trans, error) {
|
|
stream, err := this.underlying.AcceptStream(context.Background())
|
|
if err != nil { return nil, err }
|
|
return this.newTrans(stream), nil
|
|
}
|
|
|
|
func (this *b) SetSizeLimit(limit int) {
|
|
this.sizeLimit = limit
|
|
}
|
|
|
|
func (this *b) newTrans(underlying Stream) transB {
|
|
return transB {
|
|
sizeLimit: this.sizeLimit,
|
|
underlying: underlying,
|
|
}
|
|
}
|
|
|
|
type transB struct {
|
|
sizeLimit int
|
|
underlying Stream
|
|
}
|
|
|
|
func (trans transB) Close() error {
|
|
return trans.underlying.Close()
|
|
}
|
|
|
|
func (trans transB) ID() int64 {
|
|
return trans.underlying.ID()
|
|
}
|
|
|
|
func (trans transB) Send(method uint16, data []byte) error {
|
|
return encodeMessageB(trans.underlying, trans.sizeLimit, method, data)
|
|
}
|
|
|
|
func (trans transB) Receive() (uint16, []byte, error) {
|
|
return decodeMessageB(trans.underlying, trans.sizeLimit)
|
|
}
|
|
|
|
// MultiConn represens a multiplexed stream-oriented transport for use in
|
|
// [AdaptB].
|
|
type MultiConn interface {
|
|
// See documentation for [net.Conn].
|
|
io.Closer
|
|
LocalAddr() net.Addr
|
|
RemoteAddr() net.Addr
|
|
// AcceptStream accepts the next incoming stream from the other party.
|
|
AcceptStream(context.Context) (Stream, error)
|
|
// OpenStream opens a new stream.
|
|
OpenStream() (Stream, error)
|
|
}
|
|
|
|
// Stream represents a single stream returned by a [MultiConn].
|
|
type Stream interface {
|
|
// See documentation for [net.Conn].
|
|
io.ReadWriteCloser
|
|
// ID returns the stream ID
|
|
ID() int64
|
|
}
|
|
|
|
func encodeMessageB(writer io.Writer, sizeLimit int, method uint16, data []byte) error {
|
|
if len(data) > sizeLimit {
|
|
return ErrPayloadTooLarge
|
|
}
|
|
buffer := make([]byte, 10 + len(data))
|
|
tape.EncodeI16(buffer[:2], method)
|
|
tape.EncodeI64(buffer[2:10], uint64(len(data)))
|
|
copy(buffer[10:], data)
|
|
_, err := writer.Write(buffer)
|
|
return err
|
|
}
|
|
|
|
func decodeMessageB(reader io.Reader, sizeLimit int) (uint16, []byte, error) {
|
|
headerBuffer := [10]byte { }
|
|
_, err := io.ReadFull(reader, headerBuffer[:])
|
|
if err != nil { return 0, nil, err }
|
|
method, err := tape.DecodeI16[uint16](headerBuffer[:2])
|
|
if err != nil { return 0, nil, err }
|
|
length, err := tape.DecodeI64[uint64](headerBuffer[2:10])
|
|
if err != nil { return 0, nil, err }
|
|
if length > uint64(sizeLimit) {
|
|
return 0, nil, ErrPayloadTooLarge
|
|
}
|
|
payloadBuffer := make([]byte, int(length))
|
|
_, err = io.ReadFull(reader, payloadBuffer)
|
|
if err != nil { return 0, nil, err }
|
|
return method, payloadBuffer, nil
|
|
}
|