messages

Documentation

Index

• type KafkaConnection
  • func NewKafkaConnection(network string, host string) (KafkaConnection, error)
• type MessageProvider
  • func NewFakeProvider(conn KafkaConnection, writerFunc NewWriterFunc, readerFunc NewReaderFunc) MessageProvider
  • func NewMessageProvider(host string) (MessageProvider, error)
• type Messenger
  • func NewMessenger(invokerNumber int, kafkaAddress string) (*Messenger, error)
  • func (m *Messenger) Close() error
  • func (m *Messenger) GetActivation() (*types.ActivationMessage, error)
  • func (m *Messenger) SendCompletion(activation *types.ActivationMessage, function *types.FunctionDoc, ...) error
  • func (m *Messenger) SendResult(activation *types.ActivationMessage, function *types.FunctionDoc, ...) error
  • func (m *Messenger) StartHealthPings(invokerNumber int) chan bool
• type NewReaderFunc
• type NewWriterFunc
• type Ping
• type Reader
  • func NewReader(host string, topic string) Reader
• type Writer
  • func NewWriter(host string, topic string) Writer

Types

type KafkaConnection

type KafkaConnection interface {
	CreateTopics(...kafka.TopicConfig) error
	Close() error
}

func NewKafkaConnection

func NewKafkaConnection(network string, host string) (KafkaConnection, error)

type MessageProvider

type MessageProvider interface {
	EnsureTopic(string) error
	WriteMessage(string, []byte) error
	ReadMessage(string) ([]byte, error)
	Close() error
}

func NewFakeProvider

func NewFakeProvider(conn KafkaConnection, writerFunc NewWriterFunc, readerFunc NewReaderFunc) MessageProvider

func NewMessageProvider

func NewMessageProvider(host string) (MessageProvider, error)

type Messenger

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

func NewMessenger

func NewMessenger(invokerNumber int, kafkaAddress string) (*Messenger, error)

func (*Messenger) Close

func (m *Messenger) Close() error

func (*Messenger) GetActivation

func (m *Messenger) GetActivation() (*types.ActivationMessage, error)

func (*Messenger) SendCompletion

func (m *Messenger) SendCompletion(activation *types.ActivationMessage, function *types.FunctionDoc, result interface{}) error

func (*Messenger) SendResult

func (m *Messenger) SendResult(activation *types.ActivationMessage, function *types.FunctionDoc, result interface{}) error

func (*Messenger) StartHealthPings

func (m *Messenger) StartHealthPings(invokerNumber int) chan bool

type NewReaderFunc

type NewReaderFunc func(string, string) Reader

type NewWriterFunc

type NewWriterFunc func(string, string) Writer

type Ping

type Ping struct {
	Name types.Name `json:"name"`
}

type Reader

type Reader interface {
	ReadMessage(context.Context) (kafka.Message, error)
	Close() error
}

func NewReader

func NewReader(host string, topic string) Reader

type Writer

type Writer interface {
	WriteMessages(context.Context, ...kafka.Message) error
	Close() error
}

func NewWriter

func NewWriter(host string, topic string) Writer