|
- # -*- coding: utf-8 -*-
- from threading import Thread
- import time
- from traceback import format_exc
-
- from loguru import logger
-
- from enums.StatusEnum import PushStreamStatus
- from exception.CustomerException import ServiceException
- from util.PushStreamUtils import PushStreamUtil
-
-
- class PushSteamThread(Thread):
- __slots__ = ("pushStreamUtil", "requestId", "videoId", "status", "ex")
-
- def __init__(self, pullUrl, pushUrl, requestId, videoId):
- super().__init__()
- self.pushStreamUtil = PushStreamUtil(pullUrl, pushUrl, requestId)
- self.requestId = requestId
- self.videoId = videoId
- self.status = True
- self.excute_status = PushStreamStatus.WAITING.value[0]
- self.ex = None
-
- def run(self):
- logger.info("开始启动推流线程, 视频id: {}, requestId:{}", self.videoId, self.requestId)
- while True:
- try:
- self.pushStreamUtil.start_push_stream()
- self.excute_status = PushStreamStatus.RUNNING.value[0]
- out, err = self.pushStreamUtil.push_stream_sp.communicate()
- # 异常断流
- if self.status:
- logger.warning("推流异常，请检测拉流地址和推流地址是否正常！")
- if self.pushStreamUtil.push_stream_sp.returncode != 0:
- logger.error("推流异常:{}, 视频id: {}, requestId:{}", err.decode(), self.videoId,
- self.requestId)
- self.excute_status = PushStreamStatus.RETRYING.value[0]
- self.pushStreamUtil.close_push_stream_sp()
- time.sleep(5)
- # 手动断流
- if not self.status:
- self.pushStreamUtil.close_push_stream_sp()
- break
- except ServiceException as s:
- logger.error("异常: {}, 视频id: {}, requestId:{}", s.msg, self.videoId, self.requestId)
- self.pushStreamUtil.close_push_stream_sp()
- self.ex = s
- break
- except Exception as e:
- logger.error("异常：{}, 视频id: {}, requestId:{}", format_exc(), self.videoId, self.requestId)
- self.pushStreamUtil.close_push_stream_sp()
- self.ex = e
- break
- logger.info("结束推流线程, 视频id: {}, requestId:{}", self.videoId, self.requestId)
|
