Manages buffering of stream segments with thread-safe access. Attributes: buffer (Dict[int, bytes]): Maps sequence numbers to segment data lock (threading.Lock): Thread safety for buffer access Features: - Thread-safe segment storage and retrieval
| 25 | buffer_lock = threading.Lock() # Synchronizes access to buffers |
| 26 | |
| 27 | class StreamBuffer: |
| 28 | """ |
| 29 | Manages buffering of stream segments with thread-safe access. |
| 30 | |
| 31 | Attributes: |
| 32 | buffer (Dict[int, bytes]): Maps sequence numbers to segment data |
| 33 | lock (threading.Lock): Thread safety for buffer access |
| 34 | |
| 35 | Features: |
| 36 | - Thread-safe segment storage and retrieval |
| 37 | - Automatic cleanup of old segments |
| 38 | - Sequence number based indexing |
| 39 | """ |
| 40 | |
| 41 | def __init__(self): |
| 42 | self.buffer: Dict[int, bytes] = {} # Maps sequence numbers to segment data |
| 43 | self.lock: threading.Lock = threading.Lock() |
| 44 | |
| 45 | def __getitem__(self, key: int) -> Optional[bytes]: |
| 46 | """Get segment data by sequence number""" |
| 47 | return self.buffer.get(key) |
| 48 | |
| 49 | def __setitem__(self, key: int, value: bytes): |
| 50 | """Store segment data by sequence number""" |
| 51 | self.buffer[key] = value |
| 52 | # Cleanup old segments if we exceed MAX_SEGMENTS |
| 53 | if len(self.buffer) > Config.MAX_SEGMENTS: |
| 54 | keys = sorted(self.buffer.keys()) |
| 55 | # Keep the most recent MAX_SEGMENTS |
| 56 | to_remove = keys[:-Config.MAX_SEGMENTS] |
| 57 | for k in to_remove: |
| 58 | del self.buffer[k] |
| 59 | |
| 60 | def __contains__(self, key: int) -> bool: |
| 61 | """Check if sequence number exists in buffer""" |
| 62 | return key in self.buffer |
| 63 | |
| 64 | def keys(self) -> List[int]: |
| 65 | """Get list of available sequence numbers""" |
| 66 | return list(self.buffer.keys()) |
| 67 | |
| 68 | def cleanup(self, keep_sequences: List[int]): |
| 69 | """Remove segments not in keep list""" |
| 70 | for seq in list(self.buffer.keys()): |
| 71 | if seq not in keep_sequences: |
| 72 | del self.buffer[seq] |
| 73 | |
| 74 | class ClientManager: |
| 75 | """Manages client connections and activity tracking""" |