generated from maragudk/template
-
-
Notifications
You must be signed in to change notification settings - Fork 13
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
This adds a new package `jobs` which provides an abstraction to run background jobs, backed by the queue.
- Loading branch information
Showing
10 changed files
with
676 additions
and
55 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "database/sql" | ||
| "fmt" | ||
| "log/slog" | ||
| "time" | ||
|
|
||
| _ "github.com/mattn/go-sqlite3" | ||
|
|
||
| "github.com/maragudk/goqite" | ||
| "github.com/maragudk/goqite/jobs" | ||
| ) | ||
|
|
||
| func main() { | ||
| log := slog.Default() | ||
|
|
||
| // Setup the db and goqite schema. | ||
| db, err := sql.Open("sqlite3", ":memory:?_journal=WAL&_timeout=5000&_fk=true") | ||
| if err != nil { | ||
| log.Info("Error opening db", "error", err) | ||
| } | ||
| db.SetMaxOpenConns(1) | ||
| db.SetMaxIdleConns(1) | ||
|
|
||
| if err := goqite.Setup(context.Background(), db); err != nil { | ||
| log.Info("Error in setup", "error", err) | ||
| } | ||
|
|
||
| // Make a new queue for the jobs. You can have as many of these as you like, just name them differently. | ||
| q := goqite.New(goqite.NewOpts{ | ||
| DB: db, | ||
| Name: "jobs", | ||
| }) | ||
|
|
||
| // Make a job runner with a job limit of 1 and a short message poll interval. | ||
| r := jobs.NewRunner(jobs.NewRunnerOpts{ | ||
| Limit: 1, | ||
| Log: slog.Default(), | ||
| PollInterval: 10 * time.Millisecond, | ||
| Queue: q, | ||
| }) | ||
|
|
||
| // Register our "print" job. | ||
| r.Register("print", func(ctx context.Context, m []byte) error { | ||
| fmt.Println(string(m)) | ||
| return nil | ||
| }) | ||
|
|
||
| // Create a "print" job with a message. | ||
| if err := jobs.Create(context.Background(), q, "print", []byte("Yo")); err != nil { | ||
| log.Info("Error creating job", "error", err) | ||
| } | ||
|
|
||
| // Stop the job runner after a timeout. | ||
| ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) | ||
| defer cancel() | ||
|
|
||
| // Start the job runner and see the job run. | ||
| r.Start(ctx) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| package sql | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "fmt" | ||
| ) | ||
|
|
||
| func InTx(db *sql.DB, cb func(*sql.Tx) error) (err error) { | ||
| tx, txErr := db.Begin() | ||
| if txErr != nil { | ||
| return fmt.Errorf("cannot start tx: %w", txErr) | ||
| } | ||
|
|
||
| defer func() { | ||
| if rec := recover(); rec != nil { | ||
| err = rollback(tx, nil) | ||
| panic(rec) | ||
| } | ||
| }() | ||
|
|
||
| if err := cb(tx); err != nil { | ||
| return rollback(tx, err) | ||
| } | ||
|
|
||
| if txErr := tx.Commit(); txErr != nil { | ||
| return fmt.Errorf("cannot commit tx: %w", txErr) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func rollback(tx *sql.Tx, err error) error { | ||
| if txErr := tx.Rollback(); txErr != nil { | ||
| return fmt.Errorf("cannot roll back tx after error (tx error: %v), original error: %w", txErr, err) | ||
| } | ||
| return err | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| create table goqite ( | ||
| id text primary key default ('m_' || lower(hex(randomblob(16)))), | ||
| created text not null default (strftime('%Y-%m-%dT%H:%M:%fZ')), | ||
| updated text not null default (strftime('%Y-%m-%dT%H:%M:%fZ')), | ||
| queue text not null, | ||
| body blob not null, | ||
| timeout text not null default (strftime('%Y-%m-%dT%H:%M:%fZ')), | ||
| received integer not null default 0 | ||
| ) strict; | ||
|
|
||
| create trigger goqite_updated_timestamp after update on goqite begin | ||
| update goqite set updated = strftime('%Y-%m-%dT%H:%M:%fZ') where id = old.id; | ||
| end; | ||
|
|
||
| create index goqite_queue_created_idx on goqite (queue, created); |
Oops, something went wrong.