readCompressedFrame parses the compression header and reads the compressed payload from the peaker into a buffer. This allows the peaker to move on and the worker to decompress the buffer concurrently. (A more sophisticated implementation could sync the peeker movement to the decode pipeline to av
(code byte)
| 181 | // peeker buffer and be released after decompression. A reference-counted double |
| 182 | // buffer would work nicely for this.) |
| 183 | func (p *parser) readCompressedFrame(code byte) (frame, error) { |
| 184 | n, err := p.decodeLength(code) |
| 185 | if err != nil { |
| 186 | return frame{}, err |
| 187 | } |
| 188 | format, err := p.peeker.ReadByte() |
| 189 | if err != nil { |
| 190 | return frame{}, err |
| 191 | } |
| 192 | size, err := readUvarintAsInt(p.peeker) |
| 193 | if err != nil { |
| 194 | return frame{}, err |
| 195 | } |
| 196 | if size < 1 { |
| 197 | return frame{}, fmt.Errorf("bsupio: frame length (%d) too small", size) |
| 198 | } |
| 199 | if size > p.maxSize { |
| 200 | return frame{}, fmt.Errorf("bsupio: frame length (%d) exceeds maximum allowed (%d)", size, p.maxSize) |
| 201 | } |
| 202 | // The size of the compressed buffer needs to be adjusted by the |
| 203 | // byte for the format and the variable-length bytes to encode |
| 204 | // the original size. |
| 205 | n -= 1 + scode.SizeOfUvarint(uint64(size)) |
| 206 | b, err := p.peeker.Read(n) |
| 207 | if err != nil && err != io.EOF { |
| 208 | if err == peeker.ErrBufferOverflow { |
| 209 | return frame{}, fmt.Errorf("large value of %d bytes exceeds maximum read buffer", n) |
| 210 | } |
| 211 | return frame{}, errBadFormat |
| 212 | } |
| 213 | return frame{ |
| 214 | fmt: CompressionFormat(format), |
| 215 | zbuf: newBufferFromBytes(b), |
| 216 | ubuf: newBuffer(size), |
| 217 | }, nil |
| 218 | } |
| 219 | |
| 220 | func (p *parser) decodeLength(code byte) (int, error) { |
| 221 | v, err := readUvarintAsInt(p.peeker) |
no test coverage detected