ai/live: Simplify payments API around readers

Author: j0sh

server/ai_http.go

@@ -214,11 +214,9 @@ func (h *lphttp) StartLiveVideoToVideo() http.Handler {
 					clog.Infof(ctx, "Error getting local trickle segment err=%v", err)
 					return
 				}
-				reader := segment.Reader
 				if paymentProcessor != nil {
-					reader = paymentProcessor.process(ctx, segment.Reader)
+					paymentProcessor.process(ctx, segment.Reader)
 				}
-				io.Copy(io.Discard, reader)
 			}
 		}()
 

server/ai_live_video.go

@@ -78,7 +78,7 @@ func startTricklePublish(ctx context.Context, url *url.URL, params aiRequestPara
 			defer slowOrchChecker.EndSegment()
 			var r io.Reader = reader
 			if paymentProcessor != nil {
-				r = paymentProcessor.process(ctx, reader)
+				paymentProcessor.process(ctx, reader)
 			}
 
 			clog.V(8).Infof(ctx, "trickle publish writing data seq=%d", seq)

server/live_payment_processor.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/livepeer/go-livepeer/clog"
+	"github.com/livepeer/go-livepeer/media"
 	"github.com/livepeer/lpms/ffmpeg"
 )
 
@@ -103,20 +104,13 @@ func (p *LivePaymentProcessor) processOne(ctx context.Context, timestamp time.Ti
 	p.lastProcessedAt = timestamp
 }
 
-func (p *LivePaymentProcessor) process(ctx context.Context, reader io.Reader) io.Reader {
+func (p *LivePaymentProcessor) process(ctx context.Context, reader media.CloneableReader) {
 	timestamp := time.Now()
 	if p.shouldSkip(timestamp) {
 		// We don't process every segment, because it's too compute-expensive
-		return reader
-	}
-
-	pipeReader, pipeWriter, err := os.Pipe()
-	if err != nil {
-		clog.InfofErr(ctx, "Error creating pipe", err)
-		return reader
+		return
 	}
 
-	resReader := io.TeeReader(reader, pipeWriter)
 	go func() {
 		select {
 		case p.processCh <- timestamp:
@@ -126,8 +120,7 @@ func (p *LivePaymentProcessor) process(ctx context.Context, reader io.Reader) io
 
 		// read the segment into the buffer, because the direct use of the reader causes Broken pipe
 		// it's probably related to different pace of reading by trickle and ffmpeg.GetCodecInfo()
-		defer pipeReader.Close()
-		segData, err := io.ReadAll(pipeReader)
+		segData, err := io.ReadAll(reader.Clone())
 		if err != nil {
 			clog.InfofErr(ctx, "Error reading segment data", err)
 			return
@@ -139,8 +132,6 @@ func (p *LivePaymentProcessor) process(ctx context.Context, reader io.Reader) io
 			// We process one segment at the time, no need to buffer them
 		}
 	}()
-
-	return resReader
 }
 
 func (p *LivePaymentProcessor) shouldSkip(timestamp time.Time) bool {

trickle/local_subscriber.go

@@ -2,16 +2,17 @@ package trickle
 
 import (
 	"errors"
-	"io"
 	"log/slog"
 	"strconv"
 	"sync"
+
+	"github.com/livepeer/go-livepeer/media"
 )
 
 // local (in-memory) subscriber for trickle protocol
 
 type TrickleData struct {
-	Reader   io.Reader
+	Reader   media.CloneableReader
 	Metadata map[string]string
 }
 
@@ -44,7 +45,8 @@ func (c *TrickleLocalSubscriber) Read() (*TrickleData, error) {
 		return nil, errors.New("seq not found")
 	}
 	c.seq++
-	r, w := io.Pipe()
+	w := media.NewMediaWriter()
+	r := w.MakeReader()
 	go func() {
 		subscriber := &SegmentSubscriber{
 			segment: segment,