kafka

Documentation

Index

• Constants
• type Connection
  • func NewConn(conf ConnectionConfig) (c *Connection, err error)
  • func (c *Connection) Close() (err error)
  • func (c *Connection) Connect() (err error)
  • func (c *Connection) CreateTopics(tcList ...kafka.TopicConfig) (err error)
  • func (c *Connection) DeleteTopics(topicList ...string) (err error)
  • func (c *Connection) NewReader(rc kafka.ReaderConfig) (r *kafka.Reader)
  • func (c *Connection) NewWriter(topic string) (w *kafka.Writer)
  • func (c *Connection) Reconnect() (err error)
  • func (c *Connection) SetDialer(dialer *kafka.Dialer)
  • func (c *Connection) SetTransport(transport *kafka.Transport)
• type ConnectionConfig

Constants

const (
	// Error constants
	ECode080001 = e.Code0800 + "01"
	ECode080002 = e.Code0800 + "02"
	ECode080003 = e.Code0800 + "03"
	ECode080004 = e.Code0800 + "04"
	ECode080005 = e.Code0800 + "05"
	ECode080006 = e.Code0800 + "06"
	ECode080007 = e.Code0800 + "07"
	ECode080008 = e.Code0800 + "08"
	ECode080009 = e.Code0800 + "09"
	ECode08000A = e.Code0800 + "0A"
	ECode08000B = e.Code0800 + "0B"
	ECode08000C = e.Code0800 + "0C"
	ECode08000D = e.Code0800 + "0D"
	ECode08000E = e.Code0800 + "0E"
	ECode08000F = e.Code0800 + "0F"
)

Types

type Connection

type Connection struct {
	Context context.Context
	// contains filtered or unexported fields
}

Connection a kafka connection with pre-initialized address list, dialer, transport and SASL mechanism

func NewConn

func NewConn(conf ConnectionConfig) (c *Connection, err error)

NewConn create a new Kafka connection

func (*Connection) Close

func (c *Connection) Close() (err error)

Close closes the connection

func (*Connection) Connect

func (c *Connection) Connect() (err error)

Connect opens a connection

func (*Connection) CreateTopics

func (c *Connection) CreateTopics(tcList ...kafka.TopicConfig) (err error)

CreateTopics creates topics using the associated dialer

func (*Connection) DeleteTopics

func (c *Connection) DeleteTopics(topicList ...string) (err error)

DeleteTopics creates topics using the associated dialer

func (*Connection) NewReader

func (c *Connection) NewReader(rc kafka.ReaderConfig) (r *kafka.Reader)

NewReader helper to return a new kafka reader using this connection's address list and dialer. If brokers or a dialer is set in the config, then those will be used instead

func (*Connection) NewWriter

func (c *Connection) NewWriter(topic string) (w *kafka.Writer)

NewWriter helper to return a new kafka writer using this connection's address list and transport

func (*Connection) Reconnect

func (c *Connection) Reconnect() (err error)

Reconnect closes and reopens a connection

func (*Connection) SetDialer

func (c *Connection) SetDialer(dialer *kafka.Dialer)

SetDialer sets the connection's dialer

func (*Connection) SetTransport

func (c *Connection) SetTransport(transport *kafka.Transport)

SetTransport sets the connection's transport

type ConnectionConfig

type ConnectionConfig struct {
	AddressList   []string
	Context       context.Context
	NoTLS         bool
	SASLMechanism sasl.Mechanism
	Timeout       *time.Duration
	TLS           *tls.Config
}

ConnectionConfig for NewConn