utils

Documentation

Index

• Constants
• func DecodeUint16(buf []byte, offset int) (uint16, int)
• func DecodeUint32(buf []byte, offset int) (uint32, int)
• func DecodeUint64(buf []byte, offset int) (uint64, int)
• func EncodeACKMessage(seqNum uint64) []byte
• func EncodeAttachFromOffsetMessage(topic string, topicOffset uint64) []byte
• func EncodeAttachMessage(topic string) []byte
• func EncodeAttachedMessage(topic string, topicOffset uint64) []byte
• func EncodeBytes(buf []byte, offset int, b []byte) int
• func EncodeDataMessage(topic string, topicOffset uint64, data []byte) []byte
• func EncodeDataMessagePrefix(topic string, topicOffset uint64, data []byte) []byte
• func EncodeDetachMessage(topic string) []byte
• func EncodeDetachedMessage(topic string) []byte
• func EncodeHeader(buf []byte, offset int, messageType MessageType, payloadLen uint32) int
• func EncodeMessageType(buf []byte, offset int, t MessageType) int
• func EncodePingMessage(timestamp uint64) []byte
• func EncodePongMessage(timestamp uint64) []byte
• func EncodePublishMessage(topic string, seqNum uint64, data []byte) []byte
• func EncodePublishMessagePrefix(topic string, seqNum uint64, data []byte) []byte
• func EncodeUint16(buf []byte, offset int, n uint16) int
• func EncodeUint32(buf []byte, offset int, n uint32) int
• func EncodeUint64(buf []byte, offset int, n uint64) int
• type BufferedReader
  • func NewBufferedReader(r io.Reader, bufLen int) *BufferedReader
  • func (r *BufferedReader) Read() (MessageType, []byte, error)
• type BufferedWriter
  • func NewBufferedWriter(w io.Writer) *BufferedWriter
  • func (w *BufferedWriter) Close() error
  • func (w *BufferedWriter) Write(bufs ...[]byte) error
• type FakeConn
  • func NewFakeConn() *FakeConn
  • func (c *FakeConn) Close() error
  • func (c *FakeConn) LocalAddr() net.Addr
  • func (c *FakeConn) NextWritten() []byte
  • func (c *FakeConn) Push(b []byte)
  • func (c *FakeConn) Read(b []byte) (int, error)
  • func (c *FakeConn) RemoteAddr() net.Addr
  • func (c *FakeConn) SetDeadline(t time.Time) error
  • func (c *FakeConn) SetReadDeadline(t time.Time) error
  • func (c *FakeConn) SetWriteDeadline(t time.Time) error
  • func (c *FakeConn) Write(b []byte) (n int, err error)
• type MessageType
  • func DecodeHeader(buf []byte) (MessageType, int, bool)
  • func DecodeMessageType(buf []byte, offset int) (MessageType, int)
  • func (t MessageType) String() string
• type NetworkConnection

Constants

const (
	HeaderLen = 8

	FlagNone      = uint16(0)
	FlagUseOffset = uint16(1 << 15)
)

const (
	TypeAttach   = MessageType(1)
	TypeAttached = MessageType(2)
	TypeDetach   = MessageType(3)
	TypeDetached = MessageType(4)
	TypePublish  = MessageType(5)
	TypeACK      = MessageType(6)
	TypeData     = MessageType(7)
	TypePing     = MessageType(8)
	TypePong     = MessageType(9)
)

Functions

func DecodeUint16

func DecodeUint16(buf []byte, offset int) (uint16, int)

func DecodeUint32

func DecodeUint32(buf []byte, offset int) (uint32, int)

func DecodeUint64

func DecodeUint64(buf []byte, offset int) (uint64, int)

func EncodeACKMessage

func EncodeACKMessage(seqNum uint64) []byte

func EncodeAttachFromOffsetMessage

func EncodeAttachFromOffsetMessage(topic string, topicOffset uint64) []byte

func EncodeAttachMessage

func EncodeAttachMessage(topic string) []byte

func EncodeAttachedMessage

func EncodeAttachedMessage(topic string, topicOffset uint64) []byte

func EncodeBytes

func EncodeBytes(buf []byte, offset int, b []byte) int

func EncodeDataMessage

func EncodeDataMessage(topic string, topicOffset uint64, data []byte) []byte

Note avoid using, should use EncodeDataMessagePrefix instead to avoid copying data.

func EncodeDataMessagePrefix

func EncodeDataMessagePrefix(topic string, topicOffset uint64, data []byte) []byte

EncodeDataMessagePrefix returns a data message excluding the data itself. This lets us use writeev and avoid copying data twice.

func EncodeDetachMessage

func EncodeDetachMessage(topic string) []byte

func EncodeDetachedMessage

func EncodeDetachedMessage(topic string) []byte

func EncodeHeader

func EncodeHeader(buf []byte, offset int, messageType MessageType, payloadLen uint32) int

func EncodeMessageType

func EncodeMessageType(buf []byte, offset int, t MessageType) int

func EncodePingMessage

func EncodePingMessage(timestamp uint64) []byte

func EncodePongMessage

func EncodePongMessage(timestamp uint64) []byte

func EncodePublishMessage

func EncodePublishMessage(topic string, seqNum uint64, data []byte) []byte

func EncodePublishMessagePrefix

func EncodePublishMessagePrefix(topic string, seqNum uint64, data []byte) []byte

func EncodeUint16

func EncodeUint16(buf []byte, offset int, n uint16) int

func EncodeUint32

func EncodeUint32(buf []byte, offset int, n uint32) int

func EncodeUint64

func EncodeUint64(buf []byte, offset int, n uint64) int

Types

type BufferedReader

type BufferedReader struct {
	// contains filtered or unexported fields
}

BufferedReader reads full protocol messages from the given reader.

This is NOT thread safe.

func NewBufferedReader

func NewBufferedReader(r io.Reader, bufLen int) *BufferedReader

func (*BufferedReader) Read

func (r *BufferedReader) Read() (MessageType, []byte, error)

Read reads a protocol message from the underlying reader, returning the message type and payload. This keeps reading more until it has a full protocol message.

type BufferedWriter

type BufferedWriter struct {
	// contains filtered or unexported fields
}

BufferedWriter handles writing to the writer in a background thread to avoid blocking Write.

func NewBufferedWriter

func NewBufferedWriter(w io.Writer) *BufferedWriter

func (*BufferedWriter) Close

func (w *BufferedWriter) Close() error

func (*BufferedWriter) Write

func (w *BufferedWriter) Write(bufs ...[]byte) error

type FakeConn

type FakeConn struct {
	// contains filtered or unexported fields
}

FakeConn is a fake network connection used for testing. This is thread safe as may be accessed by a backgroud read/write goroutine.

func NewFakeConn

func NewFakeConn() *FakeConn

func (*FakeConn) Close

func (c *FakeConn) Close() error

func (*FakeConn) LocalAddr

func (c *FakeConn) LocalAddr() net.Addr

func (*FakeConn) NextWritten

func (c *FakeConn) NextWritten() []byte

NextWritten returns the next write to the connection. This blocks until received.

func (*FakeConn) Push

func (c *FakeConn) Push(b []byte)

Push adds a new buffer for read to return.

func (*FakeConn) Read

func (c *FakeConn) Read(b []byte) (int, error)

func (*FakeConn) RemoteAddr

func (c *FakeConn) RemoteAddr() net.Addr

func (*FakeConn) SetDeadline

func (c *FakeConn) SetDeadline(t time.Time) error

func (*FakeConn) SetReadDeadline

func (c *FakeConn) SetReadDeadline(t time.Time) error

func (*FakeConn) SetWriteDeadline

func (c *FakeConn) SetWriteDeadline(t time.Time) error

func (*FakeConn) Write

func (c *FakeConn) Write(b []byte) (n int, err error)

type MessageType

type MessageType uint16

func DecodeHeader

func DecodeHeader(buf []byte) (MessageType, int, bool)

func DecodeMessageType

func DecodeMessageType(buf []byte, offset int) (MessageType, int)

func (MessageType) String

func (t MessageType) String() string

type NetworkConnection

type NetworkConnection interface {
	io.Reader
	io.Writer
	io.Closer
}