flushBlocks decompresses and flushes consecutive blocks in buf, interleaving each of them with a `\n` in the decompressed buffer.
(buf []byte)
| 198 | // flushBlocks decompresses and flushes consecutive blocks in buf, interleaving each of |
| 199 | // them with a `\n` in the decompressed buffer. |
| 200 | func (c *BufferCache) flushBlocks(buf []byte) { |
| 201 | var ( |
| 202 | pos uint64 |
| 203 | zpos int // indices on buf and zbuf |
| 204 | n int // length of last decompressed buffer |
| 205 | ) |
| 206 | |
| 207 | // Decompress consecutive blocks. Each block is prepended with an uint32 |
| 208 | // representing the block length. The outer loop moves pos and zpos. |
| 209 | // The inner loop handles failed decompression due to not enough space in |
| 210 | // the destination buffer to hold the currently decompressed block. |
| 211 | for pos != uint64(len(buf)) { |
| 212 | // Extract block length from the first 4 bytes. |
| 213 | prefix := uint64(binary.LittleEndian.Uint32(buf[pos : pos+4])) |
| 214 | blocklen, compressed := prefix&0x7fffffff, (prefix&0x80000000) != 0 |
| 215 | pos += 4 |
| 216 | |
| 217 | if !compressed { // just copy it |
| 218 | for { |
| 219 | if uint64(len(c.decomp.buf[zpos:])) < blocklen { |
| 220 | c.decomp.grow() |
| 221 | continue |
| 222 | } |
| 223 | n = copy(c.decomp.buf[zpos:], buf[pos:pos+blocklen]) |
| 224 | break |
| 225 | } |
| 226 | } else { |
| 227 | n = c.decomp.uncompress(zpos, buf[pos:pos+blocklen]) |
| 228 | } |
| 229 | |
| 230 | pos += blocklen |
| 231 | zpos += n |
| 232 | |
| 233 | c.decomp.copyByte(zpos, '\n') |
| 234 | zpos++ |
| 235 | } |
| 236 | |
| 237 | // Return the effective part of c.zbuf |
| 238 | c.onFlush(c.decomp.bytes(zpos)) |
| 239 | } |
| 240 | |
| 241 | func (c *BufferCache) flushCold() { |
| 242 | for b := range c.cold.buckets { |