jobmap

README

jobmap

Job-processor which allows pending jobs to be superceded by newer jobs based on an arbitrary key associated with each job.

func main() {
	js := jobmap.New()
	ctx, cancelFunc := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancelFunc()
	go pumpJobs(js, cancelFunc)
	js.Run(ctx)
}

func pumpJobs(js *jobmap.JobMap, cancelFunc context.CancelFunc) {
	defer cancelFunc()
	js.TakeJob("Apple", func(ctx context.Context) { fmt.Println("Apple has been bought."); time.Sleep(time.Second) })
	js.TakeJob("Cheese", func(ctx context.Context) { fmt.Println("Cheese has been bought."); time.Sleep(time.Second) })
	js.TakeJob("Apple", func(ctx context.Context) { fmt.Println("Apple has been peeled.") })
	js.TakeJob("Cheese", func(ctx context.Context) { fmt.Println("Cheese has been cut.") })
	js.TakeJob("Apple", func(ctx context.Context) { fmt.Println("Apple has been eaten.") })
	js.TakeJob("Cheese", func(ctx context.Context) { fmt.Println("Cheese has been eaten.") })
	time.Sleep(time.Second * 2)
}

Output:

Apple has been bought.
Cheese has been bought.
Apple has been eaten.
Cheese has been eaten.

Two jobs with information about Apple and Cheese having been eaten do arrive before the first job for either finishes, so the two jobs containing information about their intermediate states (peeled and cut) are effectively skipped.

Documentation

Index

• type JobFunc
• type JobMap
  • func New() *JobMap
  • func (jm *JobMap) Run(ctx context.Context)
  • func (jm *JobMap) TakeJob(key interface{}, job JobFunc)

Types

type JobFunc

type JobFunc func(context.Context)

JobFunc is a function which represents a job.

type JobMap

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

JobMap associates function invocations ("jobs") with keys (interface{}), where no two jobs with the same key may run at the same time, and only the latest known job for a given key may be pending to run.

func New

func New() *JobMap

New constructs a new JobMap and returns a pointer to it.

func (*JobMap) Run

func (jm *JobMap) Run(ctx context.Context)

Run starts jobs queued with TakeJob() on their own go-routines. If no job for a given key is running, the next job with that key will run immediately. If a job for a given key is running, a following job with that key will be pending to run after the running job finishes. If a job for a given key is pending to run, a following job with the same key will take its place. When ctx is cancelled, Run will wait for running jobs to finish and return.

func (*JobMap) TakeJob

func (jm *JobMap) TakeJob(key interface{}, job JobFunc)

TakeJob thread-safely queues a job to run on a new go-routine for given key.