kafka

Documentation

Index

• Constants
• Variables
• func New() interface{ ... }
• type Client
• type Configuration
  • func (c *Configuration) Default()
  • func (c *Configuration) FromEnv(envs map[string]string)
  • func (c *Configuration) ToSarama() ([]string, *sarama.Config)
  • func (c *Configuration) Validate() error
• type HandleFx

Constants

const (
	EnvNameKafkaBrokers       string = "BLUDGEON_KAFKA_BROKERS"
	EnvNameKafkaClientId      string = "BLUDGEON_KAFKA_CLIENT_ID"
	EnvNameKafkaGroupId       string = "BLUDGEON_KAFKA_GROUP_ID"
	EnvNameKafkaConsumerGroup string = "BLUDGEON_KAFKA_CONSUMER_GROUP"
	EnvNameKafkaEnableLog     string = "BLUDGEON_KAFKA_ENABLE_LOG"
)

const (
	NoBrokersConfigured  string = "no brokers configured"
	NoClientIdConfigured string = "no client id configured"
	NoGroupIdConfigured  string = "no group id configured"
)

const (
	ConfigNil string = "config is nil"
)

Variables

var (
	ErrNoBrokersConfigured  = errors.New(NoBrokersConfigured)
	ErrNoClientIdConfigured = errors.New(NoClientIdConfigured)
	ErrNoGroupIdConfigured  = errors.New(NoGroupIdConfigured)
)

var ErrConfigNil = errors.New(ConfigNil)

Functions

func New

func New() interface {
	internal.Configurer
	internal.Initializer
	internal.Parameterizer
	Client
}

Types

type Client

type Client interface {
	Publish(topic string, item interface{}) (err error)
	Subscribe(topic string, handler HandleFx) (handlerId string, err error)
	Unsubscribe(topic string, handlerIds ...string)
	Topics(regEx *regexp.Regexp) ([]string, error)
}

type Configuration

type Configuration struct {
	Brokers       []string `json:"brokers"`
	ClientId      string   `json:"client_id"`
	GroupId       string   `json:"group_id"`
	EnableLog     bool     `json:"enable_log"`
	ConsumerGroup bool     `json:"consumer_group"`
}

func (*Configuration) Default

func (c *Configuration) Default()

func (*Configuration) FromEnv

func (c *Configuration) FromEnv(envs map[string]string)

func (*Configuration) ToSarama

func (c *Configuration) ToSarama() ([]string, *sarama.Config)

func (*Configuration) Validate

func (c *Configuration) Validate() error

type HandleFx

type HandleFx func(topic string, bytes []byte)