job_queue_test.go

package executor

import (
	_ "fmt"
	"sync"
	"testing"
	"time"
)

type TestJob struct {
	Created  int // unix second
	Id       string
	Key      string
	Duration int

	Executed int
	Expect   int // should executed in this time
}

func TestJobQueue(t *testing.T) {
	lock := &sync.Mutex{}
	const deltaT = 5 // for faster test time

	queue := New(func(key string, payloads []any) {
		for _, payload := range payloads {
			lock.Lock()
			// make sure all job only get executed once
			job := payload.(*TestJob)
			if job.Executed > 0 {
				t.Error("executed twice")
			}
			job.Executed = int(time.Now().UnixMilli())
			lock.Unlock()

			time.Sleep(time.Duration(job.Duration) * deltaT * time.Millisecond)
		}
	})

	//	 INCOMMING JOBS
	//	     0    1    2    3    4    5    6
	//	     +----+----+----+----+----+----+--
	//	 A1  [AAAAAAAAA]
	//   A2       [AAAAAAAAAAAAAA]
	//	 A3                 [AAAA]
	//   A4                           [AAAA]
	//   B1            [BBBB]
	//	 B2                 [BBBBBBBBBBBBBB]
	//	 B3                      [BBBB]
	//	 C1  [CCCCCCCCCCCCCC]                                ;
	//	 C2                      [CCCCCCCCCCCCCC]            ; non-overlap jobs
	//	 D1  [DDDD]                                          ;
	//	 D2       [DDDD]                                     ; 2nd job come right after 1st job done
	//
	//
	//   IDEA SCHEDULE
	//	     0    1    2    3    4    5    6    7
	//	     +----+----+----+----+----+----+----+
	//	 A1  [AAAAAAAAA]
	//   A2            [AAAAAAAAAAAAAA]
	//	 A3                           [AAAA]
	//   A4                                [AAAA]
	//   B1            [BBBB]
	//	 B2                 [BBBBBBBBBBBBBB]
	//	 B3                                [BBBB]
	//	 C1  [CCCCCCCCCCCCCC]
	//	 C2                      [CCCCCCCCCCCCCC]            ; non-overlap jobs
	//	 D1  [DDDD]                                          ;
	//	 D2       [DDDD]                                     ; 2nd job come right after 1st job done
	jobs := []*TestJob{
		{Id: "A1", Key: "A", Created: 0, Duration: 2, Expect: 0},
		{Id: "A2", Key: "A", Created: 1, Duration: 3, Expect: 2},
		{Id: "A3", Key: "A", Created: 3, Duration: 1, Expect: 5},
		{Id: "A4", Key: "A", Created: 5, Duration: 6, Expect: 6},
		{Id: "B1", Key: "B", Created: 2, Duration: 1, Expect: 2},
		{Id: "B2", Key: "B", Created: 3, Duration: 3, Expect: 3},
		{Id: "B3", Key: "B", Created: 4, Duration: 1, Expect: 6},
		{Id: "C1", Key: "C", Created: 0, Duration: 3, Expect: 0},
		{Id: "C2", Key: "C", Created: 4, Duration: 3, Expect: 4},
		{Id: "D1", Key: "D", Created: 0, Duration: 1, Expect: 0},
		{Id: "D2", Key: "D", Created: 1, Duration: 1, Expect: 1},
	}

	// find total time the test will run
	lastjob := jobs[0]
	for _, job := range jobs {
		if job.Expect+job.Duration > lastjob.Expect+lastjob.Duration {
			lastjob = job
		}
	}

	start := int(time.Now().UnixMilli())
	for ticker := range lastjob.Expect + lastjob.Duration {
		for _, job := range jobs {
			if job.Created == ticker {
				queue.Add(job.Key, job)
			}
		}
		time.Sleep(time.Duration(deltaT) * time.Millisecond)
	}

	for _, job := range jobs {
		expected := job.Expect * int(deltaT)
		if !isTheSameTime(job.Executed-start, expected) {
			t.Errorf("NOT EQ. JobId: %s. Want %d, Got %d", job.Id, expected, job.Executed-start)
		}
	}
}

func TestJobQueueClean(t *testing.T) {
	for range 100 { // make sure
		donec := make(chan bool, 100)
		processedchan := make(chan bool)

		goal := "goal"
		queue := New(func(key string, payloads []any) {
			for _, payload := range payloads {
				select {
				case processedchan <- true:
				default:
				}
				time.Sleep(1 * time.Millisecond)
				if payload.(string) == goal {
					donec <- true
				}
			}
		})
		queue.autoCleanDisabled = true

		queue.Add("key1", "1")
		queue.Add("key1", "2")
		select {
		case <-processedchan:
		default:
		}
		queue.clean()
		// while the job queue busy processing, we clean
		time.Sleep(5 * time.Millisecond)
		queue.Add("key1", goal)
		select {
		case <-donec:
		case <-time.After(5 * time.Second):
			t.Error("timeout")
			return
		}
	}
}

func ExampleJobQueue() {
	queue := New(func(key string, payloads []any) {
		// do your thing
	})
	queue.Add("key1", TestJob{Id: "job1"})
	queue.Add("key1", TestJob{Id: "job2"})
	queue.Add("key2", TestJob{Id: "job3"})
	queue.Add("key2", TestJob{Id: "job4"})

	// job 2 will wait for job 1
	// job 4 will wait for job 3
	// job 1 and job 3 executed in the same time
}