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) SetTopic(topic string) SubscribeOption
• type Subscriber
  • func NewKafkaSubscriber(option SubscribeOption) (*Subscriber, error)
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) Read(ctx context.Context) (message.InputMessage, 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
}

func (SubscribeOption) AddAddr

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

func (SubscribeOption) SetGroupID

func (option SubscribeOption) SetGroupID(groupID string) 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

func (*Subscriber) Read

func (s *Subscriber) Read(ctx context.Context) (message.InputMessage, error)

type WriteConfig

type WriteConfig = kafka.WriterConfig