(user_id, requesting_client_id, active_connections)
| 8 | |
| 9 | |
| 10 | def attempt_stream_termination(user_id, requesting_client_id, active_connections): |
| 11 | try: |
| 12 | logger.info("[stream limits]" f"[{requesting_client_id}] User {user_id} has {len(active_connections)} active connections, checking termination candidates") |
| 13 | |
| 14 | user_limit_settings = CoreSettings.get_user_limits_settings() |
| 15 | terminate_oldest = user_limit_settings.get("terminate_oldest", True) |
| 16 | prioritize_single = user_limit_settings.get("prioritize_single_client_channels", True) |
| 17 | ignore_same_channel = user_limit_settings.get("ignore_same_channel_connections", False) |
| 18 | |
| 19 | channel_counts = {} |
| 20 | for connection in active_connections: |
| 21 | media_id = connection['media_id'] |
| 22 | channel_counts[media_id] = channel_counts.get(media_id, 0) + 1 |
| 23 | |
| 24 | def prioritize(connection): |
| 25 | is_multi = channel_counts[connection['media_id']] > 1 |
| 26 | |
| 27 | # if we're ignoring same-channel connections, put them at the end |
| 28 | same_ch_key = 1 if (ignore_same_channel and is_multi) else 0 |
| 29 | |
| 30 | # key for prioritizing single-client channels |
| 31 | single_key = 0 if (prioritize_single and not is_multi) else 1 |
| 32 | |
| 33 | # sort by age setting |
| 34 | time_key = connection['connected_at'] if terminate_oldest else -connection['connected_at'] |
| 35 | |
| 36 | return (same_ch_key, single_key, time_key) |
| 37 | |
| 38 | termination_candidates = sorted(active_connections, key=prioritize) |
| 39 | |
| 40 | if not termination_candidates: |
| 41 | logger.warning("[stream limits]" f"[{requesting_client_id}] No termination candidates found for user {user_id}") |
| 42 | return False |
| 43 | |
| 44 | target = termination_candidates[0] |
| 45 | logger.info("[stream limits]" |
| 46 | f"[{requesting_client_id}] Terminating client {target['client_id']} " |
| 47 | f"on media {target['media_id']} (connected_at={target['connected_at']})" |
| 48 | ) |
| 49 | |
| 50 | # When counting by unique channel, freeing one connection from a multi-connection |
| 51 | # channel doesn't free a slot — terminate all connections to that channel so the |
| 52 | # unique-channel count actually drops by one. |
| 53 | targets = ( |
| 54 | [c for c in active_connections if c['media_id'] == target['media_id']] |
| 55 | if ignore_same_channel |
| 56 | else [target] |
| 57 | ) |
| 58 | |
| 59 | for t in targets: |
| 60 | if t['type'] == 'live': |
| 61 | result = ChannelService.stop_client(t['media_id'], t['client_id']) |
| 62 | if result.get("status") == "error": |
| 63 | logger.warning(f"[stream limits][{requesting_client_id}] Failed to stop client {t['client_id']} on channel {t['media_id']}") |
| 64 | else: |
| 65 | connection_manager = MultiWorkerVODConnectionManager.get_instance() |
| 66 | redis_client = connection_manager.redis_client |
| 67 |
no test coverage detected