me/neoq
1
0
Fork 0
mirror of https://github.com/acaloiaro/neoq synced 2026-07-21 18:29:08 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix job chaining sync issue

Browse source
This commit is contained in:
Pranav Shridhar 2024-04-02 17:10:51 +05:30
parent 5e0dcb05a5
commit e2d5d0448d
1 changed files with 16 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
backends/sqlite/sqlite_backend.go
View file

@ -80,16 +80,16 @@ func Backend(ctx context.Context, opts ...neoq.ConfigOption) (sb neoq.Neoq, err
cfg.IdleTransactionTimeout = neoq.DefaultIdleTxTimeout
s := &SqliteBackend{
cancelFuncs: []context.CancelFunc{},
config: cfg,
cron: cron.New(),
futureJobs: make(map[string]*jobs.Job),
handlers: make(map[string]handler.Handler),
newQueues: make(chan string),
readyQueues: make(chan string),
// queueListenerChan: make(map[string]chan string),
mu: &sync.RWMutex{},
mu2: &sync.RWMutex{},
cancelFuncs: []context.CancelFunc{},
config: cfg,
cron: cron.New(),
futureJobs: make(map[string]*jobs.Job),
handlers: make(map[string]handler.Handler),
newQueues: make(chan string),
readyQueues: make(chan string),
queueListenerChan: make(map[string]chan string),
mu: &sync.RWMutex{},
mu2: &sync.RWMutex{},
}
// Set all options
@ -226,6 +226,7 @@ func (s *SqliteBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID strin
s.mu.Unlock()
log.Println("PS::Enqueue released lock", jobID)
log.Println("PS::before send to channel")
// PS::send to channel
// go func() {
@ -247,6 +248,8 @@ func (s *SqliteBackend) Enqueue(ctx context.Context, job *jobs.Job) (jobID strin
// )
// }
fmt.Println("PS::after send to channel")
return jobID, nil
}
@ -255,14 +258,13 @@ func (s *SqliteBackend) Start(ctx context.Context, h handler.Handler) (err error
ctx, cancel := context.WithCancel(ctx)
s.logger.Debug("starting job processing", slog.String("queue", h.Queue))
s.mu.Lock()
s.mu2.Lock()
s.cancelFuncs = append(s.cancelFuncs, cancel)
s.handlers[h.Queue] = h
s.mu.Unlock()
s.mu2.Unlock()
s.newQueues <- h.Queue
s.queueListenerChan = make(map[string]chan string, 1000)
s.queueListenerChan[h.Queue] = make(chan string) // PS
s.queueListenerChan[h.Queue] = make(chan string, 1000) // PS
err = s.start(ctx, h)
if err != nil {