kafkastreamer

package module

v0.0.0-...-d38f95b Latest Latest Go to latest Published: Feb 11, 2025 License: BSD-3-Clause Imports: 6 Imported by: 1

Details

This section is empty.

This section is empty.

func NewConnector(c kafka.ReaderConfig, t Translator) *connector

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

func (c *Consumer) Close() error

func (c *Consumer) Recv(ctx context.Context) (*workflow.Event, workflow.Ack, error)

type Producer struct {
	Topic         string
	Writer        *kafka.Writer
	WriterTimeout time.Duration
}

func (p *Producer) Close() error

func (p *Producer) Send(ctx context.Context, foreignID string, statusType int, headers map[workflow.Header]string) error

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

func New(brokers []string) *StreamConstructor

func (s StreamConstructor) NewConsumer(ctx context.Context, topic string, name string, opts ...workflow.ConsumerOption) (workflow.Consumer, error)

func (s StreamConstructor) NewProducer(ctx context.Context, topic string) (workflow.Producer, error)

type Translator func(m kafka.Message) workflow.ConnectorEvent