Get a ring buffer by name
(self, name, blocking=True)
| 292 | return result |
| 293 | |
| 294 | def get_ring_buffer(self, name, blocking=True): |
| 295 | """Get a ring buffer by name""" |
| 296 | if not self.connected: |
| 297 | self.connect() |
| 298 | |
| 299 | while True: |
| 300 | try: |
| 301 | if self.last_epoch != self.epoch: |
| 302 | self.refresh(blocking) |
| 303 | with self.lock: |
| 304 | entry = self.directory[name] |
| 305 | if entry.type == 8: # STAT_DIR_TYPE_RING_BUFFER |
| 306 | return entry.get_counter(self) |
| 307 | else: |
| 308 | raise ValueError(f"'{name}' is not a ring buffer") |
| 309 | except IOError: |
| 310 | if not blocking: |
| 311 | raise |
| 312 | |
| 313 | def poll_ring_buffer(self, name, thread_index=0, timeout=None, callback=None): |
| 314 | """Convenience method to poll a ring buffer by name""" |