ogiconsumer

Documentation

Index

• Variables
• func Consume()
• type Consumer
  • func NewConfluentKafka() Consumer
  • func NewConsumerPlugin() Consumer
• type ConsumerPlugin
  • func (plugin *ConsumerPlugin) Consume()
• type Kafka
  • func (k *Kafka) Close()
  • func (k *Kafka) Configure()
  • func (k *Kafka) Consume()
  • func (k *Kafka) EventHandler()
  • func (k *Kafka) NewConsumer()
  • func (k *Kafka) SubscribeTopics()
• type NewConsumerFunc

Variables

var (
	BootstrapServers             = golenv.OverrideIfEnv("CONSUMER_BOOTSTRAP_SERVERS", "")
	GroupId                      = golenv.OverrideIfEnv("CONSUMER_GROUP_ID", "")
	SessionTimeoutMs             = golenv.OverrideIfEnv("CONSUMER_SESSION_TIMEOUT_MS", "6000")
	GoEventsChannelEnable        = golenv.OverrideIfEnv("CONSUMER_GOEVENTS_CHANNEL_ENABLE", "true")
	GoEventsChannelSize          = golenv.OverrideIfEnv("CONSUMER_GOEVENTS_CHANNEL_SIZE", "1000")
	GoApplicationRebalanceEnable = golenv.OverrideIfEnv("CONSUMER_GO_APPLICATION_REBALANCE_ENABLE", "true")
	ConsumerType                 = golenv.OverrideIfEnv("CONSUMER_TYPE", "confluent-kafka")
)

var (
	ConsumerPluginPath = golenv.OverrideIfEnv("CONSUMER_PLUGIN_PATH", "./consumer.so")
)

var (
	KafkaTopics = golenv.OverrideIfEnv("CONSUMER_KAFKA_TOPICS", "")
)

Functions

func Consume

func Consume()

Types

type Consumer

type Consumer interface {
	Consume()
}

func NewConfluentKafka

func NewConfluentKafka() Consumer

func NewConsumerPlugin

func NewConsumerPlugin() Consumer

type ConsumerPlugin

type ConsumerPlugin struct {
	Name        string
	ConsumeFunc plugin.Symbol
}

func (*ConsumerPlugin) Consume

func (plugin *ConsumerPlugin) Consume()

type Kafka

type Kafka struct {
	ConfigMap kafka.ConfigMap
	Consumer  *kafka.Consumer
}

func (*Kafka) Close

func (k *Kafka) Close()

func (*Kafka) Configure

func (k *Kafka) Configure()

func (*Kafka) Consume

func (k *Kafka) Consume()

func (*Kafka) EventHandler

func (k *Kafka) EventHandler()

func (*Kafka) NewConsumer

func (k *Kafka) NewConsumer()

func (*Kafka) SubscribeTopics

func (k *Kafka) SubscribeTopics()

type NewConsumerFunc

type NewConsumerFunc func() Consumer