Main execution loop using HTTP streaming with improved connection handling and stream switching
(self)
| 336 | return False |
| 337 | |
| 338 | def run(self): |
| 339 | """Main execution loop using HTTP streaming with improved connection handling and stream switching""" |
| 340 | # Add a stop flag to the class properties |
| 341 | self.stop_requested = False |
| 342 | # Add tracking for stream switching attempts |
| 343 | stream_switch_attempts = 0 |
| 344 | # Get max stream switches from config using the helper method |
| 345 | max_stream_switches = ConfigHelper.max_stream_switches() # Prevent infinite switching loops |
| 346 | |
| 347 | try: |
| 348 | |
| 349 | |
| 350 | # Start health monitor thread |
| 351 | health_thread = threading.Thread(target=self._monitor_health, daemon=True) |
| 352 | health_thread.start() |
| 353 | |
| 354 | logger.info(f"Starting stream for URL: {self.url} for channel {self.channel_id}") |
| 355 | |
| 356 | # Main stream switching loop - we'll try different streams if needed |
| 357 | while self.running and stream_switch_attempts <= max_stream_switches: |
| 358 | close_old_connections() |
| 359 | if not self._ensure_owner_or_stop(): |
| 360 | break |
| 361 | # Check for stuck switching state |
| 362 | if self.url_switching and time.time() - self.url_switch_start_time > self.url_switch_timeout: |
| 363 | logger.warning(f"URL switching state appears stuck for channel {self.channel_id} " |
| 364 | f"({time.time() - self.url_switch_start_time:.1f}s > {self.url_switch_timeout}s timeout). " |
| 365 | f"Resetting switching state.") |
| 366 | self._reset_url_switching_state() |
| 367 | |
| 368 | # NEW: Check for health monitor recovery requests |
| 369 | if hasattr(self, 'needs_reconnect') and self.needs_reconnect and not self.url_switching: |
| 370 | logger.info(f"Health monitor requested reconnect for channel {self.channel_id}") |
| 371 | self.needs_reconnect = False |
| 372 | |
| 373 | # Attempt reconnect without changing streams |
| 374 | if self._attempt_reconnect(): |
| 375 | logger.info(f"Health-requested reconnect successful for channel {self.channel_id}") |
| 376 | continue # Go back to main loop |
| 377 | else: |
| 378 | logger.warning(f"Health-requested reconnect failed, will try stream switch for channel {self.channel_id}") |
| 379 | self.needs_stream_switch = True |
| 380 | |
| 381 | if hasattr(self, 'needs_stream_switch') and self.needs_stream_switch and not self.url_switching: |
| 382 | logger.info(f"Health monitor requested stream switch for channel {self.channel_id}") |
| 383 | self.needs_stream_switch = False |
| 384 | |
| 385 | if self._try_next_stream(): |
| 386 | logger.info(f"Health-requested stream switch successful for channel {self.channel_id}") |
| 387 | stream_switch_attempts += 1 |
| 388 | self.retry_count = 0 # Reset retries for new stream |
| 389 | continue # Go back to main loop with new stream |
| 390 | else: |
| 391 | logger.error(f"Health-requested stream switch failed for channel {self.channel_id}") |
| 392 | # Continue with normal flow |
| 393 | |
| 394 | # Check stream type before connecting |
| 395 | self.stream_type = detect_stream_type(self.url) |