kafkastreamer

package module

v0.0.0-...-516621c Latest Latest Go to latest Published: Apr 20, 2026 License: BSD-3-Clause Imports: 7 Imported by: 1

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/luno/workflow

Links

Open Source Insights

Documentation ¶

Index ¶

func NewConnector(brokers []string, c *sarama.Config, t Translator, topic string) *connector
type Option
- func WithConfig(config *sarama.Config) Option
type Receiver
- func (r *Receiver) Close() error
- func (r *Receiver) Recv(ctx context.Context) (*workflow.Event, workflow.Ack, error)
type Sender
- func (p *Sender) Close() error
- func (p *Sender) Send(ctx context.Context, foreignID string, statusType int, ...) error
type StreamConstructor
- func New(brokers []string, opts ...Option) *StreamConstructor
- func (s StreamConstructor) NewReceiver(ctx context.Context, topic string, name string, ...) (workflow.EventReceiver, error)
- func (s StreamConstructor) NewSender(ctx context.Context, topic string) (workflow.EventSender, error)
type Translator

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func NewConnector ¶

func NewConnector(brokers []string, c *sarama.Config, t Translator, topic string) *connector

Types ¶

type Option ¶

type Option func(*StreamConstructor)

func WithConfig ¶

func WithConfig(config *sarama.Config) Option

type Receiver ¶

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

func (*Receiver) Close ¶

func (r *Receiver) Close() error

func (*Receiver) Recv ¶

func (r *Receiver) Recv(ctx context.Context) (*workflow.Event, workflow.Ack, error)

type Sender ¶

type Sender struct {
	Topic         string
	Writer        sarama.SyncProducer
	WriterTimeout time.Duration
}

func (*Sender) Close ¶

func (p *Sender) Close() error

func (*Sender) Send ¶

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

type StreamConstructor ¶

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

func New ¶

func New(brokers []string, opts ...Option) *StreamConstructor

func (StreamConstructor) NewReceiver ¶

func (s StreamConstructor) NewReceiver(
	ctx context.Context,
	topic string,
	name string,
	opts ...workflow.ReceiverOption,
) (workflow.EventReceiver, error)

func (StreamConstructor) NewSender ¶

func (s StreamConstructor) NewSender(ctx context.Context, topic string) (workflow.EventSender, error)

type Translator ¶

type Translator func(m *sarama.ConsumerMessage) *workflow.ConnectorEvent

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL