ReadAt reads data from the AsyncBuffer at the given offset. Please note that if pause threshold is hit in the middle of the reading, the data beyond the threshold may not be available. If the reader is paused and we try to read data beyond the pause threshold, it will wait till something could be
(p []byte, off int64)
| 234 | // If the reader is paused and we try to read data beyond the pause threshold, |
| 235 | // it will wait till something could be returned. |
| 236 | func (ab *AsyncBuffer) ReadAt(p []byte, off int64) (int, error) { |
| 237 | size := int64(len(p)) // total size of the data to read |
| 238 | |
| 239 | if off < 0 { |
| 240 | return 0, errors.New("asyncbuffer.AsyncBuffer.ReadAt: negative offset") |
| 241 | } |
| 242 | |
| 243 | // If we plan to hit threshold while reading, release the paused reader |
| 244 | if int64(len(p))+off > PauseThreshold { |
| 245 | ab.paused.Release() |
| 246 | } |
| 247 | |
| 248 | // Wait for the offset to be available. |
| 249 | // It may return io.EOF if the offset is beyond the end of the stream. |
| 250 | err := ab.WaitFor(off) |
| 251 | if err != nil { |
| 252 | return 0, err |
| 253 | } |
| 254 | |
| 255 | // We lock the mutex until current buffer is read |
| 256 | ab.mu.RLock() |
| 257 | defer ab.mu.RUnlock() |
| 258 | |
| 259 | // If the reader is closed, we return an error |
| 260 | if err := ab.closedError(); err != nil { |
| 261 | return 0, err |
| 262 | } |
| 263 | |
| 264 | // Read data from the first chunk |
| 265 | n := ab.readChunkAt(p, off) |
| 266 | if n == 0 { |
| 267 | return 0, io.EOF // Failed to read any data: means we tried to read beyond the end of the stream |
| 268 | } |
| 269 | |
| 270 | size -= int64(n) |
| 271 | off += int64(n) // Here and beyond off always points to the last read byte + 1 |
| 272 | |
| 273 | // Now, let's try to read the rest of the data from next chunks while they are available |
| 274 | for size > 0 { |
| 275 | // If data is not available at the given offset, we can return data read so far. |
| 276 | ok, err := ab.offsetAvailable(off) |
| 277 | if !ok { |
| 278 | if errors.Is(err, io.EOF) { |
| 279 | return n, nil |
| 280 | } |
| 281 | |
| 282 | return n, err |
| 283 | } |
| 284 | |
| 285 | // Read data from the next chunk |
| 286 | nX := ab.readChunkAt(p[n:], off) |
| 287 | n += nX |
| 288 | size -= int64(nX) |
| 289 | off += int64(nX) |
| 290 | |
| 291 | // If we read data shorter than ChunkSize or, in case that was the last chunk, less than |
| 292 | // the size of the tail, return kind of EOF |
| 293 | if int64(nX) < min(size, int64(ChunkSize)) { |