Skip to content

Commit a7e4822

Browse files
committed
update queuer start structure, update readme
1 parent db25a3f commit a7e4822

18 files changed

Lines changed: 598 additions & 429 deletions

.github/workflows/tag.yml

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
name: Run auto tagging
2+
on:
3+
push:
4+
branches:
5+
- main
6+
pull_request:
7+
branches:
8+
- main
9+
jobs:
10+
Patch:
11+
runs-on: ubuntu-latest
12+
steps:
13+
- name: Checkout Source
14+
uses: actions/checkout@v4
15+
- name: Minor version for each merge
16+
uses: anothrNick/github-tag-action@1.73.0
17+
env:
18+
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
19+
TAG_PREFIX: v
20+
INITIAL_VERSION: 1.0.0

README.md

Lines changed: 58 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -56,31 +56,78 @@ You can find a full example (the same as above plus a more detailed example) in
5656

5757
---
5858

59-
## New queuer
59+
## NewQueuer
6060

61-
The `NewQueuer` function initializes a new Queuer instance, setting up the core components for job processing.
61+
`NewQueuer` is a convenience constructor that creates a new Queuer instance using default database configuration derived from environment variables. It acts as a wrapper around `NewQueuerWithDB`.
62+
`NewQueuerWithDB` is the primary constructor for creating a new Queuer instance. It allows for explicit database configuration and initializes all necessary components, including database handlers, internal event listeners, and the worker.
6263

6364
```go
64-
func NewQueuer(name string, maxConcurrency int, opts ...WorkerOption) *Queuer
65+
func NewQueuer(name string, maxConcurrency int, options ...*model.OnError) *Queuer
66+
67+
func NewQueuerWithDB(name string, maxConcurrency int, dbConfig *helper.DatabaseConfiguration, options ...*model.OnError) *Queuer
68+
```
69+
70+
- `name`: A `string` identifier for this queuer instance.
71+
- `maxConcurrency`: An `int` specifying the maximum number of jobs this queuer can process concurrently.
72+
- `dbConfig`: An optional `*helper.DatabaseConfiguration`. If nil, the configuration will be loaded from environment variables.
73+
- `options`: Optional `OnError` configurations to apply to the worker.
74+
75+
This function performs the following setup:
76+
- Initializes a logger.
77+
- Sets up the database connection using the provided `dbConfig` or environment variables.
78+
- Creates `JobDBHandler` and `WorkerDBHandler` instances for database interactions.
79+
- Initializes internal `core.Listener` instances for `jobInsert`, `jobUpdate`, and `jobDelete` events.
80+
- Creates and inserts a new `model.Worker` into the database based on the provided `name`, `maxConcurrency`, and `options`.
81+
- If any critical error occurs during this initialization (e.g., database connection failure, worker creation error), the function will log a panic error and exit the program. It returns a pointer to the newly configured `Queuer` instance.
82+
83+
---
84+
85+
## Start
86+
87+
The Start method initiates the operational lifecycle of the Queuer. It sets up the main context, initializes database listeners, and begins the job processing and polling loops in a dedicated goroutine.
88+
89+
```go
90+
func (q *Queuer) Start(ctx context.Context, cancel context.CancelFunc)
6591
```
6692

67-
- `name`: A string identifier for this specific queuer instance. This is useful for distinguishing between multiple queuers in your application or logs.
68-
- `maxConcurrency`: An integer representing the maximum number of jobs that this queuer can process concurrently. This controls the worker's parallelism.
69-
- `opts ...WorkerOption`: Optional `WorkerOption` configurations that allow you to customize the worker's behavior, such as error handling strategies.
93+
- `ctx`: The parent context.Context for the queuer's operations. This context will control the overall lifetime of the queuer.
94+
- `cancel`: The context.CancelFunc associated with the provided ctx. This function should be called to gracefully stop the queuer.
95+
96+
Upon calling Start:
97+
- It performs a basic check to ensure internal listeners are initialized.
98+
- Db listeners and broadcasters are created to listen to job events (inserts, updates, deletes).
99+
- It starts a poller to periodically poll the database for new jobs to process (5 minute interval).
100+
- It signals its readiness via an internal channel, ensuring the `Start` method returns only when the core loops are active.
101+
102+
The method includes a timeout mechanism (5 seconds) to detect if the queuer fails to start its internal processes promptly, panicking if the timeout is exceeded.
103+
If the queuer is not not properly initialized (created by calling `NewQueuer`), or if there's an error creating the database listeners, the function will panic.
104+
105+
---
106+
107+
## StartWithoutWorker
108+
109+
The `StartWithoutWorker` method provides a way to start the `Queuer` instance without an active worker. This is particularly useful for scenarios where you need to interact with the job queue (e.g., add jobs, check job status) but don't intend for this specific instance to actively process them.
110+
111+
```go
112+
func (q *Queuer) StartWithoutWorker(ctx context.Context, cancel context.CancelFunc, withoutListeners bool, dbConnection ...*sql.DB)
113+
```
70114

71-
This function handles the entire setup process: it establishes the database connection, configures the necessary job listeners, and creates an associated worker. If any part of this initialization fails, `NewQueuer` will log a panic error and exit the program to prevent an improperly configured queuer from running. It returns a pointer to the newly created `Queuer` instance, ready to accept and process jobs.
115+
- `ctx`: The parent context.Context for the queuer's operations.
116+
- `cancel`: The context.CancelFunc associated with the provided ctx.
117+
- `withoutListeners`: A `bool` flag. If true, the database.NewQueuerDBListener instances for job and job_archive tables will not be created.
118+
- `dbConnection`: An optional existing `*sql.DB` connection to use. If provided, the queuer will use this connection; otherwise, it will create a new one based on environment variables.
72119

73120
---
74121

75-
## New queuer without worker
122+
## Stop
76123

77-
The `NewQueuerWithoutWorker` function provides a way to initialize a `Queuer` instance without an active worker. This is particularly useful for scenarios where you need to interact with the job queue (e.g., add jobs, check job status) but don't intend for this specific instance to actively process them.
124+
The `Stop` method gracefully shuts down the Queuer instance, releasing resources and ensuring ongoing operations are properly concluded.
78125

79126
```go
80-
func NewQueuerWithoutWorker() *Queuer
127+
func (q *Queuer) Stop() error
81128
```
82129

83-
This function only initializes the database connection and job listeners. It omits the worker component, making it suitable for services that might, for example, serve job status endpoints or solely add jobs to the queue, without consuming computational resources for job execution. Similar to `NewQueuer`, any initialization errors will result in a panic and program exit. It returns a pointer to the newly created `Queuer` instance.
130+
The `Stop` method cancels all jobs, closes db listeners and returns an error if any step of the stopping process encounters an issue
84131

85132
---
86133

database/dbJob.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,10 @@ type JobDBHandler struct {
4141

4242
// NewJobDBHandler creates a new instance of JobDBHandler.
4343
func NewJobDBHandler(dbConnection *helper.Database, withTableDrop bool) (*JobDBHandler, error) {
44+
if dbConnection == nil {
45+
return nil, fmt.Errorf("database connection is nil")
46+
}
47+
4448
jobDbHandler := &JobDBHandler{
4549
db: dbConnection,
4650
}

database/dbJob_test.go

Lines changed: 85 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,11 @@ import (
1313
)
1414

1515
func TestJobNewJobDBHandler(t *testing.T) {
16-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
16+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
17+
dbConfig, err := helper.NewDatabaseConfiguration()
18+
if err != nil {
19+
t.Fatalf("failed to create database configuration: %v", err)
20+
}
1721
database := helper.NewTestDatabase(dbConfig)
1822

1923
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -25,7 +29,11 @@ func TestJobNewJobDBHandler(t *testing.T) {
2529
}
2630

2731
func TestJobCheckTableExistance(t *testing.T) {
28-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
32+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
33+
dbConfig, err := helper.NewDatabaseConfiguration()
34+
if err != nil {
35+
t.Fatalf("failed to create database configuration: %v", err)
36+
}
2937
database := helper.NewTestDatabase(dbConfig)
3038

3139
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -37,7 +45,11 @@ func TestJobCheckTableExistance(t *testing.T) {
3745
}
3846

3947
func TestJobCreateTable(t *testing.T) {
40-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
48+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
49+
dbConfig, err := helper.NewDatabaseConfiguration()
50+
if err != nil {
51+
t.Fatalf("failed to create database configuration: %v", err)
52+
}
4153
database := helper.NewTestDatabase(dbConfig)
4254

4355
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -48,7 +60,11 @@ func TestJobCreateTable(t *testing.T) {
4860
}
4961

5062
func TestJobDropTable(t *testing.T) {
51-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
63+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
64+
dbConfig, err := helper.NewDatabaseConfiguration()
65+
if err != nil {
66+
t.Fatalf("failed to create database configuration: %v", err)
67+
}
5268
database := helper.NewTestDatabase(dbConfig)
5369

5470
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -59,7 +75,11 @@ func TestJobDropTable(t *testing.T) {
5975
}
6076

6177
func TestJobInsertJob(t *testing.T) {
62-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
78+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
79+
dbConfig, err := helper.NewDatabaseConfiguration()
80+
if err != nil {
81+
t.Fatalf("failed to create database configuration: %v", err)
82+
}
6383
database := helper.NewTestDatabase(dbConfig)
6484

6585
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -79,7 +99,11 @@ func TestJobInsertJob(t *testing.T) {
7999
}
80100

81101
func TestJobInsertJobTx(t *testing.T) {
82-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
102+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
103+
dbConfig, err := helper.NewDatabaseConfiguration()
104+
if err != nil {
105+
t.Fatalf("failed to create database configuration: %v", err)
106+
}
83107
database := helper.NewTestDatabase(dbConfig)
84108

85109
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -105,7 +129,11 @@ func TestJobInsertJobTx(t *testing.T) {
105129
}
106130

107131
func TestJobBatchInsertJobs(t *testing.T) {
108-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
132+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
133+
dbConfig, err := helper.NewDatabaseConfiguration()
134+
if err != nil {
135+
t.Fatalf("failed to create database configuration: %v", err)
136+
}
109137
database := helper.NewTestDatabase(dbConfig)
110138

111139
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -125,7 +153,11 @@ func TestJobBatchInsertJobs(t *testing.T) {
125153
}
126154

127155
func TestJobUpdateJobsInitial(t *testing.T) {
128-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
156+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
157+
dbConfig, err := helper.NewDatabaseConfiguration()
158+
if err != nil {
159+
t.Fatalf("failed to create database configuration: %v", err)
160+
}
129161
database := helper.NewTestDatabase(dbConfig)
130162

131163
// Prerequisite: Insert a worker for the job to be associated with
@@ -164,7 +196,11 @@ func TestJobUpdateJobsInitial(t *testing.T) {
164196
}
165197

166198
func TestJobUpdateJobFinal(t *testing.T) {
167-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
199+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
200+
dbConfig, err := helper.NewDatabaseConfiguration()
201+
if err != nil {
202+
t.Fatalf("failed to create database configuration: %v", err)
203+
}
168204
database := helper.NewTestDatabase(dbConfig)
169205

170206
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -186,7 +222,11 @@ func TestJobUpdateJobFinal(t *testing.T) {
186222
}
187223

188224
func TestJobDeleteJob(t *testing.T) {
189-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
225+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
226+
dbConfig, err := helper.NewDatabaseConfiguration()
227+
if err != nil {
228+
t.Fatalf("failed to create database configuration: %v", err)
229+
}
190230
database := helper.NewTestDatabase(dbConfig)
191231

192232
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -209,7 +249,11 @@ func TestJobDeleteJob(t *testing.T) {
209249
}
210250

211251
func TestJobSelectJob(t *testing.T) {
212-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
252+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
253+
dbConfig, err := helper.NewDatabaseConfiguration()
254+
if err != nil {
255+
t.Fatalf("failed to create database configuration: %v", err)
256+
}
213257
database := helper.NewTestDatabase(dbConfig)
214258

215259
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -228,7 +272,11 @@ func TestJobSelectJob(t *testing.T) {
228272
}
229273

230274
func TestJobSelectAllJobs(t *testing.T) {
231-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
275+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
276+
dbConfig, err := helper.NewDatabaseConfiguration()
277+
if err != nil {
278+
t.Fatalf("failed to create database configuration: %v", err)
279+
}
232280
database := helper.NewTestDatabase(dbConfig)
233281

234282
newJobCount := 5
@@ -254,7 +302,11 @@ func TestJobSelectAllJobs(t *testing.T) {
254302
}
255303

256304
func TestJobSelectAllJobsByWorkerRID(t *testing.T) {
257-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
305+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
306+
dbConfig, err := helper.NewDatabaseConfiguration()
307+
if err != nil {
308+
t.Fatalf("failed to create database configuration: %v", err)
309+
}
258310
database := helper.NewTestDatabase(dbConfig)
259311

260312
workerConcurrency := 3
@@ -297,7 +349,11 @@ func TestJobSelectAllJobsByWorkerRID(t *testing.T) {
297349
}
298350

299351
func TestJobSelectAllJobsBySearch(t *testing.T) {
300-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
352+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
353+
dbConfig, err := helper.NewDatabaseConfiguration()
354+
if err != nil {
355+
t.Fatalf("failed to create database configuration: %v", err)
356+
}
301357
database := helper.NewTestDatabase(dbConfig)
302358

303359
searchTerm := "TestTaskSearch"
@@ -335,7 +391,11 @@ func TestJobSelectAllJobsBySearch(t *testing.T) {
335391
}
336392

337393
func TestJobSelectJobFromArchive(t *testing.T) {
338-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
394+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
395+
dbConfig, err := helper.NewDatabaseConfiguration()
396+
if err != nil {
397+
t.Fatalf("failed to create database configuration: %v", err)
398+
}
339399
database := helper.NewTestDatabase(dbConfig)
340400

341401
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -361,7 +421,11 @@ func TestJobSelectJobFromArchive(t *testing.T) {
361421
}
362422

363423
func TestJobSelectAllJobsFromArchive(t *testing.T) {
364-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
424+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
425+
dbConfig, err := helper.NewDatabaseConfiguration()
426+
if err != nil {
427+
t.Fatalf("failed to create database configuration: %v", err)
428+
}
365429
database := helper.NewTestDatabase(dbConfig)
366430

367431
jobDBHandler, err := NewJobDBHandler(database, true)
@@ -392,7 +456,11 @@ func TestJobSelectAllJobsFromArchive(t *testing.T) {
392456
}
393457

394458
func TestJobSelectAllJobsFromArchiveBySearch(t *testing.T) {
395-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
459+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
460+
dbConfig, err := helper.NewDatabaseConfiguration()
461+
if err != nil {
462+
t.Fatalf("failed to create database configuration: %v", err)
463+
}
396464
database := helper.NewTestDatabase(dbConfig)
397465

398466
searchTerm := "TestTaskSearch"

database/dbListener_test.go

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,11 @@ import (
1111
)
1212

1313
func TestNewQueuerDBListener(t *testing.T) {
14-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
14+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
15+
dbConfig, err := helper.NewDatabaseConfiguration()
16+
if err != nil {
17+
t.Fatalf("failed to create database configuration: %v", err)
18+
}
1519
listener, err := NewQueuerDBListener(dbConfig, "test_channel")
1620
assert.NoError(t, err, "Expected NewQueuerDBListener to not return an error")
1721
assert.NotNil(t, listener, "Expected listener to be created")
@@ -23,7 +27,11 @@ func TestNewQueuerDBListener(t *testing.T) {
2327
}
2428

2529
func TestListen(t *testing.T) {
26-
dbConfig := helper.NewTestDatabaseConfig(dbPort)
30+
helper.SetTestDatabaseConfigEnvs(t, dbPort)
31+
dbConfig, err := helper.NewDatabaseConfiguration()
32+
if err != nil {
33+
t.Fatalf("failed to create database configuration: %v", err)
34+
}
2735
listener, err := NewQueuerDBListener(dbConfig, "test_channel")
2836
assert.NoError(t, err, "Expected NewQueuerDBListener to not return an error")
2937
assert.NotNil(t, listener, "Expected listener to be created")

database/dbWorker.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,10 @@ type WorkerDBHandler struct {
3232

3333
// NewWorkerDBHandler creates a new instance of WorkerDBHandler.
3434
func NewWorkerDBHandler(dbConnection *helper.Database, withTableDrop bool) (*WorkerDBHandler, error) {
35+
if dbConnection == nil {
36+
return nil, fmt.Errorf("database connection is nil")
37+
}
38+
3539
workerDbHandler := &WorkerDBHandler{
3640
db: dbConnection,
3741
}

0 commit comments

Comments
 (0)