Initialize a channel without redundant active key
(self, url, channel_id, user_agent=None, transcode=False, stream_id=None)
| 550 | return False |
| 551 | |
| 552 | def initialize_channel(self, url, channel_id, user_agent=None, transcode=False, stream_id=None): |
| 553 | """Initialize a channel without redundant active key""" |
| 554 | try: |
| 555 | if self._channel_unavailable_for_new_clients(channel_id): |
| 556 | logger.warning( |
| 557 | f"Refusing to initialize channel {channel_id}; " |
| 558 | f"teardown or pending shutdown active" |
| 559 | ) |
| 560 | return False |
| 561 | |
| 562 | if self._has_local_upstream_activity(channel_id): |
| 563 | logger.warning( |
| 564 | f"Stopping lingering local upstream before initializing channel {channel_id}" |
| 565 | ) |
| 566 | self._stop_local_stream_activity(channel_id) |
| 567 | |
| 568 | if self.redis_client: |
| 569 | metadata_key = RedisKeys.channel_metadata(channel_id) |
| 570 | if self.redis_client.exists(metadata_key): |
| 571 | metadata = self.redis_client.hgetall(metadata_key) |
| 572 | if 'state' in metadata: |
| 573 | state = metadata['state'] |
| 574 | active_states = [ChannelState.INITIALIZING, ChannelState.CONNECTING, |
| 575 | ChannelState.WAITING_FOR_CLIENTS, ChannelState.ACTIVE, ChannelState.BUFFERING] |
| 576 | if state in active_states: |
| 577 | logger.info(f"Channel {channel_id} already being initialized with state {state}") |
| 578 | # Create buffer and client manager only if we don't have them |
| 579 | if channel_id not in self.stream_buffers: |
| 580 | self.stream_buffers[channel_id] = StreamBuffer(channel_id, redis_client=RedisClient.get_buffer()) |
| 581 | if channel_id not in self.client_managers: |
| 582 | self.client_managers[channel_id] = ClientManager( |
| 583 | channel_id, |
| 584 | redis_client=self.redis_client, |
| 585 | worker_id=self.worker_id |
| 586 | ) |
| 587 | return True |
| 588 | |
| 589 | # Create buffer and client manager instances (or reuse if they exist) |
| 590 | if channel_id not in self.stream_buffers: |
| 591 | buffer = StreamBuffer(channel_id, redis_client=RedisClient.get_buffer()) |
| 592 | self.stream_buffers[channel_id] = buffer |
| 593 | |
| 594 | if channel_id not in self.client_managers: |
| 595 | client_manager = ClientManager( |
| 596 | channel_id, |
| 597 | redis_client=self.redis_client, |
| 598 | worker_id=self.worker_id |
| 599 | ) |
| 600 | self.client_managers[channel_id] = client_manager |
| 601 | |
| 602 | if self.redis_client: |
| 603 | # Set early initialization state to prevent race conditions |
| 604 | metadata_key = RedisKeys.channel_metadata(channel_id) |
| 605 | initial_metadata = { |
| 606 | "state": ChannelState.INITIALIZING, |
| 607 | "init_time": str(time.time()), |
| 608 | "owner": self.worker_id |
| 609 | } |
no test coverage detected