core

Documentation

Index

• Variables
• func Exit()
• func Initialize(internalPc *conf.InternalProcessorConfig, pc *models.ProcessorCallbacks) error
• func MetricsClient() *utils.MetricsClient
• func NewConsumerGroupHandler(logs *log.Logger, setupFunc func() error, ...) sarama.ConsumerGroupHandler
• func PassToDefaultOutputTopic(ctx context.Context, record *models.Record) error
• func PassToOutputTopic(ctx context.Context, name string, record *models.Record) error
• func Run() error
• func Work(ctx context.Context, consumerConfig *conf.ConsumerConfig, workerIndex int, ...)
• type Cron
  • func NewCron(tickInterval time.Duration, windowSize int64, w *Watermark, done <-chan bool) *Cron
• type JoinWorker
  • func NewJoinWorker(w *Watermark, s state.StateStore, expierTime int) *JoinWorker
  • func (j *JoinWorker) JoinWorkerProcessCallback(ctx context.Context, record *models.Record) error
• type Producer
• type Watermark
  • func NewWatermark(windowSize int64) *Watermark
  • func (w *Watermark) Advance()
  • func (w *Watermark) Get() int64
• type WorkerMeta

Variables

var (
	ErrRedisConfNotInitialized = fmt.Errorf("redis config is not initialized")
	ErrConsumerNotify          = fmt.Errorf("consumer exits with notify")
)

Functions

func Exit

func Exit()

func Initialize

func Initialize(internalPc *conf.InternalProcessorConfig, pc *models.ProcessorCallbacks) error

func MetricsClient

func MetricsClient() *utils.MetricsClient

func NewConsumerGroupHandler

func NewConsumerGroupHandler(logs *log.Logger, setupFunc func() error, consumeFunc func(record *models.Record) error,
	workerNotifyChannel <-chan string, workerReadyChannel chan<- struct{}) sarama.ConsumerGroupHandler

func PassToDefaultOutputTopic

func PassToDefaultOutputTopic(ctx context.Context, record *models.Record) error

func PassToOutputTopic

func PassToOutputTopic(ctx context.Context, name string, record *models.Record) error

func Run

func Run() error

func Work

func Work(ctx context.Context, consumerConfig *conf.ConsumerConfig, workerIndex int, processFunc models.ProcessCallback,
	errCh chan<- error, wg *sync.WaitGroup, workerNotifyChannel <-chan string, workerReadyChannel chan<- struct{})

Types

type Cron

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

func NewCron

func NewCron(tickInterval time.Duration, windowSize int64, w *Watermark, done <-chan bool) *Cron

type JoinWorker

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

func NewJoinWorker

func NewJoinWorker(w *Watermark, s state.StateStore, expierTime int) *JoinWorker

func (*JoinWorker) JoinWorkerProcessCallback

func (j *JoinWorker) JoinWorkerProcessCallback(ctx context.Context, record *models.Record) error

type Producer

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

type Watermark

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

func NewWatermark

func NewWatermark(windowSize int64) *Watermark

func (*Watermark) Advance

func (w *Watermark) Advance()

func (*Watermark) Get

func (w *Watermark) Get() int64

type WorkerMeta

type WorkerMeta struct {
	WorkerId   string
	TopicIndex int
	Alive      bool
}