consumer

Documentation

Index

• type KafkaArgs
• type KafkaConsumer
  • func NewKafkaConsumer(kafkaBrokers string, kwargs ...KafkaArgs) (*KafkaConsumer, error)
  • func (sc *KafkaConsumer) GetStream(streamID string) (consumer.StreamConnection, error)
  • func (sc *KafkaConsumer) Start() error
  • func (sc *KafkaConsumer) Stop() error
• type KafkaStreamConnection
  • func NewKafkaStreamConnection(streamID string, streamType consts.StreamType, kafkaConsumer *cluster.Consumer, ...) *KafkaStreamConnection
  • func (sc *KafkaStreamConnection) AddConnection(quality consts.Quality, kafkaConsumer interface{}) error
  • func (sc *KafkaStreamConnection) AddDataToStream(data []byte, quality consts.Quality) error
  • func (sc *KafkaStreamConnection) Close(quality consts.Quality) error
  • func (sc *KafkaStreamConnection) CloseAll() error
  • func (sc *KafkaStreamConnection) GetOutputChan(quality consts.Quality) (<-chan []byte, error)
  • func (sc *KafkaStreamConnection) GetQualities() []consts.Quality
  • func (sc *KafkaStreamConnection) HandleStream(quality consts.Quality) error
  • func (sc *KafkaStreamConnection) IsOpen() bool

Types

type KafkaArgs

type KafkaArgs map[string]string

type KafkaConsumer

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

func NewKafkaConsumer

func NewKafkaConsumer(kafkaBrokers string, kwargs ...KafkaArgs) (*KafkaConsumer, error)

TODO add support for TLS Config

func (*KafkaConsumer) GetStream

func (sc *KafkaConsumer) GetStream(streamID string) (consumer.StreamConnection, error)

func (*KafkaConsumer) Start

func (sc *KafkaConsumer) Start() error

func (*KafkaConsumer) Stop

func (sc *KafkaConsumer) Stop() error

type KafkaStreamConnection

type KafkaStreamConnection struct {
	consumer.BaseStreamConnection
	// contains filtered or unexported fields
}

func NewKafkaStreamConnection

func NewKafkaStreamConnection(streamID string, streamType consts.StreamType, kafkaConsumer *cluster.Consumer, quality consts.Quality) *KafkaStreamConnection

func (*KafkaStreamConnection) AddConnection

func (sc *KafkaStreamConnection) AddConnection(quality consts.Quality, kafkaConsumer interface{}) error

func (*KafkaStreamConnection) AddDataToStream

func (sc *KafkaStreamConnection) AddDataToStream(data []byte, quality consts.Quality) error

func (*KafkaStreamConnection) Close

func (sc *KafkaStreamConnection) Close(quality consts.Quality) error

func (*KafkaStreamConnection) CloseAll

func (sc *KafkaStreamConnection) CloseAll() error

func (*KafkaStreamConnection) GetOutputChan

func (sc *KafkaStreamConnection) GetOutputChan(quality consts.Quality) (<-chan []byte, error)

func (*KafkaStreamConnection) GetQualities

func (sc *KafkaStreamConnection) GetQualities() []consts.Quality

func (*KafkaStreamConnection) HandleStream

func (sc *KafkaStreamConnection) HandleStream(quality consts.Quality) error

func (*KafkaStreamConnection) IsOpen

func (sc *KafkaStreamConnection) IsOpen() bool