dune/internal/storage/memory/next.go

/*
------------------------------------------------------------------------------------------------------------------------
####### dune ####### Copyright (c) 2021 losyme ##################################################### MIT License #######
------------------------------------------------------------------------------------------------------------------------
*/

package memory

import (
	"sort"
	"time"

	"forge.chapril.org/dune/jw"
)

type jobsToRun struct { // AFINIR: sync.Pool ?
	now       time.Time
	namespace string
	jobs      []*jw.Job
}

func (jtr *jobsToRun) Next() (*jw.Job, error) {
	for i, job := range jtr.jobs {
		if job.Namespace == jtr.namespace &&
			(job.Status == jw.StatusTodo || job.Status == jw.StatusPending) && job.RunAfter.Before(jtr.now) {
			jtr.jobs = jtr.jobs[i+1:]
			return job, nil
		}
	}

	return nil, nil
}

func (m *memory) jobsRunning() jw.JobsRunning {
	jr := make(jw.JobsRunning)

	for _, job := range m.jobs {
		if job.Status != jw.StatusRunning {
			continue
		}

		jr[job.Namespace+job.Type] += 1
	}

	return jr
}

func (m *memory) NextJob(namespace string, fn jw.SelectNextJob) (*jw.Job, error) {
	m.sm.Lock()
	defer m.sm.Unlock()

	jobs := make([]*jw.Job, len(m.jobs))
	copy(jobs, m.jobs)

	sort.Slice(
		jobs,
		func(i, j int) bool {
			ji := jobs[i]
			jj := jobs[j]

			return ji.Priority > jj.Priority || ji.Session < jj.Session || ji.TimeReference.Before(jj.TimeReference)
		},
	)

	jtr := &jobsToRun{
		now:       time.Now(),
		namespace: namespace,
		jobs:      jobs,
	}

	job, err := fn(m.jobsRunning(), jtr)
	if job == nil || err != nil {
		return nil, err
	}

	// Sufffisant ?
	clone := new(jw.Job)
	*clone = *job

	job.Status = jw.StatusRunning

	return clone, nil
}

/*
######################################################################################################## @(°_°)@ #######
*/