Skip to content

Commit

Permalink
feat: optimize
Browse files Browse the repository at this point in the history
  • Loading branch information
devhaozi committed Jan 11, 2025
1 parent 88ffd04 commit e019c6d
Show file tree
Hide file tree
Showing 3 changed files with 22 additions and 28 deletions.
21 changes: 7 additions & 14 deletions queue/driver_sync_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,35 +24,28 @@ type DriverSyncTestSuite struct {
}

func TestDriverSyncTestSuite(t *testing.T) {
mockConfig := mocksconfig.NewConfig(t)
mockQueue := mocksqueue.NewQueue(t)
app := NewApplication(mockConfig)

app.Register([]queue.Job{&TestSyncJob{}, &TestChainSyncJob{}})
suite.Run(t, &DriverSyncTestSuite{
app: app,
mockConfig: mockConfig,
mockQueue: mockQueue,
})
suite.Run(t, new(DriverSyncTestSuite))
}

func (s *DriverSyncTestSuite) SetupTest() {
testSyncJob = 0
testChainSyncJob = 0
s.mockConfig = mocksconfig.NewConfig(s.T())
s.mockQueue = mocksqueue.NewQueue(s.T())
s.app = NewApplication(s.mockConfig)
s.app.Register([]queue.Job{&TestSyncJob{}, &TestChainSyncJob{}})
s.mockConfig.EXPECT().GetString("queue.default").Return("sync").Twice()
s.mockConfig.EXPECT().GetString("queue.connections.sync.queue", "default").Return("default").Once()
}

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.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{
Expand Down
3 changes: 2 additions & 1 deletion queue/worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -84,12 +84,13 @@ func (r *Worker) Run() error {
}
}()

r.wg.Wait()

return nil
}

func (r *Worker) Shutdown() error {
r.isShutdown.Store(true)
r.wg.Wait()
close(r.failedJobChan)
return nil
}
26 changes: 13 additions & 13 deletions queue/worker_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import (

type WorkerTestSuite struct {
suite.Suite
app *Application
mockConfig *mocksconfig.Config
}

Expand All @@ -25,27 +26,27 @@ func TestWorkerTestSuite(t *testing.T) {

func (s *WorkerTestSuite) SetupTest() {
s.mockConfig = mocksconfig.NewConfig(s.T())
s.mockConfig.EXPECT().GetString("queue.default").Return("async").Times(4)
s.mockConfig.EXPECT().GetString("app.name").Return("goravel").Times(3)
s.mockConfig.EXPECT().GetString("queue.connections.async.queue", "default").Return("default").Times(3)
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").Times(2)
s.mockConfig.EXPECT().GetString("queue.connections.async.driver").Return("async").Twice()
s.mockConfig.EXPECT().GetInt("queue.connections.async.size", 100).Return(10).Twice()
s.app = NewApplication(s.mockConfig)
}

func (s *WorkerTestSuite) TestRun_Success() {
app := NewApplication(s.mockConfig)
testJob := new(MockJob)
app.Register([]contractsqueue.Job{testJob})
s.app.Register([]contractsqueue.Job{testJob})

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
go func(ctx context.Context) {
s.NoError(app.Worker().Run())
s.NoError(s.app.Worker().Run())
<-ctx.Done()
s.NoError(app.Worker().Shutdown())
s.NoError(s.app.Worker().Shutdown())
}(ctx)

time.Sleep(1 * time.Second)
s.NoError(app.Job(testJob, []any{}).Dispatch())
s.NoError(s.app.Job(testJob, []any{}).Dispatch())
time.Sleep(2 * time.Second)
s.True(testJob.called)
cancel()
Expand All @@ -64,19 +65,18 @@ func (s *WorkerTestSuite) TestRun_FailedJob() {
mockQuery.EXPECT().Create(mock.Anything).Return(nil)
OrmFacade = mockOrm

app := NewApplication(s.mockConfig)
testJob := new(MockFailedJob)
app.Register([]contractsqueue.Job{testJob})
s.app.Register([]contractsqueue.Job{testJob})

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
go func(ctx context.Context) {
s.NoError(app.Worker().Run())
s.NoError(s.app.Worker().Run())
<-ctx.Done()
s.NoError(app.Worker().Shutdown())
s.NoError(s.app.Worker().Shutdown())
}(ctx)

time.Sleep(1 * time.Second)
s.NoError(app.Job(testJob, []any{}).Dispatch())
s.NoError(s.app.Job(testJob, []any{}).Dispatch())
time.Sleep(2 * time.Second)
s.True(testJob.called)
cancel()
Expand Down

0 comments on commit e019c6d

Please sign in to comment.