Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
6.0 kB · 220 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221package media
import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "sort"
"golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/aqio" c2patypes "stream.place/streamplace/pkg/c2patypes" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/iroh/generated/iroh_streamplace" "stream.place/streamplace/pkg/log")
type SplitSegment struct { Filename string Data []byte}
type ManifestResult struct { Manifests map[string]c2patypes.Manifest `json:"manifests"` Certs map[string]string `json:"certs"`}
type ManifestAndMetadata struct { Manifest c2patypes.Manifest SegmentMetadata *SegmentMetadata}
type ReadWriteSeekCloser interface { io.ReadWriteSeeker io.Closer}
type SegmentToSign struct { unsignedSeg io.ReadSeeker manifestId string cert []byte outputSeg io.ReadWriteSeeker closeCh chan struct{}}
func NewSegmentToSign(unsignedSeg io.ReadSeeker, manifestId string, cert []byte, outputSeg io.ReadWriteSeeker) *SegmentToSign { return &SegmentToSign{ unsignedSeg: unsignedSeg, manifestId: manifestId, cert: cert, outputSeg: outputSeg, closeCh: make(chan struct{}), }}
func (s *SegmentToSign) UnsignedSegStream() iroh_streamplace.Stream { return c2patypes.NewReader(s.unsignedSeg)}
func (s *SegmentToSign) ManifestId() string { return s.manifestId}
func (s *SegmentToSign) Cert() []byte { return s.cert}
func (s *SegmentToSign) OutputSegStream() iroh_streamplace.Stream { return c2patypes.NewWriter(s.outputSeg)}
func (s *SegmentToSign) Close() { close(s.closeCh)}
func (s *SegmentToSign) Done() { <-s.closeCh}
type ManySegmentsToSign struct { segmentCh chan iroh_streamplace.SegmentToSign readyCh chan struct{}}
func (m *ManySegmentsToSign) Next() *iroh_streamplace.SegmentToSign { if m.readyCh != nil { close(m.readyCh) m.readyCh = nil } seg := <-m.segmentCh if seg == nil { return nil } return &seg}
// split a signed concatenated mp4 into its constituent signed segmentsfunc SplitSegments(ctx context.Context, cli *config.CLI, input io.ReadSeeker, cb func(fname string) ReadWriteSeekCloser) error { manifestsStr, err := iroh_streamplace.GetManifests(c2patypes.NewReader(input)) if err != nil { return fmt.Errorf("failed to get manifests: %w", err) } _, err = input.Seek(0, io.SeekStart) if err != nil { return fmt.Errorf("failed to seek to start: %w", err) } var manifests ManifestResult err = json.Unmarshal([]byte(manifestsStr), &manifests) if err != nil { return fmt.Errorf("failed to unmarshal manifests: %w", err) } manifestList := []ManifestAndMetadata{} for _, manifest := range manifests.Manifests { metadata, err := ParseSegmentAssertions(ctx, &manifest) if errors.Is(err, ErrMissingMetadata) { log.Error(ctx, "missing metadata", "manifest", manifest.Label) continue } if err != nil { return fmt.Errorf("failed to parse segment assertions: %w", err) } manifestList = append(manifestList, ManifestAndMetadata{ Manifest: manifest, SegmentMetadata: metadata, }) } if len(manifestList) == 0 { return fmt.Errorf("no manifests found") } sort.Slice(manifestList, func(i, j int) bool { m1 := manifestList[i] m2 := manifestList[j] return m1.SegmentMetadata.StartTime.Time().Before(m2.SegmentMetadata.StartTime.Time()) }) certList := [][]byte{} for _, manifest := range manifestList { certList = append(certList, []byte(manifests.Certs[*manifest.Manifest.Label])) }
segmentCh := make(chan iroh_streamplace.SegmentToSign) readyCh := make(chan struct{}) mss := &ManySegmentsToSign{ segmentCh: segmentCh, readyCh: readyCh, } g, ctx := errgroup.WithContext(ctx) unsignedCh := make(chan *SplitSegment) streamer := manifestList[0].SegmentMetadata.Creator
// note: we're passing the input to two places here and need to make sure // they're not running into problems with concurrent seeking. so we use // this readyCh as a hack - it only fires after Rust is done with the input
g.Go(func() error { err := iroh_streamplace.Resign(mss, c2patypes.NewReader(input)) if err != nil { return fmt.Errorf("failed to resign segments: %w", err) } return nil }) g.Go(func() error { defer close(unsignedCh) <-readyCh // rust is done with the input, rewind and start segmenting _, err := input.Seek(0, io.SeekStart) if err != nil { return fmt.Errorf("failed to seek to start: %w", err) } err = SegmentUnsigned(ctx, cli, streamer, input, true, unsignedCh) if err != nil { return fmt.Errorf("failed to segment file: %w", err) } return nil }) validationErrors := []error{} g.Go(func() error { defer close(segmentCh) i := 0 for unsignedSeg := range unsignedCh { meta := manifestList[i].SegmentMetadata fname := fmt.Sprintf("%s.mp4", meta.StartTime.FileSafeString()) rwsc := cb(fname) rws := aqio.NewReadWriteSeeker(unsignedSeg.Data) ss := NewSegmentToSign(c2patypes.NewReader(rws), *manifestList[i].Manifest.Label, certList[i], rwsc) i += 1 segmentCh <- ss ss.Done() _, err := rwsc.Seek(0, io.SeekStart) if err != nil { return fmt.Errorf("failed to seek to start: %w", err) } bs, err := io.ReadAll(rwsc) if err != nil { return fmt.Errorf("failed to read segment file: %w", err) } _, validationError := ValidateMP4Media(ctx, bs) if validationError != nil { validationErrors = append(validationErrors, validationError) cli.DumpDebugSegment(ctx, fmt.Sprintf("%s-invalid.mp4", fname), bytes.NewReader(bs)) } log.Log(ctx, "validated segment file", "path", fname) err = rwsc.Close() if err != nil { return fmt.Errorf("failed to close segment file: %w", err) } } return nil })
err = g.Wait() if err != nil { return fmt.Errorf("failed to split segments: %w", err) } if len(validationErrors) > 0 { return fmt.Errorf("%d errors validating segments; first error: %w", len(validationErrors), validationErrors[0]) } return nil}