Skip to content

Commit c516633

Browse files
ngoyal16ngoyal16
authored
Merge pull request #6 from pixelvide/feature/queue-publisher-13572187248206151659
Add Queue Publisher
2 parents 2381294 + 445c5d1 commit c516633

30 files changed

Lines changed: 1187 additions & 55 deletions

‎.github/workflows/release.yml‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
name: Release
2+
3+
on:
4+
push:
5+
branches:
6+
- main
7+
8+
permissions:
9+
contents: write
10+
pull-requests: write
11+
12+
jobs:
13+
release-please:
14+
runs-on: ubuntu-latest
15+
steps:
16+
- uses: googleapis/release-please-action@v4
17+
with:
18+
release-type: go

‎README.md‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,10 @@ For detailed package documentation, see [doc.go](doc.go) or run `go doc github.c
2222

2323
For AI agents or developers needing a quick overview of the codebase structure and import paths, refer to [AGENTS.md](AGENTS.md).
2424

25+
See [docs/register_jobs.md](docs/register_jobs.md) for details on registering job handlers.
26+
See [docs/logging.md](docs/logging.md) for details on using the integrated logger and tracing.
27+
See [docs/scheduler.md](docs/scheduler.md) for details on using the scheduler.
28+
2529
## Usage
2630

2731
### 1. Define Handlers
@@ -65,7 +69,7 @@ func main() {
6569
})
6670

6771
// Setup Worker
68-
w := worker.NewWorker(driver, nil, "default", 5)
72+
w := worker.NewWorker(driver, nil, "default", 5, "my-app", nil)
6973

7074
// Run
7175
w.Run(context.Background())

‎cmd/worker/main.go‎

Lines changed: 40 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,22 @@ package main
22

33
import (
44
"context"
5+
"fmt"
56

7+
"github.com/pixelvide/laravel-go/pkg/config"
68
"github.com/pixelvide/laravel-go/pkg/queue"
79
"github.com/pixelvide/laravel-go/pkg/root"
10+
"github.com/pixelvide/laravel-go/pkg/schedule"
811
"github.com/pixelvide/laravel-go/pkg/telemetry"
12+
"github.com/rs/zerolog/log"
13+
"github.com/spf13/cobra"
914

10-
_ "github.com/pixelvide/laravel-go/pkg/console" // Register commands
15+
// Ensure drivers are loaded
16+
_ "github.com/go-sql-driver/mysql"
17+
_ "github.com/lib/pq"
18+
19+
// Ensure console commands are registered
20+
_ "github.com/pixelvide/laravel-go/pkg/console"
1121
)
1222

1323
// ExampleHandler is a sample job handler
@@ -33,10 +43,37 @@ func ExampleHandler(ctx context.Context, job *queue.Job) error {
3343
}
3444

3545
func main() {
36-
// 1. Register Handlers
46+
// 1. Load Configuration (Optional, for app usage)
47+
// You can load the config here to use it in your own code.
48+
// The queue:work command loads it automatically, so this is just for demonstration.
49+
cfg, err := config.Load()
50+
if err != nil {
51+
log.Warn().Err(err).Msg("Could not load config")
52+
} else {
53+
log.Info().Str("app_name", cfg.App.Name).Str("env", cfg.App.Env).Msg("Loaded application config")
54+
}
55+
56+
// 2. Register Handlers
3757
// Register a handler for a hypothetical Laravel job "App\Jobs\ProcessPodcast"
3858
queue.Register("App\\Jobs\\ProcessPodcast", ExampleHandler)
3959

40-
// 2. Execute Root Command
60+
// 3. Register Scheduled Tasks
61+
// Example: Run every minute on one server
62+
schedule.Register("* * * * *", func() {
63+
fmt.Println("Running scheduled task: Every Minute")
64+
}, schedule.OnOneServer("every-minute-task"))
65+
66+
// 4. Register Custom Commands
67+
// Example: A custom "hello" command
68+
helloCmd := &cobra.Command{
69+
Use: "hello",
70+
Short: "Prints a hello message",
71+
Run: func(cmd *cobra.Command, args []string) {
72+
fmt.Printf("Hello from %s!\n", cfg.App.Name)
73+
},
74+
}
75+
root.GetRoot().AddCommand(helloCmd)
76+
77+
// 5. Execute Root Command
4178
root.Execute()
4279
}

‎doc.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131
// func main() {
3232
// queue.Register("App\\Jobs\\MyJob", MyHandler)
3333
// driver := redis.NewRedisDriver(config.RedisConfig{Addr: "localhost:6379"})
34-
// w := worker.NewWorker(driver, nil, "default", 5)
34+
// w := worker.NewWorker(driver, nil, "default", 5, "my-app", nil)
3535
// w.Run(context.Background())
3636
// }
3737
package laravelgo

‎docs/logging.md‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
# Logging and Tracing
2+
3+
The `laravel-go` library provides integrated structured logging (using `zerolog`) and distributed tracing (using `OpenTelemetry`).
4+
5+
## Trace IDs and Span IDs
6+
7+
When a job is processed, the worker automatically:
8+
1. Starts a new OpenTelemetry trace.
9+
2. Creates a structured logger attached to the context.
10+
3. Injects the `trace_id` and `job_id` into the logger.
11+
12+
This ensures that every log message generated during the job execution is automatically tagged with the Trace ID, allowing you to correlate logs across different services or even within a single job execution.
13+
14+
## Using the Logger
15+
16+
To use the logger in your job handlers, you should retrieve it from the context using `telemetry.LoggerFromContext(ctx)`.
17+
18+
### Example
19+
20+
```go
21+
package main
22+
23+
import (
24+
"context"
25+
"github.com/pixelvide/laravel-go/pkg/queue"
26+
"github.com/pixelvide/laravel-go/pkg/telemetry"
27+
)
28+
29+
func ProcessOrder(ctx context.Context, job *queue.Job) error {
30+
// 1. Get the logger from the context
31+
// This logger already has "trace_id" and "job_uuid" fields set.
32+
logger := telemetry.LoggerFromContext(ctx)
33+
34+
orderID := job.GetArg("orderId")
35+
36+
// 2. Log messages
37+
// These logs will include the trace context automatically.
38+
logger.Info().
39+
Any("order_id", orderID).
40+
Msg("Starting to process order")
41+
42+
if err := processOrder(orderID); err != nil {
43+
logger.Error().Err(err).Msg("Failed to process order")
44+
return err
45+
}
46+
47+
logger.Info().Msg("Order processed successfully")
48+
return nil
49+
}
50+
```
51+
52+
### Log Output Example
53+
54+
The output will look something like this (formatted for readability):
55+
56+
```json
57+
{
58+
"level": "info",
59+
"time": "2023-10-27T10:00:00Z",
60+
"service": "LaravelGoApp",
61+
"command": "queue:work",
62+
"trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
63+
"job_uuid": "550e8400-e29b-41d4-a716-446655440000",
64+
"job_name": "App\\Jobs\\ProcessOrder",
65+
"order_id": 12345,
66+
"message": "Starting to process order"
67+
}
68+
```
69+
70+
## Configuring Telemetry
71+
72+
The telemetry system is initialized automatically when using the `queue:work` command. You can customize the behavior by setting the global logger or tracer provider in your application setup if needed, but the default setup is sufficient for most use cases.

‎docs/register_jobs.md‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
# Registering Jobs
2+
3+
To process Laravel jobs in Go, you need to register handlers that map to the Laravel job class names.
4+
5+
## Handler Signature
6+
7+
A job handler is a function with the following signature:
8+
9+
```go
10+
func(ctx context.Context, job *queue.Job) error
11+
```
12+
13+
* `ctx`: The context, which includes tracing information and the logger.
14+
* `job`: The job object containing the payload and unserialized PHP data.
15+
16+
## Registering a Handler
17+
18+
You should register your handlers in your application's entry point (e.g., `main.go`) before executing the root command.
19+
20+
```go
21+
package main
22+
23+
import (
24+
"context"
25+
"github.com/pixelvide/laravel-go/pkg/queue"
26+
"github.com/pixelvide/laravel-go/pkg/root"
27+
"github.com/pixelvide/laravel-go/pkg/telemetry"
28+
29+
// Import console to register queue:work command
30+
_ "github.com/pixelvide/laravel-go/pkg/console"
31+
)
32+
33+
func ProcessPodcast(ctx context.Context, job *queue.Job) error {
34+
// Get the logger
35+
logger := telemetry.LoggerFromContext(ctx)
36+
37+
// Access job arguments (public properties of the Laravel job class)
38+
podcastID := job.GetArg("podcastId")
39+
40+
logger.Info().Any("podcast_id", podcastID).Msg("Processing podcast")
41+
42+
return nil
43+
}
44+
45+
func main() {
46+
// Register the handler
47+
// The string must match the Laravel class name exactly
48+
queue.Register("App\\Jobs\\ProcessPodcast", ProcessPodcast)
49+
50+
// Run the CLI
51+
root.Execute()
52+
}
53+
```
54+
55+
## Accessing Job Data
56+
57+
The `queue.Job` struct provides a helper method `GetArg(key string)` to access public properties of the unserialized PHP job object.
58+
59+
```go
60+
if id := job.GetArg("id"); id != nil {
61+
// Use id
62+
}
63+
```
64+
65+
## Logging
66+
67+
The system uses `zerolog` and `OpenTelemetry`. You should retrieve the logger from the context to ensure logs are correlated with the trace ID and job ID.
68+
69+
```go
70+
logger := telemetry.LoggerFromContext(ctx)
71+
logger.Info().Msg("Log message")
72+
```

‎docs/scheduler.md‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
# Scheduler
2+
3+
The `laravel-go` library provides a robust task scheduler inspired by Laravel's schedule. It supports:
4+
- Cron expression scheduling
5+
- Overlapping prevention
6+
- Distributed locks (`OnOneServer`) using Redis or Database
7+
8+
## Commands
9+
10+
To run the scheduler, use the `schedule:run` command. This command will start the scheduler and block until interrupted (SIGINT/SIGTERM).
11+
12+
```bash
13+
go run main.go schedule:run
14+
```
15+
16+
## Configuration
17+
18+
The scheduler automatically uses the `CACHE_STORE` configuration from your `.env` file to determine the lock provider for `OnOneServer` tasks.
19+
20+
- `CACHE_STORE=redis`: Uses Redis locks (requires `REDIS_*` config).
21+
- `CACHE_STORE=database`: Uses Database locks (requires `DB_CONNECTION` config).
22+
- Default: No lock provider (local only).
23+
24+
## Registering Tasks
25+
26+
You should register your scheduled tasks in your application's entry point (e.g., `main.go`) before executing the root command.
27+
28+
```go
29+
package main
30+
31+
import (
32+
"fmt"
33+
"github.com/pixelvide/laravel-go/pkg/schedule"
34+
"github.com/pixelvide/laravel-go/pkg/root"
35+
36+
// Import console to register commands
37+
_ "github.com/pixelvide/laravel-go/pkg/console"
38+
)
39+
40+
func main() {
41+
// 1. Simple Cron Job
42+
// Runs every minute
43+
schedule.Register("* * * * *", func() {
44+
fmt.Println("This runs every minute")
45+
})
46+
47+
// 2. Prevent Overlapping (Local)
48+
// If the task takes longer than 1 minute, the next run is skipped.
49+
schedule.Register("* * * * *", func() {
50+
// Heavy task...
51+
}, schedule.WithoutOverlapping())
52+
53+
// 3. Distributed Lock (OnOneServer)
54+
// Ensures the task runs on only ONE server in your cluster.
55+
// Requires CACHE_STORE to be configured (redis or database).
56+
schedule.Register("0 0 * * *", func() {
57+
fmt.Println("Daily Cleanup")
58+
}, schedule.OnOneServer("daily-cleanup"))
59+
60+
root.Execute()
61+
}
62+
```
63+
64+
## Database Locking
65+
66+
If you choose `CACHE_STORE=database`, the scheduler uses:
67+
- **MySQL**: `GET_LOCK(name, 0)` / `RELEASE_LOCK(name)`
68+
- **PostgreSQL**: `pg_try_advisory_lock(key)` / `pg_advisory_unlock(key)`
69+
70+
Ensure your database user has permission to use these locking functions.

‎go.mod‎

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -9,12 +9,21 @@ require (
99
github.com/aws/aws-sdk-go-v2 v1.41.0
1010
github.com/aws/aws-sdk-go-v2/config v1.32.6
1111
github.com/aws/aws-sdk-go-v2/service/sqs v1.42.20
12+
github.com/google/uuid v1.6.0
1213
github.com/redis/go-redis/v9 v9.17.2
1314
github.com/robfig/cron/v3 v3.0.1
15+
github.com/rs/zerolog v1.34.0
16+
github.com/spf13/cobra v1.10.2
17+
github.com/stretchr/testify v1.11.1
1418
github.com/yvasiyarov/php_session_decoder v0.0.0-20180803065642-a065a3b0b7d1
19+
go.opentelemetry.io/otel v1.39.0
20+
go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.39.0
21+
go.opentelemetry.io/otel/sdk v1.39.0
22+
go.opentelemetry.io/otel/trace v1.39.0
1523
)
1624

1725
require (
26+
filippo.io/edwards25519 v1.1.0 // indirect
1827
github.com/aws/aws-sdk-go-v2/credentials v1.19.6 // indirect
1928
github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16 // indirect
2029
github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.16 // indirect
@@ -27,22 +36,24 @@ require (
2736
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 // indirect
2837
github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 // indirect
2938
github.com/aws/smithy-go v1.24.0 // indirect
39+
github.com/bradfitz/gomemcache v0.0.0-20250403215159-8d39553ac7cf // indirect
40+
github.com/caarlos0/env/v11 v11.3.1 // indirect
3041
github.com/cespare/xxhash/v2 v2.3.0 // indirect
42+
github.com/davecgh/go-spew v1.1.1 // indirect
3143
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
3244
github.com/go-logr/logr v1.4.3 // indirect
3345
github.com/go-logr/stdr v1.2.2 // indirect
34-
github.com/google/uuid v1.6.0 // indirect
46+
github.com/go-sql-driver/mysql v1.9.3 // indirect
3547
github.com/inconshreveable/mousetrap v1.1.0 // indirect
48+
github.com/joho/godotenv v1.5.1 // indirect
49+
github.com/lib/pq v1.10.9 // indirect
3650
github.com/mattn/go-colorable v0.1.13 // indirect
3751
github.com/mattn/go-isatty v0.0.19 // indirect
38-
github.com/rs/zerolog v1.34.0 // indirect
39-
github.com/spf13/cobra v1.10.2 // indirect
52+
github.com/pmezard/go-difflib v1.0.0 // indirect
4053
github.com/spf13/pflag v1.0.9 // indirect
54+
github.com/stretchr/objx v0.5.2 // indirect
4155
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
42-
go.opentelemetry.io/otel v1.39.0 // indirect
43-
go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.39.0 // indirect
4456
go.opentelemetry.io/otel/metric v1.39.0 // indirect
45-
go.opentelemetry.io/otel/sdk v1.39.0 // indirect
46-
go.opentelemetry.io/otel/trace v1.39.0 // indirect
4757
golang.org/x/sys v0.39.0 // indirect
58+
gopkg.in/yaml.v3 v3.0.1 // indirect
4859
)

0 commit comments

Comments
 (0)