service

Documentation

Index

• type Coordinator
  • func NewCoordinator(ctx context.Context, config config.Coordinator, repo Repository, ...) *Coordinator
  • func (c *Coordinator) CreateMapTask(ctx context.Context, inputFile string) (mapreduce.MapTask, error)
  • func (c *Coordinator) CreateReduceTask(ctx context.Context, inputFiles ...string) (mapreduce.ReduceTask, error)
  • func (c *Coordinator) FlushCreatedTasksToWorkers(ctx context.Context) error
  • func (c *Coordinator) GetMapTask(ctx context.Context) (mapreduce.MapTask, error)
  • func (c *Coordinator) GetReduceTask(ctx context.Context) (mapreduce.ReduceTask, error)
  • func (c *Coordinator) MapTasksRescheduler()
  • func (c *Coordinator) ReduceTasksRescheduler()
  • func (c *Coordinator) ReportMapTask(ctx context.Context, taskResult mapreduce.MapTaskResult) error
  • func (c *Coordinator) ReportReduceTask(ctx context.Context, taskResult mapreduce.ReduceTaskResult) error
  • func (c *Coordinator) TaskFlusher()
• type IDGenerator
• type Producer
• type Repository

Types

type Coordinator

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

func NewCoordinator

func NewCoordinator(
	ctx context.Context,
	config config.Coordinator,
	repo Repository,
	idGen IDGenerator,
	producer Producer,
	log logger.Logger,
) *Coordinator

func (*Coordinator) CreateMapTask

func (c *Coordinator) CreateMapTask(ctx context.Context, inputFile string) (mapreduce.MapTask, error)

func (*Coordinator) CreateReduceTask

func (c *Coordinator) CreateReduceTask(ctx context.Context, inputFiles ...string) (mapreduce.ReduceTask, error)

func (*Coordinator) FlushCreatedTasksToWorkers

func (c *Coordinator) FlushCreatedTasksToWorkers(ctx context.Context) error

func (*Coordinator) GetMapTask

func (c *Coordinator) GetMapTask(ctx context.Context) (mapreduce.MapTask, error)

func (*Coordinator) GetReduceTask

func (c *Coordinator) GetReduceTask(ctx context.Context) (mapreduce.ReduceTask, error)

func (*Coordinator) MapTasksRescheduler

func (c *Coordinator) MapTasksRescheduler()

func (*Coordinator) ReduceTasksRescheduler

func (c *Coordinator) ReduceTasksRescheduler()

func (*Coordinator) ReportMapTask

func (c *Coordinator) ReportMapTask(ctx context.Context, taskResult mapreduce.MapTaskResult) error

func (*Coordinator) ReportReduceTask

func (c *Coordinator) ReportReduceTask(ctx context.Context, taskResult mapreduce.ReduceTaskResult) error

func (*Coordinator) TaskFlusher

func (c *Coordinator) TaskFlusher()

type IDGenerator

type IDGenerator interface {
	GetID(ctx context.Context, entityType ids.EntityType) (string, error)
}

type Producer

type Producer interface {
	Produce(ctx context.Context, event events.Event) error
}

type Repository

type Repository interface {
	Transaction(ctx context.Context, transaction func(ctx context.Context) (any, error)) (any, error)

	CreateMapTask(ctx context.Context, task mapreduce.MapTask) error
	CreateReduceTask(ctx context.Context, task mapreduce.ReduceTask) error

	UpdateMapTask(ctx context.Context, task mapreduce.MapTask) error
	UpdateReduceTask(ctx context.Context, task mapreduce.ReduceTask) error

	UpdateMapTasks(ctx context.Context, ids []string, fields repository.UpdateFields) error
	UpdateReduceTasks(ctx context.Context, ids []string, fields repository.UpdateFields) error

	GetMapTask(ctx context.Context, id string) (mapreduce.MapTask, error)
	GetReduceTask(ctx context.Context, id string) (mapreduce.ReduceTask, error)

	QueryMapTasks(ctx context.Context, filter repository.Filter) ([]mapreduce.MapTask, error)
	QueryReduceTasks(ctx context.Context, filter repository.Filter) ([]mapreduce.ReduceTask, error)
}