FISLib_Old/Vinno.vCloud.Common.FIS/LiveVideos/LiveVideoPusher.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using Vinno.FIS.TRTCClient.Common.Enum;
using Vinno.IUS.Common.Log;
using Vinno.IUS.Common.Network.Leaf;
using Vinno.IUS.Common.Network.Transfer;
using Vinno.vCloud.FIS.CrossPlatform.Common.Enum;
using Vinno.vCloud.FIS.CrossPlatform.Common.LiveVideo;
using Vinno.vCloud.FIS.CrossPlatform.Common.LiveVideo.Interface;
using Vinno.vCloud.Protocol.Infrastructures;
using Vinno.vCloud.Protocol.Messages.Client;
using Vinno.vCloud.Protocol.Messages.Live;

namespace Vinno.vCloud.Common.FIS.LiveVideos
{
    public abstract class LiveVideoPusher : IDisposable
    {
        private readonly object _locker = new object();
        private readonly object _currentDeviceInfolocker = new object();
        private readonly ConcurrentDictionary<EnumPusherType, Func<ILiveVideoPusherForSonopost>> _pusherCreations;
        private readonly ClientLeaf _leaf;
        private readonly string _terminalId;
        private readonly string _terminalName;
        private readonly List<CPVideoDeviceInfo> _currentVideoDeviceInfoList;

        private IExtendedData _currentExtendData;
        private EnumPusherType _currentPushType;
        private PushHeartRateKeeper _pushHeartRateKeeper;
        private bool _isPaused;

        protected ILiveVideoPusherForSonopost _pusher;

        /// <summary>
        /// 是否正在推流
        /// </summary>
        public bool IsPushing { get; private set; }

        public LiveVideoPusher(ClientLeaf leaf, string terminalId, string terminalName)
        {
            _leaf = leaf;
            _terminalId = terminalId;
            _terminalName = terminalName;
            _pusherCreations = new ConcurrentDictionary<EnumPusherType, Func<ILiveVideoPusherForSonopost>>();
            _currentVideoDeviceInfoList = new List<CPVideoDeviceInfo>();
        }

        public void UpdateCurrentVideoDeviceInfoList(List<CPVideoDeviceInfo> infos)
        {
            lock (_currentDeviceInfolocker)
            {
                _currentVideoDeviceInfoList.Clear();
                if (infos != null)
                {
                    foreach (var info in infos)
                    {
                        var item = info.Clone() as CPVideoDeviceInfo;
                        _currentVideoDeviceInfoList.Add(item);
                    }
                }
            }
        }

        /// <summary>
        /// 注册具体推流类
        /// </summary>
        /// <param name="type">推流类型</param>
        /// <param name="pusher">推流类的创建方法</param>
        public void RegisterPusher(EnumPusherType type, Func<ILiveVideoPusherForSonopost> pusher)
        {
            _pusherCreations.TryAdd(type, pusher);
        }

        public void LiveStateChanged(LiveEventArgs e)
        {
            try
            {
                if (e.IsLive && IsPushing)
                {
                    Logger.WriteLineError($"LiveVideoPusherV2 LiveStateChanged Error,it is already pushing.");
                    return;
                }
                if (e.IsLive)
                {
                    _currentPushType = ConvertToPusherType(e);
                    Logger.WriteLineInfo($"Current Push Type:{_currentPushType}");
                    InitPusher(_currentPushType);
                    StartPusher(e.ExtendedData);
                }
                else
                {
                    StopPusher(false);
                }
            }
            catch (Exception ex)
            {
                Logger.WriteLineError($"Receive Live State Error:{ex}");
            }
        }

        public void SetIsPaused(bool isPaused)
        {
            Logger.WriteLineInfo($"LiveVideoPusher Set IsPaused:{isPaused}");
            _isPaused = isPaused;
            if (isPaused)
            {
                StopPusher(isPaused);
            }
        }

        /// <summary>
        /// 开始推流
        /// </summary>
        /// <param name="pushParams"></param>
        protected virtual void StartPusher(IExtendedData pushParams)
        {
            try
            {
                lock (_locker)
                {
                    if (_isPaused)
                    {
                        Logger.WriteLineError("StartPusherError,it is paused");
                        return;
                    }
                    _pusher.ChannelStateChanged += OnChannelStateChanged;
                    var success = _pusher.StartPusher(pushParams, _currentVideoDeviceInfoList);
                    if (success)
                    {
                        Logger.WriteLineInfo("Start Pusher Success!");
                        _currentExtendData = pushParams;
                        IsPushing = true;
                        LiveVideoStatusChecker.Instance.IsPushing = true;
                        StartHeartRateKeeper(pushParams);
                    }
                    else
                    {
                        Logger.WriteLineError("Start Pusher Fail!");
                    }
                }
            }
            catch (Exception e)
            {
                Logger.WriteLineError($"Start Pusher Error:{e}");
            }
        }

        /// <summary>
        /// 结束推流
        /// </summary>
        protected virtual void StopPusher(bool isPaused = false)
        {
            try
            {
                lock (_locker)
                {
                    if (_pusher == null)
                        return;
                    _pusher.ChannelStateChanged -= OnChannelStateChanged;
                    var success = _pusher.StopPusher();
                    if (success)
                    {
                        Logger.WriteLineInfo("Stop Pusher Success!");
                        _pusher.Dispose();
                        _pusher = null;
                        IsPushing = false;
                        LiveVideoStatusChecker.Instance.IsPushing = false;
                        StopHeartRateKeeper();
                    }
                    else
                    {
                        Logger.WriteLineError("Stop Pusher Fail!");
                    }
                }
            }
            catch (Exception e)
            {
                Logger.WriteLineError($"Stop Pusher Error:{e}");
            }
        }

        /// <summary>
        /// 停止推流
        /// </summary>
        public void StopPusher()
        {
            StopPusher(false);
        }

        public void SetMute(bool isMute)
        {
            lock (_locker)
            {
                _pusher?.SetMute(isMute);
            }
        }

        /// <summary>
        /// 重新推流
        /// </summary>
        public void ReStartPusher()
        {
            if (IsPushing == false)
                return;
            Logger.WriteLineInfo("Restart Pusher");
            StopPusher(false);

            InitPusher(_currentPushType);
            StartPusher(_currentExtendData);
        }

        private void OnChannelStateChanged(object sender, ChannelStateEventArgs e)
        {
            try
            {
                UpdateChannelState(e.Category, e.State);
            }
            catch (Exception ex)
            {
                Logger.WriteLineError($"Update Channel State Error:{ex}");
            }
        }

        public void UpdateChannelState(EnumLiveChannelCategory category, EnumLiveStates state)
        {
            Logger.WriteLineInfo($"UpdateChannelState:{category},{state}");
            using (var request = MessagePool.GetMessage<UpdateTerminalMultiLiveStateRequest>())
            {
                request.TerminalId = _terminalId;
                request.LiveChannelName = ((LiveChannelCategory)category).ToString();
                request.State = (LiveStates)state;
                var result = _leaf.Send(request);
                var resultMsg = ResultMessage.Convert(result);
                if (resultMsg == CCR.OK)
                {
                    Logger.WriteLineInfo("UpdateChannelState Success!");
                }
                else
                {
                    Logger.WriteLineError("UpdateChannelState Fail!");
                }
            }
        }

        public void StartHeartRateKeeper(IExtendedData extendedData)
        {
            if (_pushHeartRateKeeper == null && !string.IsNullOrEmpty(extendedData?.HeartRateCode))
            {
                _pushHeartRateKeeper = new PushHeartRateKeeper(_terminalId, extendedData?.HeartRateCode, _leaf);
                _pushHeartRateKeeper.Offlined += OnOfflined;
                _pushHeartRateKeeper.Start();
                Logger.WriteLineInfo("Start Pusher HeartRate Keeper ");
            }
            else
            {
                Logger.WriteLineInfo($"Start Pusher HeartRate Keeper Fail,The ConnectionUrl is{extendedData?.HeartRateCode}");
            }
        }

        public void StopHeartRateKeeper()
        {
            if (_pushHeartRateKeeper != null)
            {
                _pushHeartRateKeeper.Offlined -= OnOfflined;
                _pushHeartRateKeeper.Stop();
                _pushHeartRateKeeper = null;
                Logger.WriteLineInfo("Stop Pusher HeartRate Keeper");
            }
        }

        private void OnOfflined(object sender, EventArgs e)
        {
            Logger.WriteLineInfo($"Pusher is offline,terminal id is :{_terminalId}");
        }

        private void InitPusher(EnumPusherType type)
        {
            lock (_locker)
            {
                if (_pusherCreations.TryGetValue(type, out var creation))
                {
                    if (_pusher != null)
                    {
                        StopPusher();
                    }
                    _pusher = creation.Invoke();
                }
                else
                {
                    throw new ArgumentException($"Not support Mode {type}");
                }
            }
        }

        private EnumPusherType ConvertToPusherType(LiveEventArgs liveEventArgs)
        {
            if (liveEventArgs.Protocol == EnumLiveProtocol.RTC)
            {
                if (liveEventArgs.PushMode == EnumLiveDataMode.MergeLive)
                {
                    var rtcExtendedData = liveEventArgs.ExtendedData as RtcExtendedData;
                    if (rtcExtendedData.UserInfos.Count() == 1)
                    {
                        return EnumPusherType.RtcSingle;
                    }
                    else
                    {
                        return EnumPusherType.RtcMerge;
                    }
                }
                else if (liveEventArgs.PushMode == EnumLiveDataMode.OnlyLive)
                {
                    return EnumPusherType.RtcMulti;
                }
            }
            else if (liveEventArgs.Protocol == EnumLiveProtocol.Rtmp)
            {
                if (liveEventArgs.PushMode == EnumLiveDataMode.MergeLive)
                {
                    var rtmpExtendedData = liveEventArgs.ExtendedData as RtmpExtendedData;
                    if (rtmpExtendedData.UserInfos.Count() == 1)
                    {
                        return EnumPusherType.RtmpSingle;
                    }
                    else
                    {
                        return EnumPusherType.RtmpMerge;
                    }
                }
                else if (liveEventArgs.PushMode == EnumLiveDataMode.OnlyLive)
                {
                    return EnumPusherType.RtmpMulti;
                }
            }
            throw new ArgumentException("Unknown Pusher type");
        }

        public void Dispose()
        {
            StopPusher(false);
            _pusherCreations.Clear();
        }
    }
}