Get the worker ID that owns this channel with proper error handling
(self, channel_id)
| 422 | thread.start() |
| 423 | |
| 424 | def get_channel_owner(self, channel_id): |
| 425 | """Get the worker ID that owns this channel with proper error handling""" |
| 426 | if not self.redis_client: |
| 427 | return None |
| 428 | |
| 429 | try: |
| 430 | lock_key = RedisKeys.channel_owner(channel_id) |
| 431 | result = self._execute_redis_command( |
| 432 | lambda: self.redis_client.get(lock_key) |
| 433 | ) |
| 434 | if result is None: |
| 435 | return None |
| 436 | try: |
| 437 | return result |
| 438 | except (AttributeError, UnicodeDecodeError) as e: |
| 439 | logger.error(f"Error decoding channel owner for {channel_id}: {e}, raw={result!r}") |
| 440 | return None |
| 441 | except Exception as e: |
| 442 | logger.error(f"Error getting channel owner: {e}") |
| 443 | return None |
| 444 | |
| 445 | def am_i_owner(self, channel_id): |
| 446 | """Check if this worker is the owner of the channel""" |
no test coverage detected