kafkadriver

Documentation

Overview

Package kafkadriver contains Apache Kafka message bus driver for unifrost.StreamHandler

Index

• type Client
  • func NewClient(ctx context.Context, brokers []string, groupID string, config *sarama.Config, ...) (*Client, error)
  • func (client *Client) Close(ctx context.Context) error
  • func (client *Client) Subscribe(ctx context.Context, topic string) (*pubsub.Subscription, error)
• type Option

Types

type Client

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

Client handles the subscriptions to Kafka broker topics

func NewClient

func NewClient(
	ctx context.Context,
	brokers []string,
	groupID string,
	config *sarama.Config,
	opts ...Option,
) (*Client, error)

NewClient returns *kafkadriver.Client, also connects to the kafka cluster.

This driver uses https://github.com/Shopify/sarama internally to connect to kafka brokers. A slice of broker addresses are required by sarama to connect to the kafka cluster, specify only one if single node. Configuration is handled bysarama.Config to subscribe to a kafka topic. https://godoc.org/github.com/Shopify/sarama#Config groupID is consumer group id. Additional configuration options can be added with kafkadriver.Option functions.

func (*Client) Close

func (client *Client) Close(ctx context.Context) error

Close is just a placeholder

func (*Client) Subscribe

func (client *Client) Subscribe(ctx context.Context, topic string) (*pubsub.Subscription, error)

Subscribe method subscribes to the kafka topic

type Option

type Option func(*Client) error

Option is a self-refrential function for configuration parameters