kafka

Documentation

Index

• type Channel
  • func (channel *Channel) Consume(handler goChan.Handler) chan error
  • func (channel *Channel) Produce(ctx context.Context, proto goChan.MessageInterface) error
  • func (channel *Channel) SetReaderMiddleWares(mw ...goChan.Middleware)
  • func (channel *Channel) SetWriterMiddleWares(mw ...goChan.Middleware)
• type ChannelConfig
• type Manager
  • func NewManager(host string) Manager
  • func (manager Manager) CreateChannel(name string, config goChan.ConfigInterface) (goChan.ChannelInterface, error)
  • func (manager Manager) Ready() error
• type ReaderInterface
• type WriterInterface

Types

type Channel

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

func (*Channel) Consume

func (channel *Channel) Consume(handler goChan.Handler) chan error

func (*Channel) Produce

func (channel *Channel) Produce(ctx context.Context, proto goChan.MessageInterface) error

func (*Channel) SetReaderMiddleWares

func (channel *Channel) SetReaderMiddleWares(mw ...goChan.Middleware)

func (*Channel) SetWriterMiddleWares

func (channel *Channel) SetWriterMiddleWares(mw ...goChan.Middleware)

type ChannelConfig

type ChannelConfig struct {
	TopicGroup        string
	NumPartitions     int
	ReplicationFactor int
}

type Manager

type Manager struct {
	Host        string
	ConnFactory func(host string) (*kafka.Conn, error)
}

func NewManager

func NewManager(host string) Manager

func (Manager) CreateChannel

func (manager Manager) CreateChannel(name string, config goChan.ConfigInterface) (goChan.ChannelInterface, error)

func (Manager) Ready

func (manager Manager) Ready() error

type ReaderInterface

type ReaderInterface interface {
	FetchMessage(ctx context.Context) (kafka.Message, error)
	CommitMessages(ctx context.Context, messages ...kafka.Message) error
}

type WriterInterface

type WriterInterface interface {
	WriteMessages(ctx context.Context, messages ...kafka.Message) error
}