Start an OutputProfileManager for this (channel, profile) pair if not already running. Only the TS-owning worker starts the process; all workers (including non-owners) get a StreamBuffer wired to the same Redis keys so they can serve clients. Returns True o
(self, channel_id, profile_id, command)
| 1269 | # ------------------------------------------------------------------ |
| 1270 | |
| 1271 | def ensure_output_profile(self, channel_id, profile_id, command) -> bool: |
| 1272 | """ |
| 1273 | Start an OutputProfileManager for this (channel, profile) pair if not |
| 1274 | already running. Only the TS-owning worker starts the process; all |
| 1275 | workers (including non-owners) get a StreamBuffer wired to the same |
| 1276 | Redis keys so they can serve clients. |
| 1277 | |
| 1278 | Returns True once the profile buffer is available in self.profile_buffers. |
| 1279 | """ |
| 1280 | channel_profiles = self.profile_buffers.get(channel_id, {}) |
| 1281 | logger.debug( |
| 1282 | f"[Profile:{profile_id}:{channel_id[:8]}] ensure_output_profile() called, " |
| 1283 | f"already_buffered={profile_id in channel_profiles} " |
| 1284 | f"is_owner={self.am_i_owner(channel_id)}" |
| 1285 | ) |
| 1286 | if profile_id in channel_profiles: |
| 1287 | existing = self.profile_managers.get(channel_id, {}).get(profile_id) |
| 1288 | if existing is not None: |
| 1289 | # Owner: verify the FFmpeg process is still running. |
| 1290 | if existing._process is not None and existing._process.poll() is None: |
| 1291 | return True |
| 1292 | logger.warning( |
| 1293 | f"[Profile:{profile_id}:{channel_id[:8]}] " |
| 1294 | "Transcode process exited, restarting" |
| 1295 | ) |
| 1296 | self.profile_managers.get(channel_id, {}).pop(profile_id, None) |
| 1297 | if not self.profile_managers.get(channel_id): |
| 1298 | self.profile_managers.pop(channel_id, None) |
| 1299 | self.profile_buffers.get(channel_id, {}).pop(profile_id, None) |
| 1300 | if not self.profile_buffers.get(channel_id): |
| 1301 | self.profile_buffers.pop(channel_id, None) |
| 1302 | existing.stop() |
| 1303 | else: |
| 1304 | # Non-owner reader buffer: verify the owner's state is still active. |
| 1305 | if not self.redis_client: |
| 1306 | return True |
| 1307 | state = self.redis_client.get( |
| 1308 | RedisKeys.output_state(channel_id, f"mpegts:p{profile_id}") |
| 1309 | ) |
| 1310 | if state == PROFILE_STATE_ACTIVE: |
| 1311 | return True |
| 1312 | logger.warning( |
| 1313 | f"[Profile:{profile_id}:{channel_id[:8]}] " |
| 1314 | "Reader buffer exists but profile state not active, resetting" |
| 1315 | ) |
| 1316 | self.profile_buffers.get(channel_id, {}).pop(profile_id, None) |
| 1317 | if not self.profile_buffers.get(channel_id): |
| 1318 | self.profile_buffers.pop(channel_id, None) |
| 1319 | |
| 1320 | if not self.redis_client: |
| 1321 | return False |
| 1322 | |
| 1323 | # Check if another worker already owns the transcode |
| 1324 | state = self.redis_client.get(RedisKeys.output_state(channel_id, f"mpegts:p{profile_id}")) |
| 1325 | if state == PROFILE_STATE_ACTIVE: |
| 1326 | owner_val = self.redis_client.get(RedisKeys.output_owner(channel_id, f"mpegts:p{profile_id}")) |
| 1327 | if owner_val and owner_val != self.worker_id: |
| 1328 | # Non-owner: create a reader buffer pointing at the same Redis keys |
no test coverage detected