Dmitry Sergeev
93 lines
1.6 KiB
Go
93 lines
1.6 KiB
Go
package playback
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
)
|
|
|
|
var ErrVideoWorkerRequired = errors.New("video worker is required")
|
|
|
|
type VideoSlot struct {
|
|
worker *VideoWorker
|
|
}
|
|
|
|
func NewVideoSlot(worker *VideoWorker) (*VideoSlot, error) {
|
|
if worker == nil {
|
|
return nil, ErrVideoWorkerRequired
|
|
}
|
|
return &VideoSlot{worker: worker}, nil
|
|
}
|
|
|
|
func (s *VideoSlot) Run(
|
|
ctx context.Context,
|
|
initial FeedConfig,
|
|
commands <-chan FeedConfig,
|
|
) error {
|
|
if err := initial.Validate(); err != nil {
|
|
return fmt.Errorf("validate initial video config: %w", err)
|
|
}
|
|
|
|
var (
|
|
workerCancel context.CancelFunc
|
|
workerDone chan error
|
|
)
|
|
|
|
start := func(config FeedConfig) {
|
|
workerCtx, cancel := context.WithCancel(ctx)
|
|
done := make(chan error, 1)
|
|
|
|
workerCancel = cancel
|
|
workerDone = done
|
|
|
|
go func() {
|
|
done <- s.worker.Run(workerCtx, config)
|
|
}()
|
|
}
|
|
|
|
stop := func() {
|
|
if workerCancel == nil {
|
|
return
|
|
}
|
|
|
|
workerCancel()
|
|
<-workerDone
|
|
|
|
workerCancel = nil
|
|
workerDone = nil
|
|
}
|
|
|
|
if initial.Active {
|
|
start(initial)
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
stop()
|
|
return ctx.Err()
|
|
|
|
case config, ok := <-commands:
|
|
if !ok {
|
|
stop()
|
|
return nil
|
|
}
|
|
if err := config.Validate(); err != nil {
|
|
// Ignore invalid commands without disturbing the current worker.
|
|
continue
|
|
}
|
|
stop()
|
|
if config.Active {
|
|
start(config)
|
|
}
|
|
|
|
case <-workerDone:
|
|
// The worker stopped naturally or exhausted its retries.
|
|
// Clear its lifecycle, but keep the slot alive for future commands.
|
|
workerCancel()
|
|
workerCancel = nil
|
|
workerDone = nil
|
|
}
|
|
}
|
|
}
|