inmem

Documentation

Index

• func NewClient(options ...func(*Client)) stream.Client
• func NewClientWithStore(store *Store, options ...func(*Client)) stream.Client
• type Client
  • func (c *Client) NewConsumer() stream.Consumer
  • func (c *Client) NewConsumerAndProducer() (stream.Consumer, stream.Producer)
  • func (c *Client) NewProducer() stream.Producer
• type Consumer
  • func (c *Consumer) Close() error
  • func (c *Consumer) Messages() <-chan *stream.Message
• type Message
• type Partition
  • func (p *Partition) Messages() []*Message
• type Producer
  • func (p *Producer) Close() error
  • func (p *Producer) Messages() chan<- *stream.Message
  • func (p *Producer) PartitionKey(f func(*stream.Message) []byte)
• type Store
  • func NewStore() *Store
  • func (s *Store) NewTopic(name string) *Topic
• type Topic
  • func (t *Topic) Messages() []*Message
  • func (t *Topic) NewMessage(msg []byte, key []byte)

Functions

func NewClient

func NewClient(options ...func(*Client)) stream.Client

NewClient returns a new inmem client.

func NewClientWithStore

func NewClientWithStore(store *Store, options ...func(*Client)) stream.Client

NewClientWithStore returns a new inmem client with the predefined store.

Types

type Client

type Client struct {
	ConsumerTopic string
	ProducerTopic string

	// Logger is the configurable logger instance to log messages from this
	// streamclient. If left undefined, a noop logger will be used.
	Logger *zap.Logger
	// contains filtered or unexported fields
}

Client provides access to the streaming capabilities.

func (*Client) NewConsumer

func (c *Client) NewConsumer() stream.Consumer

NewConsumer returns a consumer that can iterate over messages on a stream.

func (*Client) NewConsumerAndProducer

func (c *Client) NewConsumerAndProducer() (stream.Consumer, stream.Producer)

NewConsumerAndProducer is a convenience method that returns both a consumer and a producer, with a single function call.

func (*Client) NewProducer

func (c *Client) NewProducer() stream.Producer

NewProducer returns a producer that outputs messages to stdout.

type Consumer

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

Consumer implements the stream.Consumer interface for standardstream.

func (*Consumer) Close

func (c *Consumer) Close() error

Close closes the consumer connection.

func (*Consumer) Messages

func (c *Consumer) Messages() <-chan *stream.Message

Messages returns the read channel for the messages that are returned by the stream.

type Message

type Message struct {
	Key   []byte
	Value []byte
}

Message is a single message in the store

type Partition

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

Partition is a single instance of a store, containing messages.

func (*Partition) Messages

func (p *Partition) Messages() []*Message

Messages returns all messages in a partition.

type Producer

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

Producer represents the object that will produce messages to a stream.

func (*Producer) Close

func (p *Producer) Close() error

Close closes the producer connection

func (*Producer) Messages

func (p *Producer) Messages() chan<- *stream.Message

Messages returns the write channel for messages to be produced.

func (*Producer) PartitionKey

func (p *Producer) PartitionKey(f func(*stream.Message) []byte)

PartitionKey can be used to define the key to use for partitioning messages.

type Store

type Store struct {
	Topics map[string]*Topic
	// contains filtered or unexported fields
}

Store hold the in-memory representation of a data storage service.

var DefaultStore *Store

DefaultStore is the global default store.

func NewStore

func NewStore() *Store

NewStore initializes a new empty in-memory store.

func (*Store) NewTopic

func (s *Store) NewTopic(name string) *Topic

NewTopic returns a new topic, or existing one, if it exists.

type Topic

type Topic struct {
	Partitions map[string]*Partition
	// contains filtered or unexported fields
}

Topic contains all the messages

func (*Topic) Messages

func (t *Topic) Messages() []*Message

Messages returns all messages in all Partitions as a map.

func (*Topic) NewMessage

func (t *Topic) NewMessage(msg []byte, key []byte)

NewMessage creates a new message in a topic.