client

Documentation

Index

• Variables
• type CfgOption
  • func WithAddress(address string) CfgOption
  • func WithConsumerGroup(group string) CfgOption
  • func WithConsumerID(consumer string) CfgOption
  • func WithHTTPClient(client *http.Client) CfgOption
  • func WithPartitions(partitions []int64) CfgOption
  • func WithTopic(topic string) CfgOption
• type Client
  • func New(cfg *Config, options ...CfgOption) *Client
  • func (c *Client) CommitRecord(ctx context.Context, record *kayakv1.Record) error
  • func (c *Client) CreateTopic(ctx context.Context, topicName string, partitions int64, ttl int64) error
  • func (c *Client) DeleteTopic(ctx context.Context, topic string) error
  • func (c *Client) FetchRecord(ctx context.Context) (*kayakv1.Record, error)
  • func (c *Client) GetRecords(ctx context.Context, topic string, start string, limit int64) ([]*kayakv1.Record, error)
  • func (c *Client) PutRecords(ctx context.Context, records ...*kayakv1.Record) error
  • func (c *Client) RegisterConsumer(ctx context.Context) error
  • func (c *Client) UpdateConfig(options ...CfgOption)
• type Config
  • func NewConfig(id string) *Config

Variables

var (
	ErrConsumerAlreadyRegistered = store.ErrConsumerAlreadyRegistered
	ErrTopicEmpty                = errors.New("topic cannot be empty")
	ErrCfgConsumerIDEmpty        = errors.New("consumer id must be set when a consumer group exists")
	ErrCfgConsumerGroupEmpty     = errors.New("consumer group must be set when a consumer id exists")
)

Types

type CfgOption

type CfgOption func(*Config)

func WithAddress

func WithAddress(address string) CfgOption

func WithConsumerGroup

func WithConsumerGroup(group string) CfgOption

func WithConsumerID

func WithConsumerID(consumer string) CfgOption

func WithHTTPClient

func WithHTTPClient(client *http.Client) CfgOption

func WithPartitions

func WithPartitions(partitions []int64) CfgOption

func WithTopic

func WithTopic(topic string) CfgOption

type Client

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

func New

func New(cfg *Config, options ...CfgOption) *Client

func (*Client) CommitRecord

func (c *Client) CommitRecord(ctx context.Context, record *kayakv1.Record) error

func (*Client) CreateTopic

func (c *Client) CreateTopic(ctx context.Context, topicName string, partitions int64, ttl int64) error

func (*Client) DeleteTopic

func (c *Client) DeleteTopic(ctx context.Context, topic string) error

func (*Client) FetchRecord

func (c *Client) FetchRecord(ctx context.Context) (*kayakv1.Record, error)

func (*Client) GetRecords

func (c *Client) GetRecords(ctx context.Context, topic string, start string, limit int64) ([]*kayakv1.Record, error)

func (*Client) PutRecords

func (c *Client) PutRecords(ctx context.Context, records ...*kayakv1.Record) error

func (*Client) RegisterConsumer

func (c *Client) RegisterConsumer(ctx context.Context) error

func (*Client) UpdateConfig

func (c *Client) UpdateConfig(options ...CfgOption)

type Config

type Config struct {
	Address       string
	ConsumerID    string
	ConsumerGroup string
	HTTPClient    *http.Client
	ID            string
	Topic         string
	Partitions    []int64
}

func NewConfig

func NewConfig(id string) *Config