kafkago

Documentation

Index

• Constants
• func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) message.Publisher
• func NewSubscriber(config SubscriberConfig, logger watermill.LoggerAdapter) (message.Subscriber, error)
• func TCP4(address ...string) net.Addr
• type DefaultMarshaler
  • func (d DefaultMarshaler) Marshal(topic string, msg *message.Message) (*kafka.Message, error)
  • func (DefaultMarshaler) Unmarshal(kmsg *kafka.Message) (*message.Message, error)
• type Marshaler
• type MarshalerUnmarshaler
• type Publisher
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(topic string, msgs ...*message.Message) error
• type PublisherConfig
  • func (c *PublisherConfig) Validate() error
• type Subscriber
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)
• type SubscriberConfig
  • func (c SubscriberConfig) Validate() error
• type Unmarshaler

Constants

const HeaderKey = "_key"

const NoSleep time.Duration = -1

const UUIDHeaderKey = "_watermill_message_uuid"

Functions

func NewPublisher

func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) message.Publisher

func NewSubscriber

func NewSubscriber(config SubscriberConfig, logger watermill.LoggerAdapter) (message.Subscriber, error)

func TCP4

func TCP4(address ...string) net.Addr

Types

type DefaultMarshaler

type DefaultMarshaler struct {
}

func (DefaultMarshaler) Marshal

func (d DefaultMarshaler) Marshal(topic string, msg *message.Message) (*kafka.Message, error)

func (DefaultMarshaler) Unmarshal

func (DefaultMarshaler) Unmarshal(kmsg *kafka.Message) (*message.Message, error)

type Marshaler

type Marshaler interface {
	Marshal(topic string, msg *message.Message) (*kafka.Message, error)
}

type MarshalerUnmarshaler

type MarshalerUnmarshaler interface {
	Marshaler
	Unmarshaler
}

type Publisher

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

func (*Publisher) Close

func (p *Publisher) Close() error

func (*Publisher) Publish

func (p *Publisher) Publish(topic string, msgs ...*message.Message) error

type PublisherConfig

type PublisherConfig struct {
	Writer *kafka.Writer

	Brokers []string

	Async bool

	// Marshaler is used to marshal messages from Watermill format into Kafka format.
	Marshaler Marshaler

	// If true then each sent message will be wrapped with Opentelemetry tracing, provided by otelsarama.
	OTELEnabled bool
	Ipv4Only    bool
	Timeout     time.Duration
}

func (*PublisherConfig) Validate

func (c *PublisherConfig) Validate() error

type Subscriber

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

func (*Subscriber) Close

func (s *Subscriber) Close() error

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

type SubscriberConfig

type SubscriberConfig struct {
	Brokers []string

	Unmarshaler Unmarshaler

	OverrideReaderConfig kafka.ReaderConfig

	ConsumerGroup string

	NackResendSleep time.Duration

	ReconnectRetrySleep time.Duration

	OTELEnabled bool
}

func (SubscriberConfig) Validate

func (c SubscriberConfig) Validate() error

type Unmarshaler

type Unmarshaler interface {
	Unmarshal(*kafka.Message) (*message.Message, error)
}