routines

Documentation

Index

• func NewJobComsumer(offsets map[int32]int64, jobs chan models.Job) error
• type JobProducer
  • func NewJobProducer(jobs <-chan []models.Job) (*JobProducer, error)
  • func (jp *JobProducer) Close()
  • func (jp *JobProducer) Indexes() map[int32]int64
• type Partition
• type TaskConsumer
  • func NewTaskConsumer() (*TaskConsumer, error)
  • func (tc *TaskConsumer) Close() (err error)
  • func (tc *TaskConsumer) Partitons() <-chan Partition
  • func (tc *TaskConsumer) WaitForDrain()
• type TaskScheduler
  • func NewTaskScheduler(partition int32, tasks <-chan models.Task) (*TaskScheduler, error)
  • func (ts *TaskScheduler) Halted()
  • func (ts *TaskScheduler) Jobs() <-chan []models.Job
  • func (ts *TaskScheduler) Start() error

Functions

func NewJobComsumer

func NewJobComsumer(offsets map[int32]int64, jobs chan models.Job) error

NewJobComsumer return a new job consumer

Types

type JobProducer

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

JobProducer handle the internal states of the producer.

func NewJobProducer

func NewJobProducer(jobs <-chan []models.Job) (*JobProducer, error)

NewJobProducer return a new job producer. Read jobs to send from jobs channel.

func (*JobProducer) Close

func (jp *JobProducer) Close()

Close the job producer

func (*JobProducer) Indexes

func (jp *JobProducer) Indexes() map[int32]int64

Indexes return the current write indexes by partition

type Partition

type Partition struct {
	Partition int32
	Tasks     chan models.Task
}

Partition handle a topic partition

type TaskConsumer

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

TaskConsumer handle the internal states of the consumer

func NewTaskConsumer

func NewTaskConsumer() (*TaskConsumer, error)

NewTaskConsumer return a new task consumer

func (*TaskConsumer) Close

func (tc *TaskConsumer) Close() (err error)

Close the task consumer

func (*TaskConsumer) Partitons

func (tc *TaskConsumer) Partitons() <-chan Partition

Partitons return the incomming partition channel

func (*TaskConsumer) WaitForDrain

func (tc *TaskConsumer) WaitForDrain()

WaitForDrain wait for consumer to EOF partitions

type TaskScheduler

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

TaskScheduler handle the internal states of the scheduler

func NewTaskScheduler

func NewTaskScheduler(partition int32, tasks <-chan models.Task) (*TaskScheduler, error)

NewTaskScheduler return a new task scheduler

func (*TaskScheduler) Halted

func (ts *TaskScheduler) Halted()

Halted wait for scheduler to be halt

func (*TaskScheduler) Jobs

func (ts *TaskScheduler) Jobs() <-chan []models.Job

Jobs return the out jobs channel

func (*TaskScheduler) Start

func (ts *TaskScheduler) Start() error

Start task scheduling