Moves cli helpers to pkgs

cli/parsefile.go

@@ -26,6 +26,7 @@ SOFTWARE.
 package main
 
 import (
+	"bufio"
 	"bytes"
 	"flag"
 	"fmt"
@@ -61,18 +62,13 @@ func main() {
 		}
 	}(tsFile)
 	// Verify if sync-byte is present and seek to the first sync-byte
-	syncIndex, err := sync(tsFile)
-	if err == nil {
-		_, err = tsFile.Seek(syncIndex, 0)
-		if err != nil {
-			fmt.Println(err)
-			return
-		}
-	} else {
+	reader := bufio.NewReader(tsFile)
+	_, err = packet.Sync(reader)
+	if err != nil {
 		fmt.Println(err)
 		return
 	}
-	pat, err := extractPat(tsFile)
+	pat, err := psi.ReadPAT(reader)
 	if err != nil {
 		println(err)
 		return
@@ -82,24 +78,25 @@ func main() {
 	if *showPmt {
 		pm := pat.ProgramMap()
 		for pn, pid := range pm {
-			pmt, err := extractPmt(tsFile, pid)
+			pmt, err := psi.ReadPMT(reader, pid)
 			if err != nil {
 				panic(err)
 			}
 			printPmt(pn, pmt)
 		}
 	}
 
-	pkt := make([]byte, packet.PacketSize, packet.PacketSize)
-	var offset int64
+	pkt := make(packet.Packet, packet.PacketSize)
 	var numPackets uint64
 	ebps := make(map[uint64]ebp.EncoderBoundaryPoint)
 	for {
-		read, err := tsFile.ReadAt(pkt, offset)
-		if err == io.EOF || read == 0 {
-			break
+		if _, err := io.ReadFull(reader, pkt); err != nil {
+			if err == io.EOF || err == io.ErrUnexpectedEOF {
+				break
+			}
+			println(err)
+			return
 		}
-		offset += packet.PacketSize
 		numPackets++
 		if *showEbp {
 			ebpBytes, err := adaptationfield.EncoderBoundaryPoint(pkt)
@@ -150,99 +147,6 @@ func printPat(pat psi.PAT) {
 	printlnf("\tNumber of Programs %v", pat.NumPrograms())
 }
 
-func extractPat(buf io.Reader) (psi.PAT, error) {
-	pkt := make([]byte, packet.PacketSize)
-	var pat psi.PAT
-	for read, err := buf.Read(pkt); pat == nil; read, err = buf.Read(pkt) {
-		if err != nil {
-			return nil, err
-		}
-		if read <= 0 {
-			return nil, fmt.Errorf("Reached EOF without PAT")
-		}
-		pid, err := packet.Pid(pkt)
-		if err != nil {
-			return nil, err
-		}
-		if pid == 0 {
-			pay, err := packet.Payload(pkt)
-			if err != nil {
-				println(err)
-				continue
-			}
-			cp := make([]byte, len(pay))
-			copy(cp, pay)
-			pat, err := psi.NewPAT(cp)
-			if err != nil {
-				println(err)
-				continue
-			}
-			return pat, nil
-		}
-	}
-	return nil, fmt.Errorf("No pat found")
-}
 func printlnf(format string, a ...interface{}) {
 	fmt.Printf(format+"\n", a...)
 }
-func extractPmt(buf io.Reader, pid uint16) (psi.PMT, error) {
-	pkt := make([]byte, packet.PacketSize)
-	pmtAcc := packet.NewAccumulator(psi.PmtAccumulatorDoneFunc)
-	var pmt psi.PMT
-	for read, err := buf.Read(pkt); pmt == nil && read > 0; read, err = buf.Read(pkt) {
-		if err != nil {
-			return nil, err
-		}
-		currPid, err := packet.Pid(pkt)
-		if err != nil {
-			return nil, err
-		}
-		if currPid == pid {
-			done, err := pmtAcc.Add(pkt)
-			if err != nil {
-				return nil, err
-			}
-			if done {
-				b, err := pmtAcc.Parse()
-				if err != nil {
-					return nil, err
-				}
-				pmt, err = psi.NewPMT(b)
-				if err != nil {
-					return nil, err
-				}
-
-			}
-
-		}
-	}
-	return pmt, nil
-}
-func sync(buf io.Reader) (int64, error) {
-	// function find the first sync byte of the array
-	data := make([]byte, 1)
-	for i := int64(0); ; i++ {
-		read, err := buf.Read(data)
-		if err != nil && err != io.EOF {
-			println(err)
-		}
-		if read == 0 {
-			break
-		}
-		if int(data[0]) == packet.SyncByte {
-			// check next 188th byte
-			nextData := make([]byte, packet.PacketSize)
-			nextRead, err := buf.Read(nextData)
-			if err != nil && err != io.EOF {
-				println(err)
-			}
-			if nextRead == 0 {
-				break
-			}
-			if nextData[187] == packet.SyncByte {
-				return i, nil
-			}
-		}
-	}
-	return 0, fmt.Errorf("Sync-byte not found.")
-}

errors.go

@@ -43,6 +43,12 @@ var (
 	ErrNoPCR = errors.New("adaptation field has no Program Clock Reference")
 	// ErrNoOPCR is returned when an attempt is made to access an adaptation field OPCR that does not exist
 	ErrNoOPCR = errors.New("adaptation field has no Original Program Clock Reference")
+	// ErrPATNotFound is returned when expected PAT packet is not found when
+	// reading TS packets.
+	ErrPATNotFound = errors.New("No PAT was found while reading TS")
+	// ErrPMTNotFound is returned when expected PMT packet(s) are not found when
+	// reading TS packets.
+	ErrPMTNotFound = errors.New("No PMT was found while reading TS")
 	// ErrParsePMTDescriptor is returned when a PMT descriptor cannot be parsed
 	ErrParsePMTDescriptor = errors.New("unable to parse PMT descriptor")
 	// ErrInvalidPATLength is returned when a PAT cannot be parsed because there are not enough bytes
@@ -70,4 +76,7 @@ var (
 	ErrSCTE35DescriptorNotFound = errors.New("Cannot close descriptor that's not in the open list")
 	// ErrNilPAT is returned when a PAT is passed into a function for which it cannot be nil.
 	ErrNilPAT = errors.New("Nil PAT not allowed here.")
+	// ErrSyncByteNotFound is returned when a packet sync byte could not be found
+	// when reading.
+	ErrSyncByteNotFound = errors.New("Sync-byte not found.")
 )

packet/io.go

@@ -0,0 +1,65 @@
+/*
+MIT License
+
+Copyright 2016 Comcast Cable Communications Management, LLC
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+package packet
+
+import (
+	"bufio"
+	"io"
+
+	"github.com/Comcast/gots"
+)
+
+// Sync finds the offset of the next packet sync byte and advances the reader
+// to the packet start. It also checks the next 188th byte to ensure a sync is
+// found. It returns the offset of the sync w.r.t. the original reader
+// position.
+func Sync(r *bufio.Reader) (int64, error) {
+	data := make([]byte, 1)
+	for i := int64(0); ; i++ {
+		_, err := io.ReadFull(r, data)
+		if err == io.EOF || err == io.ErrUnexpectedEOF {
+			break
+		}
+		if err != nil {
+			return 0, err
+		}
+		if int(data[0]) == SyncByte {
+			// check next 188th byte
+			rp := bufio.NewReaderSize(r, PacketSize) // extends only if needed
+			nextData, err := rp.Peek(PacketSize)
+			if err == io.EOF || err == io.ErrUnexpectedEOF {
+				break
+			}
+			if err != nil {
+				return 0, err
+			}
+			if nextData[187] == SyncByte {
+				r.UnreadByte()
+				return i, nil
+			}
+		}
+	}
+	return 0, gots.ErrSyncByteNotFound
+}

packet/io_test.go

@@ -0,0 +1,107 @@
+/*
+MIT License
+
+Copyright 2016 Comcast Cable Communications Management, LLC
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+package packet
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/hex"
+	"testing"
+)
+
+func TestSyncForSmoke(t *testing.T) {
+	bs, _ := hex.DecodeString("474000100000b00d0001c100000001e256f803e71bfffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"ff4742")
+	r := bufio.NewReader(bytes.NewReader(bs))
+
+	offset, err := Sync(r)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	if offset != 0 {
+		t.Errorf("Incorrect offset returned for next sync marker")
+	}
+}
+
+func TestSyncNonZeroOffset(t *testing.T) {
+	bs, _ := hex.DecodeString("ffffff474000100000b00d0001c100000001e256f803e71bfffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"ff4742")
+	r := bufio.NewReader(bytes.NewReader(bs))
+
+	offset, err := Sync(r)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	if offset != 3 {
+		t.Errorf("Incorrect offset returned for next sync marker")
+	}
+}
+
+func TestSyncNotFound(t *testing.T) {
+	// no sync byte here
+	bs, _ := hex.DecodeString("ff4000100000b00d0001c100000001e256f803e71bfffffff")
+	r := bufio.NewReader(bytes.NewReader(bs))
+
+	_, err := Sync(r)
+	if err == nil {
+		t.Errorf("Expected there to be an error, but there was not")
+	}
+}
+
+func TestSyncReaderPosAtPacketStart(t *testing.T) {
+	bs, _ := hex.DecodeString("ffffff474000100000b00d0001c100000001e256f803e71bfffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" +
+		"ff4742")
+	r := bufio.NewReader(bytes.NewReader(bs))
+
+	_, err := Sync(r)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	bytes := make([]byte, 3)
+	_, err = r.Read(bytes)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	wantHex := "474000"
+	gotHex := hex.EncodeToString(bytes)
+	if gotHex != wantHex {
+		t.Errorf("Reader not left in correct spot. Wanted next read %v, got %v", wantHex, gotHex)
+	}
+}