streaming

Documentation

Index

• type Consumer
  • func NewS3Consumer(endpoint, region, bucket string) (Consumer, error)
• type Error
• type Message
• type Messages
• type Producer
  • func NewKafkaProducer(brokers []string, version string, retry int) (Producer, error)
  • func NewS3Producer(endpoint, region, bucket string) (Producer, error)

Types

type Consumer

type Consumer interface {
	// Output gets messages until the given date.
	Output(endTime time.Time) (<-chan Messages, <-chan error)
	// Close closes the producer.
	Close() error
	// IsHealthy checks the health of the Consumer.
	IsHealthy() bool
}

Consumer represents a class that can consume messages.

func NewS3Consumer

func NewS3Consumer(endpoint, region, bucket string) (Consumer, error)

NewS3Consumer creates a consumer that gets messages to AWS S3.

type Error

type Error struct {
	Msgs []*Message
	Err  error
}

Error is the error type returned by a Producer when an error occurs while sending messages.

type Message

type Message struct {
	Topic string
	Key   []byte
	Data  []byte
}

Message is the information to be sent through a Producer.

type Messages

type Messages []*Message

Messages is an array of messages.

type Producer

type Producer interface {
	// Name is the name of the producer.
	Name() string
	// Input is the message input channel.
	Input() chan<- *Message
	// Errors is the error output channel.
	Errors() <-chan *Error
	// Close closes the producer.
	Close() error
	// IsHealthy checks the health of the producer.
	IsHealthy() bool
}

Producer represents a class that can send messages.

func NewKafkaProducer

func NewKafkaProducer(brokers []string, version string, retry int) (Producer, error)

NewKafkaProducer creates a new producer that sends messages to Kafka.

func NewS3Producer

func NewS3Producer(endpoint, region, bucket string) (Producer, error)

NewS3Producer creates a producer that sends messages to AWS S3.