| 207 | super().subscribe(observer) |
| 208 | |
| 209 | def flush_buffer(self): |
| 210 | while not self.closed: |
| 211 | source_was_closed = self.source_closed |
| 212 | |
| 213 | current_chunks = None |
| 214 | with self.buffer_lock: |
| 215 | if self.buffer_chunks: |
| 216 | current_chunks = self.buffer_chunks |
| 217 | self.buffer_chunks = [] |
| 218 | |
| 219 | with self.subscriber_lock: |
| 220 | if current_chunks: |
| 221 | if self.aggregate_function is not None: |
| 222 | current_chunks = self.aggregate_function(current_chunks) |
| 223 | |
| 224 | for chunk in current_chunks: |
| 225 | self._push(chunk) |
| 226 | |
| 227 | if source_was_closed: |
| 228 | self._close() |
| 229 | return |
| 230 | |
| 231 | time.sleep(self.period_millis / 1000.) |
| 232 | |
| 233 | |
| 234 | class _StoringObserver: |