| package ioutils |
| |
| const maxCap = 1e6 |
| |
| // BytesPipe is io.ReadWriter which works similarly to pipe(queue). |
| // All written data could be read only once. Also BytesPipe is allocating |
| // and releasing new byte slices to adjust to current needs, so there won't be |
| // overgrown buffer after high load peak. |
| // BytesPipe isn't goroutine-safe, caller must synchronize it if needed. |
| type BytesPipe struct { |
| buf [][]byte // slice of byte-slices of buffered data |
| lastRead int // index in the first slice to a read point |
| bufLen int // length of data buffered over the slices |
| } |
| |
| // NewBytesPipe creates new BytesPipe, initialized by specified slice. |
| // If buf is nil, then it will be initialized with slice which cap is 64. |
| // buf will be adjusted in a way that len(buf) == 0, cap(buf) == cap(buf). |
| func NewBytesPipe(buf []byte) *BytesPipe { |
| if cap(buf) == 0 { |
| buf = make([]byte, 0, 64) |
| } |
| return &BytesPipe{ |
| buf: [][]byte{buf[:0]}, |
| } |
| } |
| |
| // Write writes p to BytesPipe. |
| // It can allocate new []byte slices in a process of writing. |
| func (bp *BytesPipe) Write(p []byte) (n int, err error) { |
| for { |
| // write data to the last buffer |
| b := bp.buf[len(bp.buf)-1] |
| // copy data to the current empty allocated area |
| n := copy(b[len(b):cap(b)], p) |
| // increment buffered data length |
| bp.bufLen += n |
| // include written data in last buffer |
| bp.buf[len(bp.buf)-1] = b[:len(b)+n] |
| |
| // if there was enough room to write all then break |
| if len(p) == n { |
| break |
| } |
| |
| // more data: write to the next slice |
| p = p[n:] |
| // allocate slice that has twice the size of the last unless maximum reached |
| nextCap := 2 * cap(bp.buf[len(bp.buf)-1]) |
| if maxCap < nextCap { |
| nextCap = maxCap |
| } |
| // add new byte slice to the buffers slice and continue writing |
| bp.buf = append(bp.buf, make([]byte, 0, nextCap)) |
| } |
| return |
| } |
| |
| func (bp *BytesPipe) len() int { |
| return bp.bufLen - bp.lastRead |
| } |
| |
| // Read reads bytes from BytesPipe. |
| // Data could be read only once. |
| func (bp *BytesPipe) Read(p []byte) (n int, err error) { |
| for { |
| read := copy(p, bp.buf[0][bp.lastRead:]) |
| n += read |
| bp.lastRead += read |
| if bp.len() == 0 { |
| // we have read everything. reset to the beginning. |
| bp.lastRead = 0 |
| bp.bufLen -= len(bp.buf[0]) |
| bp.buf[0] = bp.buf[0][:0] |
| break |
| } |
| // break if everything was read |
| if len(p) == read { |
| break |
| } |
| // more buffered data and more asked. read from next slice. |
| p = p[read:] |
| bp.lastRead = 0 |
| bp.bufLen -= len(bp.buf[0]) |
| bp.buf[0] = nil // throw away old slice |
| bp.buf = bp.buf[1:] // switch to next |
| } |
| return |
| } |