kafkago

package

v0.0.0-...-9adee62 Latest Latest Go to latest Published: Nov 25, 2024 License: Apache-2.0 Imports: 13 Imported by: 0

Details

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

Repository

github.com/ubntc/go

Links

Open Source Insights

Documentation ¶

Index ¶

Variables
func ConfigEntries(props config.KafkaProperties) []kafka.ConfigEntry
func DefaultTopicConfigs(props config.KafkaProperties, topics ...string) []kafka.TopicConfig
func KafkaError(err error) kafka.Error
func NewDialer(cfg *config.KeyFile) *kafka.Dialer
type Client
- func NewClient(cfg *config.KeyFile, props config.KafkaProperties, group config.Group) *Client
type Message
type Reader
type Writer
- func (w *Writer) Close() error
- func (w *Writer) Write(ctx context.Context, topic string, messages ...api.Message) error

Constants ¶

This section is empty.

Variables ¶

View Source

var (
	ErrOffsetRequired = errors.New("offset required")
	ErrInvalidOffset  = errors.New("invalid offset")
)

Functions ¶

func ConfigEntries ¶

func ConfigEntries(props config.KafkaProperties) []kafka.ConfigEntry

func DefaultTopicConfigs ¶

func DefaultTopicConfigs(props config.KafkaProperties, topics ...string) []kafka.TopicConfig

func KafkaError ¶

func KafkaError(err error) kafka.Error

func NewDialer ¶

func NewDialer(cfg *config.KeyFile) *kafka.Dialer

Types ¶

type Client ¶

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

func NewClient ¶

func NewClient(cfg *config.KeyFile, props config.KafkaProperties, group config.Group) *Client

func (*Client) Close ¶

func (c *Client) Close() error

func (*Client) CreateTopics ¶

func (c *Client) CreateTopics(ctx context.Context, topics ...string) (api.TopicErrors, error)

func (*Client) DeleteTopics ¶

func (c *Client) DeleteTopics(ctx context.Context, topics ...string) (api.TopicErrors, error)

func (*Client) GetLogger ¶

func (c *Client) GetLogger() api.LoggerFunc

func (*Client) IsExistsError ¶

func (c *Client) IsExistsError(err error) bool

func (*Client) NewReader ¶

func (c *Client) NewReader(topic string) api.Reader

func (*Client) NewWriter ¶

func (c *Client) NewWriter() api.Writer

func (*Client) Read ¶

func (c *Client) Read(ctx context.Context, topic string, partition int, offset *uint64) (api.Message, error)

func (*Client) SetLogger ¶

func (c *Client) SetLogger(fn api.LoggerFunc)

func (*Client) Write ¶

func (c *Client) Write(ctx context.Context, topic string, msg ...api.Message) error

Write writes a message using the default writer.

type Message ¶

type Message struct{ kafka.Message }

func (*Message) Key ¶

func (m *Message) Key() []byte

func (*Message) Offset ¶

func (m *Message) Offset() uint64

func (*Message) String ¶

func (m *Message) String() string

func (*Message) Topic ¶

func (m *Message) Topic() string

func (*Message) Value ¶

func (m *Message) Value() []byte

type Reader ¶

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

func (*Reader) Close ¶

func (r *Reader) Close() error

func (*Reader) Commit ¶

func (r *Reader) Commit(ctx context.Context, msg api.Message) error

func (*Reader) Read ¶

func (r *Reader) Read(ctx context.Context) (api.Message, error)

type Writer ¶

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

func (*Writer) Close ¶

func (w *Writer) Close() error

func (*Writer) Write ¶

func (w *Writer) Write(ctx context.Context, topic string, messages ...api.Message) error

Source Files ¶

View all Source files

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