kafkasarama

Documentation

Index

• func NewSASLPlainPublisherConfig(username, password string) *sarama.Config
• func NewSASLPlainSubscriberConfig(username, password string) *sarama.Config
• func NewSASLPublisherConfig(username, password string) *sarama.Config
• func NewSASLSubscriberConfig(username, password string) *sarama.Config
• type Publisher
  • func NewPublisher(slogHandler slog.Handler, saramaConfig *sarama.Config, brokers []string) (*Publisher, error)
  • func (p Publisher) Close() error
  • func (p Publisher) Publish(event pubsub.Event[string, []byte], channels ...string) error
• type Subscriber
  • func NewSubscriber(slogHandler slog.Handler, saramaConfig *sarama.Config, brokers []string, ...) (*Subscriber, error)
  • func (s Subscriber) Subscribe(channels ...string) (pubsub.Subscription[string, []byte], error)
• type Subscription
  • func (s Subscription) C() <-chan pubsub.Event[string, []byte]
  • func (s Subscription) Close() error

Functions

func NewSASLPlainPublisherConfig

func NewSASLPlainPublisherConfig(username, password string) *sarama.Config

NewSASLPlainPublisherConfig creates a new kafka publisher config with Plain SASL authentication.

func NewSASLPlainSubscriberConfig

func NewSASLPlainSubscriberConfig(username, password string) *sarama.Config

NewSASLPlainSubscriberConfig creates a new kafka subscriber config with Plain SASL authentication.

func NewSASLPublisherConfig

func NewSASLPublisherConfig(username, password string) *sarama.Config

NewSASLPublisherConfig creates a new kafka publisher config with SASL authentication.

func NewSASLSubscriberConfig

func NewSASLSubscriberConfig(username, password string) *sarama.Config

NewSASLSubscriberConfig creates a new kafka subscriber config with SASL authentication.

Types

type Publisher

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

Publisher represents a kafka publisher.

func NewPublisher

func NewPublisher(
	slogHandler slog.Handler,
	saramaConfig *sarama.Config,
	brokers []string,
) (*Publisher, error)

NewPublisher creates a new kafka publisher.

func (Publisher) Close

func (p Publisher) Close() error

Close closes the kafka publisher.

func (Publisher) Publish

func (p Publisher) Publish(event pubsub.Event[string, []byte], channels ...string) error

Publish publishes an event to a kafka topic.

type Subscriber

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

Subscriber represents a kafka subscriber.

func NewSubscriber

func NewSubscriber(
	slogHandler slog.Handler,
	saramaConfig *sarama.Config,
	brokers []string,
	consumerGroup string,
) (*Subscriber, error)

NewSubscriber creates a new kafka subscriber. nolint: revive // allow unused consumerGroup parameter for future proofing.

func (Subscriber) Subscribe

func (s Subscriber) Subscribe(channels ...string) (pubsub.Subscription[string, []byte], error)

Subscribe creates a new subscription that runs in the background.

type Subscription

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

Subscription represents a stream of events published to a kafka topic.

func (Subscription) C

func (s Subscription) C() <-chan pubsub.Event[string, []byte]

C returns a receive-only go channel of events published.

func (Subscription) Close

func (s Subscription) Close() error

Close closes the subscription.