add tests for multiple queues, cron

backends/sqlite/sqlite_backend_test.go

@@ -18,12 +18,16 @@ import (
 	"github.com/pranavmodx/neoq-sqlite"
 	"github.com/pranavmodx/neoq-sqlite/backends/sqlite"
 	"github.com/pranavmodx/neoq-sqlite/handler"
+	"github.com/pranavmodx/neoq-sqlite/internal"
 	"github.com/pranavmodx/neoq-sqlite/jobs"
+	"github.com/pranavmodx/neoq-sqlite/logging"
 )
 
 //go:embed migrations/*.sql
 var sqliteMigrationsFS embed.FS
 
+var errPeriodicTimeout = errors.New("timed out waiting for periodic job")
+
 func prepareAndCleanupDB(t *testing.T) (dbURL string, db *sql.DB) {
 	t.Helper()
 
@@ -138,3 +142,128 @@ func TestBasicJobProcessing(t *testing.T) {
 		t.Error(fmt.Errorf("job queue does not match its expected value: %v != %v", jqueue, queue))
 	}
 }
+
+// TestBasicJobMultipleQueue tests that the postgres backend is able to process jobs on multiple queues
+func TestBasicJobMultipleQueue(t *testing.T) {
+	const queue = "testing"
+	const queue2 = "testing2"
+	done := make(chan bool)
+	doneCnt := 0
+
+	timeoutTimer := time.After(5 * time.Second)
+
+	connString, _ := prepareAndCleanupDB(t)
+
+	ctx := context.TODO()
+	nq, err := neoq.New(ctx,
+		neoq.WithBackend(sqlite.Backend),
+		sqlite.WithConnectionString(connString),
+		neoq.WithLogLevel(logging.LogLevelDebug),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nq.Shutdown(ctx)
+
+	h := handler.New(queue, func(_ context.Context) (err error) {
+		done <- true
+		return
+	})
+
+	h2 := handler.New(queue2, func(_ context.Context) (err error) {
+		done <- true
+		return
+	})
+
+	err = nq.Start(ctx, h)
+	if err != nil {
+		t.Error(err)
+	}
+
+	err = nq.Start(ctx, h2)
+	if err != nil {
+		t.Error(err)
+	}
+
+	go func() {
+		jid, e := nq.Enqueue(ctx, &jobs.Job{
+			Queue: queue,
+			Payload: map[string]interface{}{
+				"message": fmt.Sprintf("hello world: %d", internal.RandInt(10000000000)),
+			},
+		})
+		if e != nil || jid == jobs.DuplicateJobID {
+			t.Error(e)
+		}
+
+		jid2, e := nq.Enqueue(ctx, &jobs.Job{
+			Queue: queue2,
+			Payload: map[string]interface{}{
+				"message": fmt.Sprintf("hello world: %d", internal.RandInt(10000000000)),
+			},
+		})
+		if e != nil || jid2 == jobs.DuplicateJobID {
+			t.Error(e)
+		}
+	}()
+
+results_loop:
+	for {
+		select {
+		case <-timeoutTimer:
+			err = jobs.ErrJobTimeout
+			break results_loop
+		case <-done:
+			doneCnt++
+			if doneCnt == 2 {
+				break results_loop
+			}
+		}
+	}
+
+	if err != nil {
+		t.Error(err)
+	}
+}
+
+func TestCron(t *testing.T) {
+	done := make(chan bool, 1)
+	defer close(done)
+	const cron = "* * * * * *"
+	connString, _ := prepareAndCleanupDB(t)
+
+	ctx := context.TODO()
+	nq, err := neoq.New(ctx, neoq.WithBackend(sqlite.Backend), sqlite.WithConnectionString(connString))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nq.Shutdown(ctx)
+
+	h := handler.NewPeriodic(func(ctx context.Context) (err error) {
+		done <- true
+		return
+	})
+
+	h.WithOptions(
+		handler.JobTimeout(500*time.Millisecond),
+		handler.Concurrency(1),
+	)
+
+	err = nq.StartCron(ctx, cron, h)
+	if err != nil {
+		t.Error(err)
+	}
+
+	// allow time for listener to start
+	time.Sleep(5 * time.Millisecond)
+
+	select {
+	case <-time.After(1 * time.Second):
+		err = errPeriodicTimeout
+	case <-done:
+	}
+
+	if err != nil {
+		t.Error(err)
+	}
+}