consumer

Documentation

Index

• type Consumer
  • func NewConsumer(config *config.KafkaConfiguration, overrideKafkaConsumer IConsume) (*Consumer, error)
  • func (c *Consumer) Close()
  • func (c *Consumer) Consume(topic string, handler JsonHandler, readyChan chan bool)
• type EventListener
  • func NewEventListener(stateKey string, portClient *cli.PortClient) (*EventListener, error)
  • func (l *EventListener) Run(resync func()) error
• type IConsume
• type IncomingMessage
• type JsonHandler

Types

type Consumer

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

func NewConsumer

func NewConsumer(config *config.KafkaConfiguration, overrideKafkaConsumer IConsume) (*Consumer, error)

func (*Consumer) Close

func (c *Consumer) Close()

func (*Consumer) Consume

func (c *Consumer) Consume(topic string, handler JsonHandler, readyChan chan bool)

type EventListener

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

func NewEventListener

func NewEventListener(stateKey string, portClient *cli.PortClient) (*EventListener, error)

func (*EventListener) Run

func (l *EventListener) Run(resync func()) error

type IConsume

type IConsume interface {
	SubscribeTopics(topics []string, rebalanceCb kafka.RebalanceCb) (err error)
	Poll(timeoutMs int) (event kafka.Event)
	Commit() (offsets []kafka.TopicPartition, err error)
	Close() (err error)
}

type IncomingMessage

type IncomingMessage struct {
	Diff *struct {
		After *struct {
			Identifier string `json:"installationId"`
		} `json:"after"`
	} `json:"diff"`
}

type JsonHandler

type JsonHandler func(value []byte)