prevent completed download restart
This commit is contained in:
@@ -65,7 +65,9 @@ func (m *MessageQueue) downloadConsumer() {
|
||||
slog.String("id", p.getShortId()),
|
||||
)
|
||||
|
||||
if p.Progress.Status != StatusCompleted {
|
||||
p.Start()
|
||||
}
|
||||
|
||||
m.logger.Info("started process",
|
||||
slog.String("bus", queueName),
|
||||
@@ -92,6 +94,14 @@ func (m *MessageQueue) metadataSubscriber() {
|
||||
slog.String("id", p.getShortId()),
|
||||
)
|
||||
|
||||
if p.Progress.Status == StatusCompleted {
|
||||
m.logger.Warn("proccess has an illegal state",
|
||||
slog.String("id", p.getShortId()),
|
||||
slog.Int("status", p.Progress.Status),
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
if err := p.SetMetadata(); err != nil {
|
||||
m.logger.Error("failed to retrieve metadata",
|
||||
slog.String("id", p.getShortId()),
|
||||
|
||||
@@ -198,6 +198,9 @@ func (p *Process) Complete() {
|
||||
|
||||
// Kill a process and remove it from the memory
|
||||
func (p *Process) Kill() error {
|
||||
defer func() {
|
||||
p.Progress.Status = StatusCompleted
|
||||
}()
|
||||
// yt-dlp uses multiple child process the parent process
|
||||
// has been spawned with setPgid = true. To properly kill
|
||||
// all subprocesses a SIGTERM need to be sent to the correct
|
||||
|
||||
@@ -2,7 +2,6 @@ package rpc
|
||||
|
||||
import (
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
@@ -30,8 +29,6 @@ func WebSocket(w http.ResponseWriter, r *http.Request) {
|
||||
for {
|
||||
mtype, reader, err := c.NextReader()
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
log.Println(err)
|
||||
break
|
||||
}
|
||||
|
||||
@@ -40,7 +37,6 @@ func WebSocket(w http.ResponseWriter, r *http.Request) {
|
||||
writer, err := c.NextWriter(mtype)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
log.Println(err)
|
||||
break
|
||||
}
|
||||
|
||||
|
||||
@@ -122,7 +122,10 @@ func (s *Service) KillAll(args NoArgs, killed *string) error {
|
||||
|
||||
var (
|
||||
keys = s.db.Keys()
|
||||
err error
|
||||
removeFunc = func(p *internal.Process) error {
|
||||
defer s.db.Delete(p.Id)
|
||||
return p.Kill()
|
||||
}
|
||||
)
|
||||
|
||||
for _, key := range *keys {
|
||||
@@ -131,22 +134,24 @@ func (s *Service) KillAll(args NoArgs, killed *string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
if proc != nil {
|
||||
err := proc.Kill()
|
||||
if err != nil {
|
||||
if proc == nil {
|
||||
s.db.Delete(key)
|
||||
continue
|
||||
}
|
||||
|
||||
if err := removeFunc(proc); err != nil {
|
||||
s.logger.Info(
|
||||
"failed killing process",
|
||||
slog.String("id", proc.Id),
|
||||
slog.String("err", err.Error()),
|
||||
slog.Any("err", err),
|
||||
)
|
||||
return err
|
||||
continue
|
||||
}
|
||||
|
||||
s.logger.Info("succesfully killed process", slog.String("id", proc.Id))
|
||||
}
|
||||
}
|
||||
|
||||
return err
|
||||
return nil
|
||||
}
|
||||
|
||||
// Remove a process from the db rendering it unusable if active
|
||||
|
||||
Reference in New Issue
Block a user