kafka

Documentation

Index

• func BuildDLQTopicForTopic(topic string) string
• func ConsumeFromTopic(ctx context.Context, cancel context.CancelFunc, groupID string, topic string, ...)
• func MarshalKafkaDataToStruct(data []byte, structToMarshal interface{}) error
• func NewReader(cfg kafka.ReaderConfig, log logger.Logger) *kafka.Reader
• func NewWriter(writer kafka.Writer, log logger.Logger) *kafka.Writer
• func PublishToDLQ(ctx context.Context, writer *kafka.Writer, dlqTopic string, ...) error
• func UnmarshalToKafkaMessage(data []byte) (models.KafkaMessage, error)
• func WriteToTopicInBackground(ctx context.Context, writer *kafka.Writer, topic string, key string, ...) error
• type ConsumerReaderConfig
• type ConsumerWriterConfig
• type Reader
• type WriterConfig

Functions

func BuildDLQTopicForTopic

func BuildDLQTopicForTopic(topic string) string

func ConsumeFromTopic

func ConsumeFromTopic(
	ctx context.Context,
	cancel context.CancelFunc,
	groupID string,
	topic string,
	workersNum int,
	workerFunc func(ctx context.Context, cancel context.CancelFunc, r *kafka.Reader, w *kafka.Writer, wg *sync.WaitGroup, workerID int),
	readerConfig ConsumerReaderConfig,
	writerConfig ConsumerWriterConfig,
	logger logger.Logger,
)

func MarshalKafkaDataToStruct

func MarshalKafkaDataToStruct(data []byte, structToMarshal interface{}) error

func NewReader

func NewReader(cfg kafka.ReaderConfig, log logger.Logger) *kafka.Reader

func NewWriter

func NewWriter(writer kafka.Writer, log logger.Logger) *kafka.Writer

func PublishToDLQ

func PublishToDLQ(ctx context.Context, writer *kafka.Writer, dlqTopic string, message models.KafkaMessage, error string) error

func UnmarshalToKafkaMessage

func UnmarshalToKafkaMessage(data []byte) (models.KafkaMessage, error)

func WriteToTopicInBackground

func WriteToTopicInBackground(
	ctx context.Context,
	writer *kafka.Writer,
	topic string,
	key string,
	value models.KafkaMessage,
	log logger.Logger,
) error

Types

type ConsumerReaderConfig

type ConsumerReaderConfig struct {
	Brokers                []string
	MinBytes               int
	MaxBytes               int
	QueueCapacity          int
	HeartbeatInterval      time.Duration
	CommitInterval         time.Duration
	MaxAttempts            int
	DialTimeout            time.Duration
	PartitionWatchInterval time.Duration
}

type ConsumerWriterConfig

type ConsumerWriterConfig struct {
	Brokers      []string
	RequiredAcks int
	MaxAttempts  int
	ReadTimeout  time.Duration
	WriteTimeout time.Duration
}

type Reader

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

type WriterConfig

type WriterConfig struct {
	Logger logger.Logger
}