Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

sinks

package

Go to main page

Versions in this module

v0

v0.2.4

Mar 30, 2023

v0.2.2

May 20, 2020

Changes in this version

+ const DefaultChannelSize

+ var Factory = map[string]sinkCreator

+ func MsgByteSize(msg *sarama.ProducerMessage) int

+ func SendMsg(msgs []*sarama.ProducerMessage, producer sarama.SyncProducer) int64

type Kafka

+ KafkaConf *KafkaSinkConfig

+ func (k *Kafka) ProcessGenericEvent(genericMsg *events.GenericEvent) (*KafkaMessage, error)

+ func (k *Kafka) ProcessSQLEvent(sqlEvent *events.SQLEvent) (*KafkaMessage, error)

+ func (k *Kafka) ProducerLoop(producer sarama.SyncProducer, in chan *KafkaMessage)

+ func (k *Kafka) StartConsumer(kafkaChan chan *KafkaMessage)

+ func (k *Kafka) StartProducer(in chan *KafkaMessage, stop chan error)

+ type KafkaSinkConfig struct

+ Brokers []string

+ ClientID string

+ Consumer *KafkaUser

+ GSSAPI *KafkaGSSAPI

+ Kerberos bool

+ MaxMessageBytes int

+ NbProducer int

+ Producer *KafkaUser

+ ShuffleEvent bool

+ TLS bool

+ Topic string

+ TopicPrefix string

+ type KafkaUser struct

+ Password string

+ User string

type Pulsar

+ Producer pulsar.Producer

+ PulsarConf *PulsarSinkConfig

+ func (p *Pulsar) ProcessEvent(msg events.LookatchEvent) error

+ func (p *Pulsar) StartProducer()

+ type PulsarSinkConfig struct

+ Token string

+ Topic string

+ URL string

type Sink

+ Commit chan interface{}

+ Conf *viper.Viper

+ EncryptionKey string

+ In chan events.LookatchEvent

+ Name string

+ Stop chan error

type SinkI

+ func NewKafka(s *Sink) (SinkI, error)

+ func NewPulsar(s *Sink) (SinkI, error)

+ func NewStdout(s *Sink) (SinkI, error)

v0.2.0

Apr 7, 2020

Changes in this version

+ const PulsarType

+ const SinkStatusOnError

+ const SinkStatusRunning

+ const SinkStatusWaiting

+ type KafkaGSSAPI struct

+ KerberosConfigPath string

+ KeyTabPath string

+ Realm string

+ ServiceName string

+ type KafkaMessage struct

+ Key string

+ Offset *events.Offset

+ Topic string

+ Value []byte

+ type Pulsar struct

+ func (p *Pulsar) GetInputChan() chan events.LookatchEvent

+ func (p *Pulsar) Start(_ ...interface{}) error

type Sink

+ func (s *Sink) GetCommitChan() chan interface{}

+ func (s *Sink) GetInputChan() chan events.LookatchEvent

+ func (s *Sink) SendCommit(payload interface{})

type SinkI

+ GetCommitChan func() chan interface{}

v0.1.1

Mar 3, 2019

v0.1.0

Dec 16, 2018

Changes in this version

+ const MaxRetry

v0.0.1

Jun 28, 2018

Changes in this version

+ const KafkaType

+ const OvhKafkaType

+ const StdoutType

+ type Kafka struct

+ func (k *Kafka) GetInputChan() chan *events.LookatchEvent

+ func (k *Kafka) Start(_ ...interface{}) error

+ type Sink struct

+ type SinkI interface

+ GetInputChan func() chan *events.LookatchEvent

+ Start func(...interface{}) error

+ func New(name string, sinkType string, conf *viper.Viper, stop chan error, ...) (SinkI, error)

+ type Stdout struct

+ func (s *Stdout) GetInputChan() chan *events.LookatchEvent

+ func (s *Stdout) Start(i ...interface{}) (err error)