kafka

package

v0.0.0-...-bf83fb3 Latest Latest Go to latest Published: Oct 14, 2023 License: MIT Imports: 10 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/Skyrin/go-lib

Documentation ¶

Index ¶

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

Constants ¶

View Source

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"
)

Variables ¶

This section is empty.

Functions ¶

This section is empty.

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

Source Files ¶

View all Source files

connection.go

Directories ¶

Path	Synopsis
aws
ec2
ec2/example

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL