5670260071

Committed 26 Jul 2023 02:46PM UTC coverage: 78.3% (+1.0%) from 77.339%

Build # 5670260071

Build Type

push

github

Committed by

asticode

Commit Message

Added github action

Run Details

3132 of 4000 relevant lines covered (78.3%)

14.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.56

/packet_buffer.go

package astits

import (
        "bufio"
        "errors"
        "fmt"
        "io"

        "github.com/asticode/go-astikit"
)

// packetBuffer represents a packet buffer
type packetBuffer struct {
        packetSize       int
        s                PacketSkipper
        r                io.Reader
        packetReadBuffer []byte
}

// newPacketBuffer creates a new packet buffer
func newPacketBuffer(r io.Reader, packetSize int, s PacketSkipper) (pb *packetBuffer, err error) {
        // Init
        pb = &packetBuffer{
                packetSize: packetSize,
                s:          s,
                r:          r,
        }

        // Packet size is not set
        if pb.packetSize == 0 {
                // Auto detect packet size
                if pb.packetSize, err = autoDetectPacketSize(r); err != nil {
                        err = fmt.Errorf("astits: auto detecting packet size failed: %w", err)
                        return
                }
        }
        return
}

// autoDetectPacketSize updates the packet size based on the first bytes
// Minimum packet size is 188 and is bounded by 2 sync bytes
// Assumption is made that the first byte of the reader is a sync byte
func autoDetectPacketSize(r io.Reader) (packetSize int, err error) {
        // Read first bytes
        const l = 193
        var b = make([]byte, l)
        shouldRewind, rerr := peek(r, b)
        if rerr != nil {
                err = fmt.Errorf("astits: reading first %d bytes failed: %w", l, rerr)
                return
        }

        // Packet must start with a sync byte
        if b[0] != syncByte {
                err = ErrPacketMustStartWithASyncByte
                return
        }

        // Look for sync bytes
        for idx, b := range b {
                if b == syncByte && idx >= MpegTsPacketSize {
                        // Update packet size
                        packetSize = idx

                        if !shouldRewind {
                                return
                        }

                        // Rewind or sync reader
                        var n int64
                        if n, err = rewind(r); err != nil {
                                err = fmt.Errorf("astits: rewinding failed: %w", err)
                                return
                        } else if n == -1 {
                                var ls = packetSize - (l - packetSize)
                                if _, err = r.Read(make([]byte, ls)); err != nil {
                                        err = fmt.Errorf("astits: reading %d bytes to sync reader failed: %w", ls, err)
                                        return
                                }
                        }
                        return
                }
        }
        err = fmt.Errorf("astits: only one sync byte detected in first %d bytes", l)
        return
}

// bufio.Reader can't be rewinded, which leads to packet loss on packet size autodetection
// but it has handy Peek() method
// so what we do here is peeking bytes for bufio.Reader and falling back to rewinding/syncing for all other readers
func peek(r io.Reader, b []byte) (shouldRewind bool, err error) {
        if br, ok := r.(*bufio.Reader); ok {
                var bs []byte
                bs, err = br.Peek(len(b))
                if err != nil {
                        return
                }
                copy(b, bs)
                return false, nil
        }

        _, err = r.Read(b)
        shouldRewind = true
        return
}

// rewind rewinds the reader if possible, otherwise n = -1
func rewind(r io.Reader) (n int64, err error) {
        if s, ok := r.(io.Seeker); ok {
                if n, err = s.Seek(0, 0); err != nil {
                        err = fmt.Errorf("astits: seeking to 0 failed: %w", err)
                        return
                }
                return
        }
        n = -1
        return
}

// next fetches the next packet from the buffer
func (pb *packetBuffer) next() (p *Packet, err error) {
        // Read
        if pb.packetReadBuffer == nil || len(pb.packetReadBuffer) != pb.packetSize {
                pb.packetReadBuffer = make([]byte, pb.packetSize)
        }

        // Loop to make sure we return a packet even if first packets are skipped
        for p == nil {
                if _, err = io.ReadFull(pb.r, pb.packetReadBuffer); err != nil {
                        if err == io.EOF || err == io.ErrUnexpectedEOF {
                                err = ErrNoMorePackets
                        } else {
                                err = fmt.Errorf("astits: reading %d bytes failed: %w", pb.packetSize, err)
                        }
                        return
                }

                // Parse packet
                if p, err = parsePacket(astikit.NewBytesIterator(pb.packetReadBuffer), pb.s); err != nil {
                        if !errors.Is(err, errSkippedPacket) {
                                err = fmt.Errorf("astits: building packet failed: %w", err)
                                return
                        }
                }
        }

        return
}

1	package astits
2
3	import (
4	"bufio"
5	"errors"
6	"fmt"
7	"io"
8
9	"github.com/asticode/go-astikit"
10	)
11
12	// packetBuffer represents a packet buffer
13	type packetBuffer struct {
14	packetSize int
15	s PacketSkipper
16	r io.Reader
17	packetReadBuffer []byte
18	}
19
20	// newPacketBuffer creates a new packet buffer
21	func newPacketBuffer(r io.Reader, packetSize int, s PacketSkipper) (pb *packetBuffer, err error) {	16✔
22	// Init	16✔
23	pb = &packetBuffer{	16✔
24	packetSize: packetSize,	16✔
25	s: s,	16✔
26	r: r,	16✔
27	}	16✔
28		16✔
29	// Packet size is not set	16✔
30	if pb.packetSize == 0 {	19✔
31	// Auto detect packet size	3✔
32	if pb.packetSize, err = autoDetectPacketSize(r); err != nil {	3✔
33	err = fmt.Errorf("astits: auto detecting packet size failed: %w", err)	×
34	return	×
35	}	×
36	}
37	return	16✔
38	}
39
40	// autoDetectPacketSize updates the packet size based on the first bytes
41	// Minimum packet size is 188 and is bounded by 2 sync bytes
42	// Assumption is made that the first byte of the reader is a sync byte
43	func autoDetectPacketSize(r io.Reader) (packetSize int, err error) {	5✔
44	// Read first bytes	5✔
45	const l = 193	5✔
46	var b = make([]byte, l)	5✔
47	shouldRewind, rerr := peek(r, b)	5✔
48	if rerr != nil {	5✔
49	err = fmt.Errorf("astits: reading first %d bytes failed: %w", l, rerr)	×
50	return	×
51	}	×
52
53	// Packet must start with a sync byte
54	if b[0] != syncByte {	6✔
55	err = ErrPacketMustStartWithASyncByte	1✔
56	return	1✔
57	}	1✔
58
59	// Look for sync bytes
60	for idx, b := range b {	772✔
61	if b == syncByte && idx >= MpegTsPacketSize {	772✔
62	// Update packet size	4✔
63	packetSize = idx	4✔
64		4✔
65	if !shouldRewind {	4✔
66	return	×
67	}	×
68
69	// Rewind or sync reader
70	var n int64	4✔
71	if n, err = rewind(r); err != nil {	4✔
72	err = fmt.Errorf("astits: rewinding failed: %w", err)	×
73	return	×
74	} else if n == -1 {	4✔
75	var ls = packetSize - (l - packetSize)	×
76	if _, err = r.Read(make([]byte, ls)); err != nil {	×
77	err = fmt.Errorf("astits: reading %d bytes to sync reader failed: %w", ls, err)	×
78	return	×
79	}	×
80	}
81	return	4✔
82	}
83	}
84	err = fmt.Errorf("astits: only one sync byte detected in first %d bytes", l)	×
85	return	×
86	}
87
88	// bufio.Reader can't be rewinded, which leads to packet loss on packet size autodetection
89	// but it has handy Peek() method
90	// so what we do here is peeking bytes for bufio.Reader and falling back to rewinding/syncing for all other readers
91	func peek(r io.Reader, b []byte) (shouldRewind bool, err error) {	5✔
92	if br, ok := r.(*bufio.Reader); ok {	5✔
93	var bs []byte	×
94	bs, err = br.Peek(len(b))	×
95	if err != nil {	×
96	return	×
97	}	×
98	copy(b, bs)	×
99	return false, nil	×
100	}
101
102	_, err = r.Read(b)	5✔
103	shouldRewind = true	5✔
104	return	5✔
105	}
106
107	// rewind rewinds the reader if possible, otherwise n = -1
108	func rewind(r io.Reader) (n int64, err error) {	6✔
109	if s, ok := r.(io.Seeker); ok {	12✔
110	if n, err = s.Seek(0, 0); err != nil {	6✔
111	err = fmt.Errorf("astits: seeking to 0 failed: %w", err)	×
112	return	×
113	}	×
114	return	6✔
115	}
116	n = -1	×
117	return	×
118	}
119
120	// next fetches the next packet from the buffer
121	func (pb packetBuffer) next() (p Packet, err error) {	89✔
122	// Read	89✔
123	if pb.packetReadBuffer == nil \|\| len(pb.packetReadBuffer) != pb.packetSize {	105✔
124	pb.packetReadBuffer = make([]byte, pb.packetSize)	16✔
125	}	16✔
126
127	// Loop to make sure we return a packet even if first packets are skipped
128	for p == nil {	178✔
129	if _, err = io.ReadFull(pb.r, pb.packetReadBuffer); err != nil {	108✔
130	if err == io.EOF \|\| err == io.ErrUnexpectedEOF {	38✔
131	err = ErrNoMorePackets	19✔
132	} else {	19✔
133	err = fmt.Errorf("astits: reading %d bytes failed: %w", pb.packetSize, err)	×
134	}	×
135	return	19✔
136	}
137
138	// Parse packet
139	if p, err = parsePacket(astikit.NewBytesIterator(pb.packetReadBuffer), pb.s); err != nil {	76✔
140	if !errors.Is(err, errSkippedPacket) {	12✔
141	err = fmt.Errorf("astits: building packet failed: %w", err)	6✔
142	return	6✔
143	}	6✔
144	}
145	}
146
147	return	64✔
148	}

asticode / go-astits / 5670260071

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous