sinks

Documentation

Index

• Constants
• type Kafka
  • func (k *Kafka) Start(_ ...interface{}) error
• type KafkaGSSAPI
• type KafkaMessage
• type Pulsar
  • 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
  • func New(name string, sinkType string, conf *viper.Viper, stop chan error) (SinkI, error)
• type Stdout
  • func (s *Stdout) Start(i ...interface{}) (err error)

Constants

const (
	// sink Possible Statuses
	SinkStatusOnError = "ON_ERROR"
	SinkStatusRunning = "RUNNING"
	SinkStatusWaiting = "WAITING"
)

Possible Statuses

const KafkaType = "Kafka"

KafkaType type of sink

const MaxRetry = 20

MaxRetry retry max

const PulsarType = "Pulsar"

PulsarType type of sink

const StdoutType = "Stdout"

StdoutType type of sink

Types

type Kafka

type Kafka struct {
	*Sink
	// contains filtered or unexported fields
}

Kafka representation of kafka sink

func (*Kafka) Start

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

Start kafka sink

type KafkaGSSAPI

type KafkaGSSAPI struct {
	ServiceName        string `json:"service_name" mapstructure:"service_name"`
	Realm              string `json:"realm"`
	KerberosConfigPath string `json:"kerberos_config_path" mapstructure:"kerberos_config_path"`
	KeyTabPath         string `json:"key_tab_path" mapstructure:"key_tab_path"`
}

type KafkaMessage

type KafkaMessage struct {
	// The Kafka topic for this message.
	Topic string
	// The partitioning key for this message.
	Key string
	// The source offset of message
	Offset *events.Offset
	// The actual serialized message to store in Kafka.
	Value []byte
}

kafkaSinkConfig representation of Kafka Message

type Pulsar

type Pulsar struct {
	*Sink
	// contains filtered or unexported fields
}

Pulsar representation of Pulsar sink

func (*Pulsar) GetInputChan

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

GetInputChan return the input channel attached to this sink

func (*Pulsar) Start

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

Start connect to pulsar and start producer

type Sink

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

Sink representation of sink

func (*Sink) GetCommitChan

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

GetCommitChan return the commit channel attached to this sink

func (*Sink) GetInputChan

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

GetInputChan return input channel attach to sink

func (*Sink) SendCommit

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

SendCommit send a commit message into the commit channel of this sink

type SinkI

type SinkI interface {
	Start(...interface{}) error
	GetInputChan() chan events.LookatchEvent
	GetCommitChan() chan interface{}
}

SinkI sink interface

func New

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

New create new sink

type Stdout

type Stdout struct {
	*Sink
}

Stdout representation of sink

func (*Stdout) Start

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

Start stdout sink