Dmitry Sergeev
123 lines
2.1 KiB
Go
123 lines
2.1 KiB
Go
package playback
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
)
|
|
|
|
type SyncPairConfig struct {
|
|
Video FeedConfig
|
|
Audio FeedConfig
|
|
}
|
|
|
|
var (
|
|
ErrSyncWorkerRequired = errors.New("sync worker is required")
|
|
ErrSyncActivityMismatch = errors.New(
|
|
"synchronized video and audio must have matching active states",
|
|
)
|
|
)
|
|
|
|
func (c SyncPairConfig) Validate() error {
|
|
if err := c.Video.Validate(); err != nil {
|
|
return fmt.Errorf("video: %w", err)
|
|
}
|
|
if err := c.Audio.Validate(); err != nil {
|
|
return fmt.Errorf("audio: %w", err)
|
|
}
|
|
if c.Video.Active != c.Audio.Active {
|
|
return ErrSyncActivityMismatch
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c SyncPairConfig) Active() bool {
|
|
return c.Video.Active && c.Audio.Active
|
|
}
|
|
|
|
type SyncSlot struct {
|
|
worker *SyncWorker
|
|
}
|
|
|
|
func NewSyncSlot(worker *SyncWorker) (*SyncSlot, error) {
|
|
if worker == nil {
|
|
return nil, ErrSyncWorkerRequired
|
|
}
|
|
return &SyncSlot{
|
|
worker: worker,
|
|
}, nil
|
|
}
|
|
|
|
func (s *SyncSlot) Run(
|
|
ctx context.Context,
|
|
initial SyncPairConfig,
|
|
commands <-chan SyncPairConfig,
|
|
) error {
|
|
if err := initial.Validate(); err != nil {
|
|
return fmt.Errorf("validate initial sync config: %w", err)
|
|
}
|
|
|
|
var (
|
|
workerCancel context.CancelFunc
|
|
workerDone chan error
|
|
)
|
|
|
|
start := func(config SyncPairConfig) {
|
|
workerCtx, cancel := context.WithCancel(ctx)
|
|
done := make(chan error, 1)
|
|
|
|
workerCancel = cancel
|
|
workerDone = done
|
|
|
|
go func() {
|
|
done <- s.worker.Run(workerCtx, config.Video, config.Audio)
|
|
}()
|
|
}
|
|
|
|
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.
|
|
workerCancel()
|
|
workerCancel = nil
|
|
workerDone = nil
|
|
}
|
|
}
|
|
}
|