broker

Documentation

Index

• Constants
• type API
• type Client
• type Config
  • func NewConfig(api API) (Config, error)
• type ErrStorageFull
  • func (e ErrStorageFull) Error() string
• type Fs
• type MQTTProducer
  • func NewMQTTProducer(cfg Config) (*MQTTProducer, error)
  • func (p MQTTProducer) Serve(in MessageSource)
• type Message
  • func (m Message) Remove()
• type MessageLoader
  • func NewMessageLoader(dir string, fs Fs) MessageLoader
  • func (l MessageLoader) Output() <-chan Message
• type MessageSource
• type Persistor
  • func NewPersistor(dir string, fs Fs, conf <-chan *int64) *Persistor
  • func (p *Persistor) CreateMessage(m *Message) (err error)
• type Topic

Constants

const (
	Profile   Topic = "profiler"
	Event           = "events"
	Log             = "logs"
	DataPoint       = "datapoints"
)

Profile, Event, Log, and DataPoint are Message topics.

Types

type API

type API interface {
	backend.Credentialer
	BrokerAddress() (string, error)
	Certificates() (*tls.Config, error)
}

API consists of the backend interface needed to generate a Config.

type Client

type Client interface {
	Connect() mqtt.Token
	Publish(string, byte, bool, interface{}) mqtt.Token
	Disconnect(uint)
}

Client provides an MQTT client interface.

type Config

type Config struct {
	Creds  *backend.Credentials
	Client Client
}

Config provides parameters for an MQTTProducer.

func NewConfig

func NewConfig(api API) (Config, error)

NewConfig returns a Config from the given API.

type ErrStorageFull

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

ErrStorageFull indicates that the corresponding Persistor is full.

func (ErrStorageFull) Error

func (e ErrStorageFull) Error() string

Error returns e as a string.

type Fs

type Fs interface {
	Open(string) (afero.File, error)
	Stat(string) (os.FileInfo, error)
	MkdirAll(string, os.FileMode) error
	OpenFile(string, int, os.FileMode) (afero.File, error)
	Remove(path string) error
}

Fs provides file system functions.

type MQTTProducer

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

MQTTProducer wraps an MQTT Client.

func NewMQTTProducer

func NewMQTTProducer(cfg Config) (*MQTTProducer, error)

NewMQTTProducer returns a new producer for the given input.

func (MQTTProducer) Serve

func (p MQTTProducer) Serve(in MessageSource)

Serve launches p, enabling it to send and receive messages.

type Message

type Message struct {
	Error string `json:"error"`
	Topic Topic  `json:"topic"`
	Bytes []byte `json:"bytes"`
	// contains filtered or unexported fields
}

Message represents a broker message.

func (Message) Remove

func (m Message) Remove()

Remove deletes m from the persistence layer.

type MessageLoader

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

MessageLoader generates a stream of messages from the filesystem.

func NewMessageLoader

func NewMessageLoader(dir string, fs Fs) MessageLoader

NewMessageLoader reads dir for messages and returns them as a stream.

func (MessageLoader) Output

func (l MessageLoader) Output() <-chan Message

Output returns l's output stream.

type MessageSource

type MessageSource interface {
	// Output returns a channel of Messages provided by a Source. A source
	// indicates when it has no more Messages to send by closing the
	// channel.
	Output() <-chan Message
}

MessageSource is implemented by types that can generate a Message stream.

type Persistor

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

Persistor controls a persistence layer for Messages.

func NewPersistor

func NewPersistor(dir string, fs Fs, conf <-chan *int64) *Persistor

NewPersistor creates a new Persistor in dir.

func (*Persistor) CreateMessage

func (p *Persistor) CreateMessage(m *Message) (err error)

CreateMessage creates a new Message under p.

type Topic

type Topic string

Topic encodes a Message topic.