kafka

Documentation

Index

• func NewPublisher(connection *Connection) pubsub.Publisher
• func NewSubscriber(connection *Connection, opts ...SubscriberOption) pubsub.Subscriber
• type Connection
  • func NewKafkaConnection(cfg *pubsub.Config, logger log.Logger) *Connection
  • func (c *Connection) Close() error
  • func (c *Connection) Ping(ctx context.Context) error
• type MigrationInfo
  • func (i MigrationInfo) MigrationID() string
• type Migrator
  • func NewMigrator(connection *Connection) *Migrator
  • func (m *Migrator) IsMigrated(_ context.Context, _ []pubsub.MigrationInfo) (bool, error)
  • func (m *Migrator) Migrate(ctx context.Context, migrations []pubsub.MigrationInfo) error
• type Publisher
  • func (p *Publisher) Publish(ctx context.Context, subject string, msgs ...*pubsub.Message) error
• type SubscriberOption
  • func WithMeterSubscriberOption(meter metric.Meter) SubscriberOption
  • func WithTracerSubscriberOption(tracer tracing.Tracer) SubscriberOption

Functions

func NewPublisher

func NewPublisher(connection *Connection) pubsub.Publisher

func NewSubscriber

func NewSubscriber(connection *Connection, opts ...SubscriberOption) pubsub.Subscriber

Types

type Connection

type Connection struct {
	Logger log.Logger
	// contains filtered or unexported fields
}

func NewKafkaConnection

func NewKafkaConnection(cfg *pubsub.Config, logger log.Logger) *Connection

func (*Connection) Close

func (c *Connection) Close() error

func (*Connection) Ping

func (c *Connection) Ping(ctx context.Context) error

type MigrationInfo

type MigrationInfo struct {
	ID string
	kafka.TopicConfig
}

func (MigrationInfo) MigrationID

func (i MigrationInfo) MigrationID() string

type Migrator

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

func NewMigrator

func NewMigrator(connection *Connection) *Migrator

func (*Migrator) IsMigrated

func (m *Migrator) IsMigrated(_ context.Context, _ []pubsub.MigrationInfo) (bool, error)

func (*Migrator) Migrate

func (m *Migrator) Migrate(ctx context.Context, migrations []pubsub.MigrationInfo) error

type Publisher

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

func (*Publisher) Publish

func (p *Publisher) Publish(ctx context.Context, subject string, msgs ...*pubsub.Message) error

type SubscriberOption

type SubscriberOption func(*subscriber)

func WithMeterSubscriberOption

func WithMeterSubscriberOption(meter metric.Meter) SubscriberOption

func WithTracerSubscriberOption

func WithTracerSubscriberOption(tracer tracing.Tracer) SubscriberOption