kafka

Documentation

Index

• Constants
• type Connection
  • func New(options ...Option) (*Connection, error)
  • func (c *Connection) Close() error
  • func (c *Connection) IsLogName(_ context.Context, name string) (bool, error)
  • func (c *Connection) LogNames(_ context.Context) ([]string, error)
  • func (c *Connection) Stream(ctx context.Context, name string, start *monitor.Time, finish *monitor.Time) (itr iter.Seq[monitor.Entry], finished func() error)
• type Option
  • func AddBrokerAddr(addr ...string) Option
  • func Formatter(f logger.Formatter) Option
  • func Lifetime(d time.Duration) Option

Constants

const (
	DefaultBrokerAddr = "localhost:9092" // The default broker address to use if no broker addresses are specified.
	DefaultLifetime   = time.Minute
)

The default option values.

Types

type Connection

type Connection struct {
	log.BasicLogable
	// contains filtered or unexported fields
}

Connection represents a connection to Kafka.

func New

func New(options ...Option) (*Connection, error)

New connects to the Kafka server.

func (*Connection) Close

func (c *Connection) Close() error

Close closes the connection to Kafka.

func (*Connection) IsLogName

func (c *Connection) IsLogName(_ context.Context, name string) (bool, error)

IsLogName returns true iff the given string is a log name.

func (*Connection) LogNames

func (c *Connection) LogNames(_ context.Context) ([]string, error)

LogNames returns the log names.

func (*Connection) Stream

func (c *Connection) Stream(ctx context.Context, name string, start *monitor.Time, finish *monitor.Time) (itr iter.Seq[monitor.Entry], finished func() error)

Stream returns an iterator 'itr' down which entries from the log 'name' are passed. Only entries falling into the range determined by start and finish will be returned. If start is unassigned then the start of the log will be used; if end is unassigned then no end is used. The iterator 'itr' is bound to the given context. Iteration will be terminated on error. Any error will be reported by the returned 'finished' function.

It returns a single-use iterator.

type Option

type Option interface {
	// contains filtered or unexported methods
}

Option sets options on a Kafka logger.

func AddBrokerAddr

func AddBrokerAddr(addr ...string) Option

AddBrokerAddr adds the given addresses to the slice of Kafka broker address. Broker addresses should be in the format "hostname:port". DefaultBrokerAddr if and only if no broker addresses are added.

func Formatter

func Formatter(f logger.Formatter) Option

Formatter sets the formatter to use when converting strings to log messages.

func Lifetime

func Lifetime(d time.Duration) Option

Lifetime sets the lifetime of a Kafka client connection in the connection pool.