kafka

Documentation

Index

• Constants
• type Decoder
• type Publisher
  • func NewPublisher(option PublisherOption) *Publisher
  • func (publisher *Publisher) Publish(ctx context.Context, messages []message.OutputMessage) error
• type PublisherOption
• type SubscribeOption
  • func (option SubscribeOption) AddAddr(addr ...string) SubscribeOption
  • func (option SubscribeOption) SetGroupID(groupID string) SubscribeOption
  • func (option SubscribeOption) SetHealthReporter(reporter *health.StatusReporter) SubscribeOption
  • func (option SubscribeOption) SetLimiter(limiter rate.Limit) SubscribeOption
  • func (option SubscribeOption) SetRecords(records record.Factory) SubscribeOption
  • func (option SubscribeOption) SetTopic(topic string) SubscribeOption
• type Subscriber
  • func NewKafkaSubscriber(option SubscribeOption) (*Subscriber, error)
  • func (s *Subscriber) Close() error
• type WriteConfig

Constants

const (
	PartVarPartitionIDField = "kafkaPartId"
)

Types

type Decoder

type Decoder interface {
	Decode(message kafka.Message) (message.Message, error)
}

type Publisher

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

func NewPublisher

func NewPublisher(option PublisherOption) *Publisher

func (*Publisher) Publish

func (publisher *Publisher) Publish(ctx context.Context, messages []message.OutputMessage) error

type PublisherOption

type PublisherOption struct {
	WriteConfig WriteConfig
}

type SubscribeOption

type SubscribeOption struct {
	Addr        []string
	Topic       *string
	GroupID     *string
	PartitionID int64
	Records     record.Factory
	Reporter    *health.StatusReporter
	Limiter     *rate.Limiter
}

func (SubscribeOption) AddAddr

func (option SubscribeOption) AddAddr(addr ...string) SubscribeOption

func (SubscribeOption) SetGroupID

func (option SubscribeOption) SetGroupID(groupID string) SubscribeOption

func (SubscribeOption) SetHealthReporter

func (option SubscribeOption) SetHealthReporter(reporter *health.StatusReporter) SubscribeOption

func (SubscribeOption) SetLimiter

func (option SubscribeOption) SetLimiter(limiter rate.Limit) SubscribeOption

func (SubscribeOption) SetRecords

func (option SubscribeOption) SetRecords(records record.Factory) SubscribeOption

func (SubscribeOption) SetTopic

func (option SubscribeOption) SetTopic(topic string) SubscribeOption

type Subscriber

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

func NewKafkaSubscriber

func NewKafkaSubscriber(option SubscribeOption) (*Subscriber, error)

func (*Subscriber) Close

func (s *Subscriber) Close() error

type WriteConfig

type WriteConfig = kafka.WriterConfig