mirror of
https://github.com/acaloiaro/neoq
synced 2026-07-21 18:29:08 +00:00
feat: accept a database transaction when queueing jobs
This commit is contained in:
Adriano Caloiaro
parent
0cab9ce7dc
commit
0697e621c3
10 changed files with 209 additions and 27 deletions
2
.helix/languages.toml
Normal file
2
.helix/languages.toml
Normal file
|
|
@ -0,0 +1,2 @@
|
|||
[language-server.gopls.config]
|
||||
buildFlags = ["-tags=testing"]
|
||||
|
|
@ -66,6 +66,11 @@ func Backend(_ context.Context, opts ...neoq.ConfigOption) (backend neoq.Neoq, e
|
|||
return
|
||||
}
|
||||
|
||||
// Enqueue adds jobs to the specified queue
|
||||
func (p *MemBackend) EnqueueTx(ctx context.Context, tx neoq.Tx, job *jobs.Job) (jobID string, err error) {
|
||||
panic("this backend does not implement enqueueing jobs transactionally")
|
||||
}
|
||||
|
||||
// Enqueue queues jobs to be executed asynchronously
|
||||
func (m *MemBackend) Enqueue(_ context.Context, job *jobs.Job) (jobID string, err error) {
|
||||
var queueChan chan *jobs.Job
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
nqpgxtx "github.com/acaloiaro/neoq/backends/postgres/tx/pgx"
|
||||
"github.com/acaloiaro/neoq/handler"
|
||||
"github.com/acaloiaro/neoq/internal"
|
||||
"github.com/acaloiaro/neoq/jobs"
|
||||
|
|
@ -400,8 +401,18 @@ func (p *PgBackend) initializeDB() (err error) {
|
|||
return nil
|
||||
}
|
||||
|
||||
// Enqueue adds jobs to the specified queue using the provided database transaction
|
||||
func (p *PgBackend) EnqueueTx(ctx context.Context, tx neoq.Tx, job *jobs.Job) (jobID string, err error) {
|
||||
return p.enqueue(ctx, tx, job)
|
||||
}
|
||||
|
||||
// Enqueue adds jobs to the specified queue
|
||||
func (p *PgBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID string, err error) {
|
||||
return p.enqueue(ctx, nil, job)
|
||||
}
|
||||
|
||||
// Enqueue adds jobs to the specified queue
|
||||
func (p *PgBackend) enqueue(ctx context.Context, tx neoq.Tx, job *jobs.Job) (jobID string, err error) {
|
||||
if job.Queue == "" {
|
||||
err = jobs.ErrNoQueueSpecified
|
||||
return
|
||||
|
|
@ -417,11 +428,15 @@ func (p *PgBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID string, e
|
|||
}
|
||||
defer conn.Release()
|
||||
|
||||
p.logger.Debug("beginning new transaction to enqueue job", slog.String("queue", job.Queue))
|
||||
tx, err := conn.Begin(ctx)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error creating transaction: %w", err)
|
||||
return
|
||||
if tx == nil {
|
||||
p.logger.Debug("beginning new transaction to enqueue job", slog.String("queue", job.Queue))
|
||||
pgxtx, err := p.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("error creating transaction: %w", err)
|
||||
return "", err
|
||||
}
|
||||
|
||||
tx = nqpgxtx.FromPgx(pgxtx)
|
||||
}
|
||||
|
||||
// Rollback is safe to call even if the tx is already closed, so if
|
||||
|
|
@ -580,7 +595,7 @@ func (p *PgBackend) Shutdown(ctx context.Context) {
|
|||
//
|
||||
// Jobs that are not already fingerprinted are fingerprinted before being added
|
||||
// Duplicate jobs are not added to the queue. Any two unprocessed jobs with the same fingerprint are duplicates
|
||||
func (p *PgBackend) enqueueJob(ctx context.Context, tx pgx.Tx, j *jobs.Job) (jobID string, err error) {
|
||||
func (p *PgBackend) enqueueJob(ctx context.Context, tx neoq.Tx, j *jobs.Job) (jobID string, err error) {
|
||||
err = jobs.FingerprintJob(j)
|
||||
if err != nil {
|
||||
return
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import (
|
|||
|
||||
"github.com/acaloiaro/neoq"
|
||||
"github.com/acaloiaro/neoq/backends/postgres"
|
||||
nqpgxtx "github.com/acaloiaro/neoq/backends/postgres/tx/pgx"
|
||||
"github.com/acaloiaro/neoq/handler"
|
||||
"github.com/acaloiaro/neoq/internal"
|
||||
"github.com/acaloiaro/neoq/jobs"
|
||||
|
|
@ -152,6 +153,84 @@ func TestBasicJobProcessing(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// TestBasicJobProcessingWithTransaction tests that the postgres backend is able to process the most basic jobs with the
|
||||
// most basic configuration, inside a transaction.
|
||||
func TestBasicJobProcessingWithTransaction(t *testing.T) {
|
||||
connString, conn := prepareAndCleanupDB(t)
|
||||
const queue = "testing"
|
||||
maxRetries := 5
|
||||
done := make(chan bool)
|
||||
defer close(done)
|
||||
|
||||
timeoutTimer := time.After(5 * time.Second)
|
||||
|
||||
ctx := context.Background()
|
||||
nq, err := neoq.New(ctx, neoq.WithBackend(postgres.Backend), postgres.WithConnectionString(connString))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer nq.Shutdown(ctx)
|
||||
|
||||
h := handler.New(queue, func(_ context.Context) (err error) {
|
||||
done <- true
|
||||
return
|
||||
})
|
||||
|
||||
err = nq.Start(ctx, h)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
ptx, err := conn.Begin(ctx)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
tx := nqpgxtx.FromPgx(ptx)
|
||||
deadline := time.Now().UTC().Add(5 * time.Second)
|
||||
jid, e := nq.EnqueueTx(ctx, tx, &jobs.Job{
|
||||
Queue: queue,
|
||||
Payload: map[string]any{
|
||||
"message": "hello world",
|
||||
},
|
||||
Deadline: &deadline,
|
||||
MaxRetries: &maxRetries,
|
||||
})
|
||||
if e != nil || jid == jobs.DuplicateJobID {
|
||||
t.Error(e)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-timeoutTimer:
|
||||
err = jobs.ErrJobTimeout
|
||||
case <-done:
|
||||
}
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
// ensure job has fields set correctly
|
||||
var jdl time.Time
|
||||
var jmxrt int
|
||||
|
||||
err = conn.
|
||||
QueryRow(context.Background(), "SELECT deadline,max_retries FROM neoq_jobs WHERE id = $1", jid).
|
||||
Scan(&jdl, &jmxrt)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
jdl = jdl.In(time.UTC)
|
||||
// dates from postgres come out with only 6 decimal places of millisecond precision, naively format dates as
|
||||
// strings for comparison reasons. Ref https://www.postgresql.org/docs/current/datatype-datetime.html
|
||||
if jdl.Format(time.RFC3339) != deadline.Format(time.RFC3339) {
|
||||
t.Error(fmt.Errorf("job deadline does not match its expected value: %v != %v", jdl, deadline)) // nolint: goerr113
|
||||
}
|
||||
|
||||
if jmxrt != maxRetries {
|
||||
t.Error(fmt.Errorf("job MaxRetries does not match its expected value: %v != %v", jmxrt, maxRetries)) // nolint: goerr113
|
||||
}
|
||||
}
|
||||
func TestMultipleProcessors(t *testing.T) {
|
||||
const queue = "testing"
|
||||
var execCount uint32
|
||||
|
|
|
|||
14
backends/postgres/tx/pgx/tx.go
Normal file
14
backends/postgres/tx/pgx/tx.go
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
package pgx
|
||||
|
||||
import (
|
||||
"github.com/acaloiaro/neoq"
|
||||
nqptx "github.com/acaloiaro/neoq/backends/postgres/tx"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// FromStdlib creates a neoq DbTx from a standard library sql.Tx
|
||||
func FromPgx(tx pgx.Tx) neoq.Tx {
|
||||
return nqptx.TxWrapper{
|
||||
Tx: tx,
|
||||
}
|
||||
}
|
||||
35
backends/postgres/tx/tx.go
Normal file
35
backends/postgres/tx/tx.go
Normal file
|
|
@ -0,0 +1,35 @@
|
|||
package tx
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/acaloiaro/neoq"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// TxWrapper wraps a standard library sql.Tx in the neoq Tx interface so that sql.Tx
|
||||
// may be used as a transaction type for enqueueing jobs.
|
||||
//
|
||||
// sql.Tx is _nearly_ compatible with pgx's Tx type, after which neoq's Tx is modeled.
|
||||
type TxWrapper struct {
|
||||
Tx pgx.Tx
|
||||
}
|
||||
|
||||
func (t TxWrapper) Commit(ctx context.Context) (err error) {
|
||||
return t.Tx.Commit(ctx)
|
||||
}
|
||||
|
||||
func (t TxWrapper) Rollback(ctx context.Context) (err error) {
|
||||
return t.Tx.Rollback(ctx)
|
||||
}
|
||||
|
||||
func (t TxWrapper) QueryRow(ctx context.Context, query string, args ...any) neoq.Row {
|
||||
return t.Tx.QueryRow(ctx, query, args)
|
||||
}
|
||||
|
||||
// FromStdlib creates a neoq DbTx from a standard library sql.Tx
|
||||
func FromPgx(tx pgx.Tx) (t neoq.Tx) {
|
||||
return &TxWrapper{
|
||||
Tx: tx,
|
||||
}
|
||||
}
|
||||
|
|
@ -178,6 +178,11 @@ func WithShutdownTimeout(timeout time.Duration) neoq.ConfigOption {
|
|||
}
|
||||
}
|
||||
|
||||
// Enqueue adds jobs to the specified queue
|
||||
func (p *RedisBackend) EnqueueTx(ctx context.Context, tx neoq.Tx, job *jobs.Job) (jobID string, err error) {
|
||||
panic("this backend does not implement enqueueing jobs transactionally")
|
||||
}
|
||||
|
||||
// Enqueue queues jobs to be executed asynchronously
|
||||
func (b *RedisBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID string, err error) {
|
||||
if job.Queue == "" {
|
||||
|
|
|
|||
14
go.mod
14
go.mod
|
|
@ -10,7 +10,7 @@ require (
|
|||
github.com/hibiken/asynq v0.24.0
|
||||
github.com/iancoleman/strcase v0.2.0
|
||||
github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa
|
||||
github.com/jackc/pgx/v5 v5.6.0
|
||||
github.com/jackc/pgx/v5 v5.7.6
|
||||
github.com/jsuar/go-cron-descriptor v0.1.0
|
||||
github.com/pkg/errors v0.9.1
|
||||
github.com/robfig/cron v1.2.0
|
||||
|
|
@ -25,18 +25,18 @@ require (
|
|||
github.com/hashicorp/errwrap v1.1.0 // indirect
|
||||
github.com/hashicorp/go-multierror v1.1.1 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.1 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.2 // indirect
|
||||
github.com/lib/pq v1.10.2 // indirect
|
||||
github.com/robfig/cron/v3 v3.0.1 // indirect
|
||||
github.com/spf13/cast v1.3.1 // indirect
|
||||
go.uber.org/atomic v1.10.0 // indirect
|
||||
go.uber.org/multierr v1.9.0 // indirect
|
||||
go.uber.org/zap v1.24.0 // indirect
|
||||
golang.org/x/crypto v0.35.0 // indirect
|
||||
golang.org/x/sync v0.11.0 // indirect
|
||||
golang.org/x/sys v0.30.0 // indirect
|
||||
golang.org/x/text v0.22.0 // indirect
|
||||
golang.org/x/crypto v0.37.0 // indirect
|
||||
golang.org/x/sync v0.13.0 // indirect
|
||||
golang.org/x/sys v0.32.0 // indirect
|
||||
golang.org/x/text v0.24.0 // indirect
|
||||
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 // indirect
|
||||
google.golang.org/protobuf v1.33.0 // indirect
|
||||
)
|
||||
|
|
|
|||
28
go.sum
28
go.sum
|
|
@ -80,12 +80,12 @@ github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa h1:s+4MhCQ6YrzisK6
|
|||
github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
|
||||
github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
|
||||
github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
|
||||
github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
|
||||
github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
|
||||
github.com/jackc/pgx/v5 v5.7.6 h1:rWQc5FwZSPX58r1OQmkuaNicxdmExaEz5A2DO2hUuTk=
|
||||
github.com/jackc/pgx/v5 v5.7.6/go.mod h1:aruU7o91Tc2q2cFp5h4uP3f6ztExVpyVv88Xl/8Vl8M=
|
||||
github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
|
||||
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
|
||||
github.com/jsuar/go-cron-descriptor v0.1.0 h1:Q97ujk+/xhcz1lA9nmUMq750FZV7RlV23TNyMB74xkQ=
|
||||
github.com/jsuar/go-cron-descriptor v0.1.0/go.mod h1:PFR+Y6Lr86uYZpwsWRoFMRA3CX4a6q7zfPltp3SLkSU=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
|
|
@ -154,8 +154,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
|
|||
golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs=
|
||||
golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ=
|
||||
golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
|
||||
golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
|
||||
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
|
||||
golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
|
||||
golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
|
||||
|
|
@ -185,8 +185,8 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ
|
|||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
|
||||
golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
|
||||
golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610=
|
||||
golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
|
||||
golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
|
|
@ -200,13 +200,13 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
|
|||
golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
|
||||
golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
|
||||
golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
|
||||
golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
|
||||
golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0=
|
||||
golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU=
|
||||
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ=
|
||||
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
|
|
|
|||
27
neoq.go
27
neoq.go
|
|
@ -72,6 +72,9 @@ type Neoq interface {
|
|||
// Enqueue queues jobs to be executed asynchronously
|
||||
Enqueue(ctx context.Context, job *jobs.Job) (jobID string, err error)
|
||||
|
||||
// Enqueue queues jobs to be executed asynchronously
|
||||
EnqueueTx(ctx context.Context, tx Tx, job *jobs.Job) (jobID string, err error)
|
||||
|
||||
// Start starts processing jobs on the queue specified in the Handler
|
||||
Start(ctx context.Context, h handler.Handler) (err error)
|
||||
|
||||
|
|
@ -87,6 +90,30 @@ type Neoq interface {
|
|||
Shutdown(ctx context.Context)
|
||||
}
|
||||
|
||||
// Tx is neoq's generic transaction interface.
|
||||
//
|
||||
// This interface's purpose is to enable interoperability with external library
|
||||
// transaction types. By being a small (but common) subset of the ususal transaction
|
||||
// type methods, this interface allows transactions from multiple libraries to become
|
||||
// become compatible with Tx.
|
||||
//
|
||||
// This interface is compatible with the following libraries' transaction types:
|
||||
// - pgx: This interface is a subset of pgx's Tx type, so it is fully compatible
|
||||
// - sql: Use helper function tx.FromStdTx to get a Tx from sql.Tx
|
||||
type Tx interface {
|
||||
Commit(ctx context.Context) error
|
||||
Rollback(ctx context.Context) error
|
||||
QueryRow(ctx context.Context, query string, args ...any) Row
|
||||
}
|
||||
|
||||
// Row interface is the subset of both pgx.Row and sql.Row needed by neoq to enqueue
|
||||
// jobs useing either's Tx type.
|
||||
//
|
||||
// It is the return type of neoq.Tx.QueryRow, to be used as a generic row type
|
||||
type Row interface {
|
||||
Scan(dest ...any) error
|
||||
}
|
||||
|
||||
// New creates a new backend instance for job processing.
|
||||
//
|
||||
// By default, neoq initializes [memory.Backend] if New() is called without a backend configuration option.
|
||||
|
|
|
|||
Loading…
Reference in a new issue