diff --git a/internal/source/source.go b/internal/source/source.go index 16a6db4..1be29f2 100644 --- a/internal/source/source.go +++ b/internal/source/source.go @@ -296,47 +296,98 @@ func OpenAudio(domain, flowID string) (*AudioSource, error) { }, nil } -func (s *AudioSource) NextAudio(ctx context.Context, batch uint64, timeout time.Duration) (AudioFrame, error) { - for { - select { - case <-ctx.Done(): - return AudioFrame{}, ctx.Err() - default: +func (s *AudioSource) ReadAudioOnceCtx( + ctx context.Context, + batch uint64, + timeout time.Duration, +) (AudioFrame, error) { + if err := ctx.Err(); err != nil { + return AudioFrame{}, err + } + + value, err := s.r.GetSamples(s.idx, int(batch), timeout) + if ctxErr := ctx.Err(); ctxErr != nil { + return AudioFrame{}, ctxErr + } + + switch { + case err == nil: + samples := make([][]byte, s.chans) + for channel := uint64(0); channel < s.chans; channel++ { + first, second, _ := value.ChannelFragments(channel) + if len(second) > 0 { + samples[channel] = append(first, second...) + } else { + samples[channel] = first + } } - v, err := s.r.GetSamples(s.idx, int(batch), timeout) - switch { - case err == nil: - samples := make([][]byte, s.chans) - for ch := uint64(0); ch < s.chans; ch++ { - f1, f2, _ := v.ChannelFragments(ch) - if len(f2) > 0 { - samples[ch] = append(f1, f2...) - } else { - samples[ch] = f1 + + frame := AudioFrame{ + Index: s.idx, + SampleCount: batch, + Channels: s.chans, + Samples: samples, + } + s.idx += batch + return frame, nil + case errors.Is(err, mxl.ErrOutOfRangeEarly): + return AudioFrame{}, wrapError( + "read audio", + ErrorKindTemporary, + err, + ) + case errors.Is(err, mxl.ErrOutOfRangeLate): + runtimeInfo, runtimeErr := s.r.Runtime() + if runtimeErr != nil { + return AudioFrame{}, wrapError( + "read audio runtime", + ErrorKindUnavailable, + runtimeErr, + ) + } + s.idx = runtimeInfo.HeadIndex + return AudioFrame{}, wrapError( + "read audio", + ErrorKindTemporary, + err, + ) + default: + return AudioFrame{}, wrapError( + "read audio", + ErrorKindUnavailable, + err, + ) + } +} + +func (s *AudioSource) NextAudio( + ctx context.Context, + batch uint64, + timeout time.Duration, +) (AudioFrame, error) { + for { + frame, err := s.ReadAudioOnceCtx(ctx, batch, timeout) + if err == nil { + return frame, nil + } + if ctx.Err() != nil { + return AudioFrame{}, ctx.Err() + } + if KindOf(err) != ErrorKindTemporary { + return AudioFrame{}, err + } + + timer := time.NewTimer(10 * time.Millisecond) + select { + case <-timer.C: + case <-ctx.Done(): + if !timer.Stop() { + select { + case <-timer.C: + default: } } - f := AudioFrame{ - Index: s.idx, - SampleCount: batch, - Channels: s.chans, - Samples: samples, - } - s.idx += batch - return f, nil - case errors.Is(err, mxl.ErrOutOfRangeEarly): - select { - case <-time.After(10 * time.Millisecond): - case <-ctx.Done(): - return AudioFrame{}, ctx.Err() - } - case errors.Is(err, mxl.ErrOutOfRangeLate): - rt, rerr := s.r.Runtime() - if rerr != nil { - return AudioFrame{}, fmt.Errorf("Runtime: %w", rerr) - } - s.idx = rt.HeadIndex - default: - return AudioFrame{}, fmt.Errorf("GetSamples: %w", err) + return AudioFrame{}, ctx.Err() } } }