|
- # -*- coding: utf-8 -*-
- import time
- from traceback import format_exc
-
- from multiprocessing import Process, Queue
-
- from loguru import logger
-
- from concurrency.PushStreamThread import PushSteamThread
- from enums.StatusEnum import PushStreamStatus, ExecuteStatus
- from util.LogUtils import init_log
-
- from enums.ExceptionEnum import ExceptionType
- from entity.FeedBack import pull_stream_feedback
- from exception.CustomerException import ServiceException
- from util.QueUtil import get_no_block_queue
-
-
- class PushStreamProcess(Process):
- __slots__ = [
- '_fbQueue',
- '_eventQueue',
- '_context',
- '_msg',
- '_analysisType',
- '_base_dir'
- ]
-
- def __init__(self, param):
- super().__init__()
- self._fbQueue = param.fbqueue
- self.eventQueue = Queue()
- self._context = param.context
- self._msg = param.msg
- self._analysisType = param.analyse_type
- self._base_dir = param.base_dir
-
- def sendEvent(self, eBody):
- self.eventQueue.put(eBody)
-
- # 获取下一个事件
- def getEvent(self):
- return get_no_block_queue(self.eventQueue)
-
- # 推送执行结果
- def sendResult(self, result):
- self._fbQueue.put(result)
-
- def sendhbMessage(self, status, videoIds, error_code="", error_msg=""):
- self.sendResult({"pull_stream": pull_stream_feedback(self._msg.get("requestId"),
- status,
- error_code,
- error_msg, videoIds)})
-
- def run(self):
- task = {}
- videoStatus = {}
- requestId = self._msg.get("requestId")
- try:
- # 程序开始时间
- init_log(self._base_dir)
- videoInfo = [{"id": url.get("id"), "status": PushStreamStatus.WAITING.value[0]} for url in
- self._msg.get("videoUrls", []) if url.get("id")]
- self.sendResult({"pull_stream": pull_stream_feedback(requestId, ExecuteStatus.WAITING.value[0], "", "",
- videoInfo)})
- videoUrls = self._msg.get("videoUrls")
- requestId = self._msg.get("requestId")
- if videoUrls is None or len(videoUrls) == 0:
- raise ServiceException(ExceptionType.PUSH_STREAM_URL_IS_NULL.value[0],
- ExceptionType.PUSH_STREAM_URL_IS_NULL.value[1])
- if len(videoUrls) > 5:
- logger.error("推流数量超过限制, 当前推流数量: {}, requestId:{}", len(videoUrls), requestId)
- raise ServiceException(ExceptionType.PULL_STREAM_NUM_LIMIT_EXCEPTION.value[0],
- ExceptionType.PULL_STREAM_NUM_LIMIT_EXCEPTION.value[1])
- for videoUrl in videoUrls:
- pushThread = PushSteamThread(videoUrl["pullUrl"], videoUrl["pushUrl"], requestId, videoUrl["id"])
- pushThread.start()
- task[videoUrl["id"]] = pushThread
- enable_time = time.time()
- for video in videoInfo:
- videoStatus[video.get("id")] = video.get("status")
- count = 0
- while True:
- # 整个推流任务超时时间
- if time.time() - enable_time > 43200:
- logger.error("任务执行超时, requestId:{}", requestId)
- for t in list(task.keys()):
- if task[t].is_alive():
- task[t].pushStreamUtil.status = False
- task[t].pushStreamUtil.close_push_stream_sp()
- task[t].join(120)
- videoStatus[t] = PushStreamStatus.TIMEOUT.value[0]
- videoInfo_timeout = [{"id": k, "status": v} for k, v in videoStatus.items()]
- self.sendResult({"pull_stream": pull_stream_feedback(requestId,
- ExecuteStatus.TIMEOUT.value[0],
- ExceptionType.TASK_EXCUTE_TIMEOUT_EXCEPTION.value[
- 0],
- ExceptionType.TASK_EXCUTE_TIMEOUT_EXCEPTION.value[
- 1],
- videoInfo_timeout)})
- break
- # 接受停止指令
- event_result = get_no_block_queue(self.eventQueue)
- if event_result is not None:
- command = event_result.get("command")
- videoIds = event_result.get("videoIds")
- if "stop" == command:
- # 如果videoIds是空停止所有任务
- if videoIds is None or len(videoIds) == 0:
- logger.info("停止所有执行的推流任务, requestId:{}", requestId)
- for t in list(task.keys()):
- if task[t].is_alive():
- task[t].pushStreamUtil.status = False
- task[t].pushStreamUtil.close_push_stream_sp()
- task[t].join(120)
- videoStatus[t] = PushStreamStatus.SUCCESS.value[0]
- videoInfo_sucess = [{"id": k, "status": v} for k, v in videoStatus.items()]
- self.sendResult({"pull_stream": pull_stream_feedback(requestId,
- ExecuteStatus.SUCCESS.value[0],
- "",
- "",
- videoInfo_sucess)})
- break
- else:
- logger.info("停止指定的推流任务, requestId:{}", requestId)
- alive_thread = 0
- for t in list(task.keys()):
- if task[t].is_alive():
- if t in videoIds:
- task[t].pushStreamUtil.status = False
- task[t].pushStreamUtil.close_push_stream_sp()
- task[t].join(120)
- videoStatus[t] = PushStreamStatus.SUCCESS.value[0]
- else:
- alive_thread += 1
- if alive_thread == 0:
- videoInfo_sucess = [{"id": k, "status": v} for k, v in videoStatus.items()]
- self.sendResult({"pull_stream": pull_stream_feedback(requestId,
- ExecuteStatus.SUCCESS.value[0],
- "",
- "",
- videoInfo_sucess)})
- break
- for t in list(task.keys()):
- if task[t].pushStreamUtil.status and not task[t].is_alive():
- logger.error("检测到推流线程异常停止！videoId:{}, requestId:{}", t, requestId)
- raise Exception("检测到推流线程异常停止！")
- statusQueue = get_no_block_queue(task[t].statusQueue)
- if statusQueue is not None and statusQueue[0] == 1:
- logger.error("推流任务异常: videoId:{}, requestId:{}", t, requestId)
- task[t].join(timeout=120)
- videoStatus[t] = PushStreamStatus.FAILED.value[0]
- raise statusQueue[1]
- if task[t].is_alive() and statusQueue is not None and statusQueue[0] == 2:
- if PushStreamStatus.RETRYING.value[0] == statusQueue[1]:
- videoStatus[t] = PushStreamStatus.RETRYING.value[0]
- task[t].pushStreamUtil.start_time = time.time()
- if task[t].is_alive() and time.time() - task[t].pushStreamUtil.start_time > 21:
- videoStatus[t] = PushStreamStatus.RUNNING.value[0]
- if count % 10 == 0:
- videoInfo_hb = [{"id": k, "status": v} for k, v in videoStatus.items()]
- self.sendResult({"pull_stream": pull_stream_feedback(requestId,
- ExecuteStatus.RUNNING.value[0],
- "",
- "",
- videoInfo_hb)})
- count = 0
- count += 1
- time.sleep(1)
- except ServiceException as s:
- logger.error("服务异常，异常编号:{}, 异常描述：{}, requestId: {}", s.code, s.msg, requestId)
- for t in list(task.keys()):
- if task[t].is_alive():
- task[t].pushStreamUtil.status = False
- task[t].pushStreamUtil.close_push_stream_sp()
- task[t].join(120)
- videoStatus[t] = PushStreamStatus.FAILED.value[0]
- videoInfo_ex = [{"id": k, "status": v} for k, v in videoStatus.items()]
- self.sendResult({"pull_stream": pull_stream_feedback(requestId,
- ExecuteStatus.FAILED.value[0],
- s.code,
- s.msg,
- videoInfo_ex)})
- except Exception:
- logger.error("服务异常: {}, requestId: {},", format_exc(), requestId)
- for t in list(task.keys()):
- if task[t].is_alive():
- task[t].pushStreamUtil.status = False
- task[t].pushStreamUtil.close_push_stream_sp()
- task[t].join(120)
- videoStatus[t] = PushStreamStatus.FAILED.value[0]
- videoInfo_ex = [{"id": k, "status": v} for k, v in videoStatus.items()]
- self.sendResult({"pull_stream": pull_stream_feedback(requestId,
- ExecuteStatus.FAILED.value[0],
- ExceptionType.SERVICE_INNER_EXCEPTION.value[0],
- ExceptionType.SERVICE_INNER_EXCEPTION.value[1],
- videoInfo_ex)})
- finally:
- if len(task) > 0:
- for t in list(task.keys()):
- if task[t].is_alive():
- task[t].pushStreamUtil.status = False
- task[t].pushStreamUtil.close_push_stream_sp()
- task[t].join(120)
- logger.info("推流任务完成, requestId: {}", requestId)
|
