message

Documentation

Overview

Package message implements types for manipulating streams of broker messages.

Index

• type DataLimiter
  • func NewDataLimiter(store Persistor, conf <-chan api.CellularConfig, src ...broker.MessageSource) *DataLimiter
  • func (l *DataLimiter) Decode(r io.Reader) (err error)
  • func (l *DataLimiter) Encode(w io.Writer) (err error)
  • func (l *DataLimiter) Output() <-chan broker.Message
• type Decodable
• type Encodable
• type FilePersistor
  • func (fp FilePersistor) Load(object Decodable) error
  • func (fp FilePersistor) Save(object Encodable) error
• type MemPersistor
  • func (mp *MemPersistor) Load(object Decodable) error
  • func (mp *MemPersistor) Save(object Encodable) error
• type Merger
  • func Merge(src ...broker.MessageSource) Merger
  • func (m Merger) Output() <-chan broker.Message
• type Persistor

Types

type DataLimiter

type DataLimiter struct {

	// Budget is how many bytes can be transmitted per period.
	// If HasBudget is false, any number of bytes can be transmitted.
	Budget    int  `json:"budget"`
	HasBudget bool `json:"hasBudget"`

	// Count is how many bytes have been transmitted during the current
	// period.
	Count int `json:"count"`

	// PeriodEnd marks the end of the current period.
	PeriodEnd time.Time `json:"periodEnd"`
	// contains filtered or unexported fields
}

DataLimiter is a passthrough that limits the number of application-layer bytes transmitted per period.

func NewDataLimiter

func NewDataLimiter(
	store Persistor,
	conf <-chan api.CellularConfig,
	src ...broker.MessageSource,
) *DataLimiter

NewDataLimiter returns a DataLimiter for input whose state persists on the filesystem.

func (*DataLimiter) Decode

func (l *DataLimiter) Decode(r io.Reader) (err error)

Decode updates l's state by reading bytes from r.

func (*DataLimiter) Encode

func (l *DataLimiter) Encode(w io.Writer) (err error)

Encode writes the l's state to w.

func (*DataLimiter) Output

func (l *DataLimiter) Output() <-chan broker.Message

Output returns a channel on which messages can be received. The channel closes when l's input closes.

type Decodable

type Decodable interface {
	Decode(io.Reader) error
}

Decodable is an object that can decode itself from a byte stream.

type Encodable

type Encodable interface {
	Encode(io.Writer) error
}

Encodable is an object that can encode itself to a byte stream.

type FilePersistor

type FilePersistor struct {
	Path string
}

FilePersistor permits saving and loading data from a file.

func (FilePersistor) Load

func (fp FilePersistor) Load(object Decodable) error

Load decodes object from the underlying file.

func (FilePersistor) Save

func (fp FilePersistor) Save(object Encodable) error

Save encodes object to the underlying file.

type MemPersistor

type MemPersistor struct {
	Buf bytes.Buffer
}

MemPersistor permits saving and loading data from an internal memory buffer.

func (*MemPersistor) Load

func (mp *MemPersistor) Load(object Decodable) error

Load decodes object from mp's internal buffer.

func (*MemPersistor) Save

func (mp *MemPersistor) Save(object Encodable) error

Save encodes object to mp's internal buffer.

type Merger

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

Merger is a broker.MessageSource that merges multiple other MessageSources into one stream.

func Merge

func Merge(src ...broker.MessageSource) Merger

Merge returns a Merger that merges the streams of each element in src.

func (Merger) Output

func (m Merger) Output() <-chan broker.Message

Output returns m's output channel. It closes when all input streams have closed.

type Persistor

type Persistor interface {
	Save(Encodable) error
	Load(Decodable) error
}

Persistor can save and load an object to some kind of storage.