streaming

commit 3ddfbb7700d66859a807bc21c6c22892838669c6
parent 338f65c65978b1fbc8a5236b29a5725535ae2a46
Author: Oliver Lowe <o@olowe.co>
Date:   Fri, 14 Jun 2024 14:00:48 +1000

mpegts: implement a streaming MPEG-TS scanner

This has the same interface as bufio.Scanner, but deals with MPEG-TS
packets instead of text. Most useful for applications using this package.

Diffstat:
M
mpegts/codec.go
 | 
42
++++++++++++++++++++++++++----------------
A
mpegts/scanner.go
 | 
57
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++

2 files changed, 83 insertions(+), 16 deletions(-)
diff --git a/mpegts/codec.go b/mpegts/codec.go
@@ -6,22 +6,13 @@ import (
 	"io"
 )
 
-func Decode(r io.Reader) (*Packet, error) {
-	buf := make([]byte, PacketSize)
-	n, err := r.Read(buf)
-	if n != PacketSize {
-		if err != nil {
-			return nil, fmt.Errorf("short read (%d bytes): %w", n, err)
-		}
-		return nil, fmt.Errorf("short read (%d bytes)", n)
-	}
-	if err != nil {
-		return nil, err
+func Unmarshal(buf []byte, p *Packet) error {
+	if len(buf) != PacketSize {
+		return fmt.Errorf("need exactly %d bytes, have %d", PacketSize, len(buf))
 	}
 	if buf[0] != Sync {
-		return nil, fmt.Errorf("expected sync byte, got %x", buf[0])
+		return fmt.Errorf("expected sync byte, got %x", buf[0])
 	}
-	var p Packet
 	if buf[1]&0x80 > 0 {
 		p.Error = true
 	}
@@ -52,10 +43,29 @@ func Decode(r io.Reader) (*Packet, error) {
 		alen := int(buf[4])
 		buf = buf[4+1+alen:]
 	default:
-		return nil, fmt.Errorf("neither adaptation field or payload present")
+		return fmt.Errorf("neither adaptation field or payload present")
+	}
+	if err := unmarshalPayload(buf, p); err != nil {
+		return fmt.Errorf("unmarshal payload: %w", err)
+	}
+	return nil
+}
+
+func Decode(r io.Reader) (*Packet, error) {
+	buf := make([]byte, PacketSize)
+	n, err := r.Read(buf)
+	if n != PacketSize {
+		if err != nil {
+			return nil, fmt.Errorf("short read (%d bytes): %w", n, err)
+		}
+		return nil, fmt.Errorf("short read (%d bytes)", n)
 	}
-	if err := unmarshalPayload(buf, &p); err != nil {
-		return nil, fmt.Errorf("unmarshal payload: %w", err)
+	if err != nil {
+		return nil, err
+	}
+	var p Packet
+	if err := Unmarshal(buf, &p); err != nil {
+		return &p, fmt.Errorf("unmarshal packet: %w", err)
 	}
 	return &p, nil
 }
diff --git a/mpegts/scanner.go b/mpegts/scanner.go
@@ -0,0 +1,57 @@
+package mpegts
+
+import (
+	"errors"
+	"fmt"
+	"io"
+)
+
+type Scanner struct {
+	buf    []byte
+	rd     io.Reader
+	packet *Packet
+	err    error
+}
+
+func NewScanner(rd io.Reader) *Scanner {
+	return &Scanner{
+		buf: make([]byte, PacketSize, PacketSize),
+		rd:  rd,
+	}
+}
+
+func (sc *Scanner) Err() error { return sc.err }
+
+func (sc *Scanner) Packet() *Packet { return sc.packet }
+
+func (sc *Scanner) Scan() bool {
+	n, err := sc.rd.Read(sc.buf)
+	if n < PacketSize {
+		if errors.Is(err, io.EOF) && n == 0 {
+			return false
+		} else if err != nil {
+			sc.err = fmt.Errorf("short read (%d bytes): %w", n, err)
+		} else {
+			sc.err = fmt.Errorf("short read: read %d bytes", n)
+		}
+		return false
+	}
+	if errors.Is(err, io.EOF) && n == PacketSize {
+		return false
+	} else if errors.Is(err, io.EOF) && n == 0 {
+		return false
+	} else if err != nil {
+		sc.err = err
+		return false
+	}
+	if n == PacketSize {
+		p := new(Packet)
+		if err := Unmarshal(sc.buf, p); err != nil {
+			sc.err = fmt.Errorf("unmarshal: %w", err)
+			return false
+		}
+		sc.packet = p
+		return true
+	}
+	return false
+}