quacktorstreams

Documentation

Index

• func NewProducer(producer Producer, topic string) *quacktors.Pid
• type Consumer
• type ConsumerActor
  • func NewConsumer(consumer Consumer) (*ConsumerActor, *quacktors.Pid)
  • func (c *ConsumerActor) Init(ctx *quacktors.Context)
  • func (c *ConsumerActor) Run(ctx *quacktors.Context, message quacktors.Message)
  • func (c *ConsumerActor) Subscribe(topic string, handler *quacktors.Pid, ...) error
• type Producer
• type ProducerActor
  • func (p *ProducerActor) Init(ctx *quacktors.Context)
  • func (p *ProducerActor) Run(ctx *quacktors.Context, message quacktors.Message)
• type StreamMessage

Functions

func NewProducer

func NewProducer(producer Producer, topic string) *quacktors.Pid

NewProducer creates a new producer actor by a consumer implementation and returns the PID of the ProducerActor. When a message is sent to the ProducerActor, it is automatically forwarded (i.e. published) to the provided topic in the stream.

Types

type Consumer

type Consumer interface {
	Init() error
	Subscribe(topic string) error
	NextMessage() (StreamMessage, error)
}

type ConsumerActor

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

func NewConsumer

func NewConsumer(consumer Consumer) (*ConsumerActor, *quacktors.Pid)

NewConsumer creates a new ConsumerActor by a consumer implementation and returns both a pointer to the ConsumerActor itself and the PID of the consumer. The pointer to the ConsumerActor can be used to subscribe to topics of the stream.

func (*ConsumerActor) Init

func (c *ConsumerActor) Init(ctx *quacktors.Context)

func (*ConsumerActor) Run

func (c *ConsumerActor) Run(ctx *quacktors.Context, message quacktors.Message)

func (*ConsumerActor) Subscribe

func (c *ConsumerActor) Subscribe(topic string, handler *quacktors.Pid, conv func([]byte) (quacktors.Message, error)) error

type Producer

type Producer interface {
	Init() error
	SetTopic(topic string)
	Emit(message quacktors.Message)
}

type ProducerActor

type ProducerActor struct {
	Producer
}

func (*ProducerActor) Init

func (p *ProducerActor) Init(ctx *quacktors.Context)

func (*ProducerActor) Run

func (p *ProducerActor) Run(ctx *quacktors.Context, message quacktors.Message)

type StreamMessage

type StreamMessage struct {
	Bytes []byte
	Topic string
	Meta  interface{}
}