core

package

v0.6.0 Latest Latest Go to latest Published: Apr 8, 2024 License: Apache-2.0 Imports: 15 Imported by: 1

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/TTraveller7/invokerlib

Links

Open Source Insights

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

Constants ¶

This section is empty.

Variables ¶

View Source

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 ¶ added in v0.5.7

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 ¶ added in v0.5.1

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

func NewCron ¶ added in v0.5.1

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

type JoinWorker ¶ added in v0.5.1

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

func NewJoinWorker ¶ added in v0.5.1

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

func (*JoinWorker) JoinWorkerProcessCallback ¶ added in v0.5.1

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

type Producer ¶

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

type Watermark ¶ added in v0.5.1

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

func NewWatermark ¶ added in v0.5.1

func NewWatermark(windowSize int64) *Watermark

func (*Watermark) Advance ¶ added in v0.5.1

func (w *Watermark) Advance()

func (*Watermark) Get ¶ added in v0.5.1

func (w *Watermark) Get() int64

type WorkerMeta ¶ added in v0.5.1

type WorkerMeta struct {
	WorkerId   string
	TopicIndex int
	Alive      bool
}

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL