refactor: queue pr part 1

contracts/queue/queue.go

@@ -1,12 +1,12 @@
 package queue
 
 type Queue interface {
-	// All get all jobs
-	All() []Job
 	// Chain creates a chain of jobs to be processed one by one, passing
 	Chain(jobs []Jobs) Task
 	// GetJob get job by signature
 	GetJob(signature string) (Job, error)
+	// GetJobs get all jobs
+	GetJobs() []Job
 	// Job add a job to queue
 	Job(job Job, args []any) Task
 	// Register register jobs

queue/application.go

@@ -1,7 +1,7 @@
 package queue
 
 import (
-	configcontract "github.com/goravel/framework/contracts/config"
+	contractsconfig "github.com/goravel/framework/contracts/config"
 	"github.com/goravel/framework/contracts/queue"
 )
 
@@ -10,24 +10,25 @@
 	job    queue.JobRepository
 }
 
-func NewApplication(config configcontract.Config) *Application {
+func NewApplication(config contractsconfig.Config) *Application {
 	return &Application{
 		config: NewConfig(config),
 		job:    NewJobImpl(),
 	}
 }
 
-func (app *Application) All() []queue.Job {
-	return app.job.All()
-}
 func (app *Application) Chain(jobs []queue.Jobs) queue.Task {
 	return NewChainTask(app.config, jobs)
 }
 
 func (app *Application) GetJob(signature string) (queue.Job, error) {
 	return app.job.Get(signature)
 }
 
+func (app *Application) GetJobs() []queue.Job {
+	return app.job.All()
+}
+
 func (app *Application) Job(job queue.Job, args []any) queue.Task {
 	return NewTask(app.config, job, args)
 }

queue/config.go

@@ -3,16 +3,16 @@
 import (
 	"fmt"
 
-	configcontract "github.com/goravel/framework/contracts/config"
+	contractsconfig "github.com/goravel/framework/contracts/config"
 	"github.com/goravel/framework/contracts/database/orm"
 	"github.com/goravel/framework/contracts/queue"
 )
 
 type Config struct {
-	config configcontract.Config
+	config contractsconfig.Config
 }
 
-func NewConfig(config configcontract.Config) queue.Config {
+func NewConfig(config contractsconfig.Config) queue.Config {
 	return &Config{
 		config: config,
 	}
@@ -28,8 +28,8 @@
 
 func (r *Config) Driver(connection string) string {
 	if connection == "" {
 		connection = r.DefaultConnection()
 	}
 
 	return r.config.GetString(fmt.Sprintf("queue.connections.%s.driver", connection))
 }
@@ -76,16 +76,16 @@
 
 func (r *Config) Size(connection string) int {
 	if connection == "" {
 		connection = r.DefaultConnection()
 	}
 
 	return r.config.GetInt(fmt.Sprintf("queue.connections.%s.size", connection), 100)
 }
 
 func (r *Config) Via(connection string) any {
 	if connection == "" {
 		connection = r.DefaultConnection()
 	}
 
 	return r.config.Get(fmt.Sprintf("queue.connections.%s.via", connection))
 }

queue/driver_async_test.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/goravel/framework/contracts/queue"
 	mocksconfig "github.com/goravel/framework/mocks/config"
-	mocksorm "github.com/goravel/framework/mocks/database/orm"
 	mocksqueue "github.com/goravel/framework/mocks/queue"
 )
 
@@ -29,31 +28,23 @@ type DriverAsyncTestSuite struct {
 }
 
 func TestDriverAsyncTestSuite(t *testing.T) {
-	mockConfig := mocksconfig.NewConfig(t)
-	mockQueue := mocksqueue.NewQueue(t)
-	app := NewApplication(mockConfig)
-
-	app.Register([]queue.Job{&TestAsyncJob{}, &TestDelayAsyncJob{}, &TestCustomAsyncJob{}, &TestErrorAsyncJob{}, &TestChainAsyncJob{}})
-	suite.Run(t, &DriverAsyncTestSuite{
-		app:        app,
-		mockConfig: mockConfig,
-		mockQueue:  mockQueue,
-	})
+	suite.Run(t, new(DriverAsyncTestSuite))
 }
 
 func (s *DriverAsyncTestSuite) SetupTest() {
 	testAsyncJob = 0
+	s.mockQueue = mocksqueue.NewQueue(s.T())
 	s.mockConfig = mocksconfig.NewConfig(s.T())
+	s.app = NewApplication(s.mockConfig)
+	s.app.Register([]queue.Job{&TestAsyncJob{}, &TestDelayAsyncJob{}, &TestCustomAsyncJob{}, &TestErrorAsyncJob{}, &TestChainAsyncJob{}})
 }
 
 func (s *DriverAsyncTestSuite) TestDefaultAsyncQueue() {
-	s.mockConfig.On("GetString", "queue.default").Return("async").Times(4)
-	s.mockConfig.On("GetString", "app.name").Return("goravel").Times(2)
-	s.mockConfig.On("GetString", "queue.connections.async.queue", "default").Return("default").Twice()
-	s.mockConfig.On("GetString", "queue.connections.async.driver").Return("async").Twice()
-	s.mockConfig.On("GetInt", "queue.connections.async.size", 100).Return(10).Twice()
-	s.mockConfig.On("GetString", "queue.failed.database").Return("database").Once()
-	s.mockConfig.On("GetString", "queue.failed.table").Return("failed_jobs").Once()
+	s.mockConfig.EXPECT().GetString("queue.default").Return("async").Times(3)
+	s.mockConfig.EXPECT().GetString("app.name").Return("goravel").Times(2)
+	s.mockConfig.EXPECT().GetString("queue.connections.async.queue", "default").Return("default").Twice()
+	s.mockConfig.EXPECT().GetString("queue.connections.async.driver").Return("async").Twice()
+	s.mockConfig.EXPECT().GetInt("queue.connections.async.size", 100).Return(10).Twice()
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
@@ -71,13 +62,11 @@ func (s *DriverAsyncTestSuite) TestDefaultAsyncQueue() {
 }
 
 func (s *DriverAsyncTestSuite) TestDelayAsyncQueue() {
-	s.mockConfig.On("GetString", "queue.default").Return("async").Times(4)
-	s.mockConfig.On("GetString", "app.name").Return("goravel").Times(3)
-	s.mockConfig.On("GetString", "queue.connections.async.queue", "default").Return("default").Once()
-	s.mockConfig.On("GetString", "queue.connections.async.driver").Return("async").Twice()
-	s.mockConfig.On("GetInt", "queue.connections.async.size", 100).Return(10).Twice()
-	s.mockConfig.On("GetString", "queue.failed.database").Return("database").Once()
-	s.mockConfig.On("GetString", "queue.failed.table").Return("failed_jobs").Once()
+	s.mockConfig.EXPECT().GetString("queue.default").Return("async").Times(3)
+	s.mockConfig.EXPECT().GetString("app.name").Return("goravel").Times(3)
+	s.mockConfig.EXPECT().GetString("queue.connections.async.queue", "default").Return("default").Once()
+	s.mockConfig.EXPECT().GetString("queue.connections.async.driver").Return("async").Twice()
+	s.mockConfig.EXPECT().GetInt("queue.connections.async.size", 100).Return(10).Twice()
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
@@ -99,13 +88,11 @@ func (s *DriverAsyncTestSuite) TestDelayAsyncQueue() {
 }
 
 func (s *DriverAsyncTestSuite) TestCustomAsyncQueue() {
-	s.mockConfig.On("GetString", "queue.default").Return("custom").Times(4)
-	s.mockConfig.On("GetString", "app.name").Return("goravel").Times(3)
-	s.mockConfig.On("GetString", "queue.connections.custom.queue", "default").Return("default").Once()
-	s.mockConfig.On("GetString", "queue.connections.custom.driver").Return("async").Times(2)
-	s.mockConfig.On("GetInt", "queue.connections.custom.size", 100).Return(10).Twice()
-	s.mockConfig.On("GetString", "queue.failed.database").Return("database").Once()
-	s.mockConfig.On("GetString", "queue.failed.table").Return("failed_jobs").Once()
+	s.mockConfig.EXPECT().GetString("queue.default").Return("custom").Times(3)
+	s.mockConfig.EXPECT().GetString("app.name").Return("goravel").Times(3)
+	s.mockConfig.EXPECT().GetString("queue.connections.custom.queue", "default").Return("default").Once()
+	s.mockConfig.EXPECT().GetString("queue.connections.custom.driver").Return("async").Times(2)
+	s.mockConfig.EXPECT().GetInt("queue.connections.custom.size", 100).Return(10).Twice()
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
@@ -127,21 +114,12 @@ func (s *DriverAsyncTestSuite) TestCustomAsyncQueue() {
 }
 
 func (s *DriverAsyncTestSuite) TestErrorAsyncQueue() {
-	s.mockConfig.On("GetString", "queue.default").Return("async").Times(4)
-	s.mockConfig.On("GetString", "app.name").Return("goravel").Times(3)
-	s.mockConfig.On("GetString", "queue.connections.async.queue", "default").Return("default").Once()
-	s.mockConfig.On("GetString", "queue.connections.async.driver").Return("async").Once()
-	s.mockConfig.On("GetInt", "queue.connections.async.size", 100).Return(10).Once()
-	s.mockConfig.On("GetString", "queue.connections.redis.driver").Return("").Once()
-	s.mockConfig.On("GetString", "queue.failed.database").Return("database").Once()
-	s.mockConfig.On("GetString", "queue.failed.table").Return("failed_jobs").Once()
-
-	mockOrm := mocksorm.NewOrm(s.T())
-	mockQuery := mocksorm.NewQuery(s.T())
-	mockOrm.EXPECT().Connection("database").Return(mockOrm)
-	mockOrm.On("Query").Return(mockQuery)
-	mockQuery.On("Table", "failed_jobs").Return(mockQuery)
-	OrmFacade = mockOrm
+	s.mockConfig.EXPECT().GetString("queue.default").Return("async").Times(3)
+	s.mockConfig.EXPECT().GetString("app.name").Return("goravel").Times(3)
+	s.mockConfig.EXPECT().GetString("queue.connections.async.queue", "default").Return("default").Once()
+	s.mockConfig.EXPECT().GetString("queue.connections.async.driver").Return("async").Once()
+	s.mockConfig.EXPECT().GetInt("queue.connections.async.size", 100).Return(10).Once()
+	s.mockConfig.EXPECT().GetString("queue.connections.redis.driver").Return("").Twice()
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
@@ -161,13 +139,11 @@ func (s *DriverAsyncTestSuite) TestErrorAsyncQueue() {
 }
 
 func (s *DriverAsyncTestSuite) TestChainAsyncQueue() {
-	s.mockConfig.On("GetString", "queue.default").Return("async").Times(4)
-	s.mockConfig.On("GetString", "app.name").Return("goravel").Times(3)
-	s.mockConfig.On("GetString", "queue.connections.async.queue", "default").Return("default").Once()
-	s.mockConfig.On("GetString", "queue.connections.async.driver").Return("async").Twice()
-	s.mockConfig.On("GetInt", "queue.connections.async.size", 100).Return(10).Twice()
-	s.mockConfig.On("GetString", "queue.failed.database").Return("database").Once()
-	s.mockConfig.On("GetString", "queue.failed.table").Return("failed_jobs").Once()
+	s.mockConfig.EXPECT().GetString("queue.default").Return("async").Times(3)
+	s.mockConfig.EXPECT().GetString("app.name").Return("goravel").Times(3)
+	s.mockConfig.EXPECT().GetString("queue.connections.async.queue", "default").Return("default").Once()
+	s.mockConfig.EXPECT().GetString("queue.connections.async.driver").Return("async").Twice()
+	s.mockConfig.EXPECT().GetInt("queue.connections.async.size", 100).Return(10).Twice()
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()

queue/driver_sync_test.go

@@ -37,22 +37,23 @@ func TestDriverSyncTestSuite(t *testing.T) {
 }
 
 func (s *DriverSyncTestSuite) SetupTest() {
-	s.mockConfig.On("GetString", "queue.default").Return("sync").Once()
-	s.mockConfig.On("GetString", "app.name").Return("goravel").Once()
-	s.mockConfig.On("GetString", "queue.connections.sync.queue", "default").Return("default").Once()
 	testSyncJob = 0
 	testChainSyncJob = 0
 }
 
 func (s *DriverSyncTestSuite) TestSyncQueue() {
+	s.mockConfig.EXPECT().GetString("queue.default").Return("sync").Twice()
+	s.mockConfig.EXPECT().GetString("app.name").Return("goravel").Once()
+	s.mockConfig.EXPECT().GetString("queue.connections.sync.queue", "default").Return("default").Once()
 	s.Nil(s.app.Job(&TestSyncJob{}, []any{"TestSyncQueue", 1}).DispatchSync())
 	s.Equal(1, testSyncJob)
 }
 
 func (s *DriverSyncTestSuite) TestChainSyncQueue() {
-	s.mockConfig.On("GetString", "queue.default").Return("sync").Times(2)
-	s.mockConfig.On("GetString", "app.name").Return("goravel").Once()
-	s.mockConfig.On("GetString", "queue.connections.sync.driver").Return("sync").Once()
+	s.mockConfig.EXPECT().GetString("queue.default").Return("sync").Twice()
+	s.mockConfig.EXPECT().GetString("app.name").Return("goravel").Twice()
+	s.mockConfig.EXPECT().GetString("queue.connections.sync.queue", "default").Return("default").Once()
+	s.mockConfig.EXPECT().GetString("queue.connections.sync.driver").Return("sync").Once()
 
 	s.Nil(s.app.Chain([]queue.Jobs{
 		{

queue/worker.go

@@ -21,7 +21,6 @@
 	job           queue.JobRepository
 	queue         string
 	wg            sync.WaitGroup
-	failedWg      sync.WaitGroup
 }
 
 func NewWorker(config queue.Config, concurrent int, connection string, queue string, job queue.JobRepository) *Worker {
@@ -43,7 +42,7 @@
 		return err
 	}
 	if driver.Driver() == queue.DriverSync {
 		return errors.QueueDriverSyncNotNeedRun.Args(r.queue)
 	}
 
 	for i := 0; i < r.concurrent; i++ {
@@ -62,30 +61,26 @@
 				}
 
 				if err = r.job.Call(job.Signature(), args); err != nil {
-					select {
-					case r.failedJobChan <- FailedJob{
+					r.failedJobChan <- FailedJob{
 						UUID:       uuid.New(),
 						Connection: r.connection,
 						Queue:      r.queue,
 						Payload:    args,
 						Exception:  err.Error(),
 						FailedAt:   carbon.DateTime{Carbon: carbon.Now()},
-					}:
-					default:
-						LogFacade.Error(errors.New("failed to send failed job to channel"))
 					}
 				}
 			}
 		}()
 	}
 
-	r.failedWg.Add(1)
+	r.wg.Add(1)
 	go func() {
-		defer r.failedWg.Done()
+		defer r.wg.Done()
 		for job := range r.failedJobChan {
 			if err = r.config.FailedJobsQuery().Create(&job); err != nil {
 				LogFacade.Error(errors.QueueFailedToSaveFailedJob.Args(err))
 			}
 		}
 	}()
 
@@ -96,6 +91,5 @@
 	r.isShutdown.Store(true)
 	r.wg.Wait()
 	close(r.failedJobChan)
-	r.failedWg.Wait()
 	return nil
 }