cloud

Documentation

Index

• func GetTopics(bootstrapKafkaUrl string, regex *regexp.Regexp) (result []string, err error)
• func GetWorkerTopics(config configuration.Config) (result []string, err error)
• func ImportIdToTopic(id string) string
• func InitTopics(bootstrapUrl string, config []kafka.ConfigEntry, topics ...string) (err error)
• func NewKafkaLastOffsetConsumer(ctx context.Context, wg *sync.WaitGroup, broker string, groupId string, ...) error
• func NewKafkaLastOffsetConsumerGroup(ctx context.Context, wg *sync.WaitGroup, broker string, groupId string, ...) error
• func ServiceIdToTopic(id string) string
• func Start(basectx context.Context, wg *sync.WaitGroup, config configuration.Config, ...) error
• type Worker

Functions

func GetTopics

func GetTopics(bootstrapKafkaUrl string, regex *regexp.Regexp) (result []string, err error)

func GetWorkerTopics

func GetWorkerTopics(config configuration.Config) (result []string, err error)

func ImportIdToTopic

func ImportIdToTopic(id string) string

func InitTopics

func InitTopics(bootstrapUrl string, config []kafka.ConfigEntry, topics ...string) (err error)

func NewKafkaLastOffsetConsumer

func NewKafkaLastOffsetConsumer(ctx context.Context, wg *sync.WaitGroup, broker string, groupId string, topic string, listener func(delivery []byte) error, errhandler func(err error)) error

func NewKafkaLastOffsetConsumerGroup

func NewKafkaLastOffsetConsumerGroup(ctx context.Context, wg *sync.WaitGroup, broker string, groupId string, topics []string, listener func(msg model.ConsumerMessage) error, errhandler func(topic string, err error)) error

func ServiceIdToTopic

func ServiceIdToTopic(id string) string

func Start

func Start(basectx context.Context, wg *sync.WaitGroup, config configuration.Config, worker Worker) error

Types

type Worker

type Worker interface {
	Do(msg model.ConsumerMessage) error
}