packetio

Documentation

Overview

Package packetio provides streaming utilities.

Stream, ReadStream and WriteStream are tools for implementing streaming. State implements simple binary serialization of suspended I/O state.

Index

• Constants
• func Subscribe(s FinishSubscriber, operand int32) error
• func Write(w CloseWriter, packet EOFData) (int, error)
• type CloseWriter
• type EOFData
• type FinishSubscriber
• type ReadStream
  • func MakeReadStream() ReadStream
  • func NewReadStream() *ReadStream
  • func (s *ReadStream) FinishSubscription() error
  • func (s *ReadStream) Live() bool
  • func (s *ReadStream) Reading() bool
  • func (s *ReadStream) StopTransfer()
  • func (s *ReadStream) Subscribe(increment int32) error
  • func (s *ReadStream) Transfer(ctx Context, config packet.Service, streamID int32, send chan<- packet.Thunk, ...) error
• type State
  • func InitialState() State
  • func InitialStateWithDataBuffer(databuf []byte) State
  • func (s *State) Live() bool
  • func (s *State) Marshal(dest []byte) []byte
  • func (s *State) MarshaledSize() (n int)
  • func (s *State) Unmarshal(src []byte, maxDataSize int) ([]byte, error)
• type Stream
  • func MakeStream(writeBufferSize int) Stream
  • func NewStream(writeBufferSize int) *Stream
  • func (s *Stream) Live() bool
  • func (s *Stream) Marshal(dest []byte) []byte
  • func (s *Stream) MarshaledSize() int
  • func (s *Stream) StopTransfer()
  • func (s *Stream) Transfer(ctx Context, config packet.Service, streamID int32, r io.Reader, ...) error
  • func (s *Stream) Unmarshal(src []byte, config packet.Service) ([]byte, error)
• type WriteStream
  • func MakeWriteStream(bufferSize int) WriteStream
  • func NewWriteStream(bufferSize int) *WriteStream
  • func (s *WriteStream) CloseWrite() error
  • func (s *WriteStream) Live() bool
  • func (s *WriteStream) StopTransfer()
  • func (s *WriteStream) Transfer(ctx Context, config packet.Service, streamID int32, w io.WriteCloser, ...) error
  • func (s *WriteStream) Write(data []byte) (int, error)
  • func (s *WriteStream) Writing() bool

Constants

const (
	FlagSubscribing = 1 << iota
	FlagReadWriting
	FlagSendReceiving
)

State flags.

Functions

func Subscribe

func Subscribe(s FinishSubscriber, operand int32) error

Subscribe to more data or signal that no more data will be subscribed to. The operand must be non-negative.

func Write

func Write(w CloseWriter, packet EOFData) (int, error)

Write a data packet to a stream.

io.EOF is returned only if it is returned by the writer. If the input packet represents the end-of-file condition, zero length and nil error are returned on success.

Types

type CloseWriter

type CloseWriter interface {
	io.Writer
	CloseWrite() error
}

CloseWriter can signal end-of-file condition to its peer.

type EOFData

type EOFData interface {
	Data() []byte
	EOF() bool
}

EOFData contains data or represents the end-of-file condition.

Data() must return an empty slice (or nil) when EOF() returns true.

type FinishSubscriber

type FinishSubscriber interface {
	Subscribe(increment int32) error
	FinishSubscription() error
}

FinishSubscriber can subscribe to more data and signal that no more data will be subscribed to.

type ReadStream

type ReadStream struct {
	State State
	// contains filtered or unexported fields
}

ReadStream is a unidirectional stream between a reader and a channel.

The channel side calls Subscribe, FinishSubscription, and StopTransfer. The reader side calls Transfer.

State can be unmarshaled before subscription and transfer, and it can be marshaled afterwards.

func MakeReadStream

func MakeReadStream() ReadStream

MakeReadStream is useful for initializing a field. Don't use multiple copies of the value.

func NewReadStream

func NewReadStream() *ReadStream

NewReadStream object. Use MakeReadStream when embedding ReadStream in a struct.

func (*ReadStream) FinishSubscription

func (s *ReadStream) FinishSubscription() error

FinishSubscription signals that no more data will be subscribed to.

func (*ReadStream) Live

func (s *ReadStream) Live() bool

Live state?

The state is undefined during subscription or transfer.

func (*ReadStream) Reading

func (s *ReadStream) Reading() bool

Reading state? When the stream is no longer in the reading state, reader may be specified as nil in Transfer invocation.

The state is undefined during transfer.

func (*ReadStream) StopTransfer

func (s *ReadStream) StopTransfer()

StopTransfer the transfer. The subscribe methods must not be called after this.

func (*ReadStream) Subscribe

func (s *ReadStream) Subscribe(increment int32) error

Subscribe to more data. Causes the concurrent transfer to read and send data.

func (*ReadStream) Transfer

func (s *ReadStream) Transfer(ctx Context, config packet.Service, streamID int32, send chan<- packet.Thunk, r io.Reader) error

Transfer data from a reader to a service's data stream according to subscription. Buffer size is limited by config.MaxSendSize.

Read or context error is returned, excluding EOF.

type State

type State struct {
	Flags      uint32
	Subscribed uint32
	Data       []byte
}

State of a suspended unidirectional stream.

Zero value represents a final state, not the initial state.

func InitialState

func InitialState() State

func InitialStateWithDataBuffer

func InitialStateWithDataBuffer(databuf []byte) State

func (*State) Live

func (s *State) Live() bool

Live?

func (*State) Marshal

func (s *State) Marshal(dest []byte) []byte

Marshal the state into dest buffer. len(dest) must be at least s.MarshaledSize(). The unused tail of dest is returned.

func (*State) MarshaledSize

func (s *State) MarshaledSize() (n int)

MarshaledSize of the state.

func (*State) Unmarshal

func (s *State) Unmarshal(src []byte, maxDataSize int) ([]byte, error)

Unmarshal state from src buffer. State might keep a reference to src, unless a preallocated buffer was passed to InitialStateWithDataBuffer. The unconsumed tail of src is returned.

maxDataSize is the Data buffer size limit. For ReadStream state, it should be config.MaxSendSize where config is the Transfer argument. For WriteStream state, it should match the stream's bufsize.

type Stream

type Stream struct {
	ReadStream
	WriteStream
}

Stream is a bidirectional stream between a connection and a channel.

The channel side calls Subscribe, FinishSubscription, Write, CloseWrite, and StopTransfer. The connection side calls Transfer.

Unmarshal may be called before subscription, writing and transfer, and Marshal may be called afterwards.

func MakeStream

func MakeStream(writeBufferSize int) Stream

MakeStream is useful for initializing a field. Don't use multiple copies of the value.

Write buffer size must be a power of two.

func NewStream

func NewStream(writeBufferSize int) *Stream

NewStream object. Use MakeStream when embedding Stream in a struct.

Write buffer size must be a power of two.

func (*Stream) Live

func (s *Stream) Live() bool

Live state.

The state is undefined during subscription, writing, or transfer.

func (*Stream) Marshal

func (s *Stream) Marshal(dest []byte) []byte

Marshal the state into dest buffer. len(dest) must be at least s.MarshaledSize(). The unused tail of dest is returned.

func (*Stream) MarshaledSize

func (s *Stream) MarshaledSize() int

MarshaledSize of the state.

func (*Stream) StopTransfer

func (s *Stream) StopTransfer()

func (*Stream) Transfer

func (s *Stream) Transfer(ctx Context, config packet.Service, streamID int32, r io.Reader, w io.WriteCloser, send chan<- packet.Thunk) error

Transfer data bidirectionally between a connection (r, w) and a user program's stream. Read buffer size is limited by config.MaxSendSize.

I/O or context errors are returned, excluding EOF.

func (*Stream) Unmarshal

func (s *Stream) Unmarshal(src []byte, config packet.Service) ([]byte, error)

Unmarshal state from src buffer. Stream might keep a reference to src until Transfer is called. The unconsumed tail of src is returned.

type WriteStream

type WriteStream struct {
	State State
	// contains filtered or unexported fields
}

WriteStream is a unidirectional stream between a channel and a writer.

The channel side calls Write, CloseWrite, and StopTransfer. The writer side calls Transfer.

State can be unmarshaled before writing and transfer, and it can be marshaled afterwards.

func MakeWriteStream

func MakeWriteStream(bufferSize int) WriteStream

MakeWriteStream is useful for initializing a field. Don't use multiple copies of the value.

Buffer size must be a power of two.

func NewWriteStream

func NewWriteStream(bufferSize int) *WriteStream

NewWriteStream object. Use MakeWriteStream when embedding WriteStream in a struct.

Buffer size must be a power of two.

func (*WriteStream) CloseWrite

func (s *WriteStream) CloseWrite() error

CloseWrite signals that no more data will be written.

func (*WriteStream) Live

func (s *WriteStream) Live() bool

Live state?

The state is undefined during writing or transfer.

func (*WriteStream) StopTransfer

func (s *WriteStream) StopTransfer()

StopTransfer the transfer. The write methods must not be called after this.

func (*WriteStream) Transfer

func (s *WriteStream) Transfer(ctx Context, config packet.Service, streamID int32, w io.WriteCloser, send chan<- packet.Thunk) error

Transfer data from a service's data stream while managing its flow.

Write or context error is returned, excluding EOF.

func (*WriteStream) Write

func (s *WriteStream) Write(data []byte) (int, error)

Write all data or return an error.

func (*WriteStream) Writing

func (s *WriteStream) Writing() bool

Writing state? When the stream is no longer in the writing state, writer may be specified as nil in Transfer invocation.

The state is undefined during transfer.