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
 	}
-		v, err := s.r.GetSamples(s.idx, int(batch), timeout)
+
+	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 ch := uint64(0); ch < s.chans; ch++ {
-				f1, f2, _ := v.ChannelFragments(ch)
-				if len(f2) > 0 {
-					samples[ch] = append(f1, f2...)
+		for channel := uint64(0); channel < s.chans; channel++ {
+			first, second, _ := value.ChannelFragments(channel)
+			if len(second) > 0 {
+				samples[channel] = append(first, second...)
 			} else {
-					samples[ch] = f1
+				samples[channel] = first
 			}
 		}
-			f := AudioFrame{
+
+		frame := AudioFrame{
 			Index:       s.idx,
 			SampleCount: batch,
 			Channels:    s.chans,
 			Samples:     samples,
 		}
 		s.idx += batch
-			return f, nil
+		return frame, nil
 	case errors.Is(err, mxl.ErrOutOfRangeEarly):
-			select {
-			case <-time.After(10 * time.Millisecond):
-			case <-ctx.Done():
+		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()
 		}
-		case errors.Is(err, mxl.ErrOutOfRangeLate):
-			rt, rerr := s.r.Runtime()
-			if rerr != nil {
-				return AudioFrame{}, fmt.Errorf("Runtime: %w", rerr)
+		if KindOf(err) != ErrorKindTemporary {
+			return AudioFrame{}, err
 		}
-			s.idx = rt.HeadIndex
+
+		timer := time.NewTimer(10 * time.Millisecond)
+		select {
+		case <-timer.C:
+		case <-ctx.Done():
+			if !timer.Stop() {
+				select {
+				case <-timer.C:
 				default:
-			return AudioFrame{}, fmt.Errorf("GetSamples: %w", err)
+				}
+			}
+			return AudioFrame{}, ctx.Err()
 		}
 	}
 }