skyscraper8/skyscraper8/Dvb/DataBroadcasting/MultiprotocolEncapsulationDecoder.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net.NetworkInformation;
using System.Text;
using System.Threading.Tasks;
using log4net;
using skyscraper5.Ietf.Rfc768;
using skyscraper5.Mpeg2;
using skyscraper5.Skyscraper.IO;
using skyscraper8.Dvb.DataBroadcasting;
using skyscraper8.Skyscraper.Net.VirtualNetworks;

namespace skyscraper5.Dvb.DataBroadcasting
{
    class MultiprotocolEncapsulationDecoder : IPsiProcessor
    {
	    private static readonly ILog logger = LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType.Name);
        private const bool ENABLE_MULTITHREADING = false;

		public IMultiprotocolEncapsulationEventHandler EventHandler { get; }

        public MultiprotocolEncapsulationDecoder(IMultiprotocolEncapsulationEventHandler eventHandler)
        {
            EventHandler = eventHandler;
            restartRequired = true;
        }

        public void GatherPsi(PsiSection section, int sourcePid)
        {
            if (!ENABLE_MULTITHREADING)
            {
                GatherPsiEx(section, sourcePid);
            }
            else
            {
                if (_actionQueue == null)
                    _actionQueue = new Queue<Action>();

                lock (_actionQueue)
                {
                    _actionQueue.Enqueue(() => { GatherPsiEx(section, sourcePid); });
                }

                if (restartRequired)
                {
                    restartRequired = false;
                    _actionQueueThread = new Thread(SeperatedThreadFunction);
                    _actionQueueThread.Name = String.Format("MPE Decoder Thread for PID 0x{0:X4}", sourcePid);
                    _actionQueueThread.Start();
                }
            }
        }

        private void GatherPsiEx(PsiSection section, int sourcePid)
        {
			MemoryStream ms = new MemoryStream(section.GetDataCopy(), false);
			if (ms.Length < 12)
				return;
			byte tableId = ms.ReadUInt8();

			ushort readUInt16Be = ms.ReadUInt16BE();
			bool sectionSyntaxIndicator = (readUInt16Be & 0x8000) != 0;
			bool privateIndicator = (readUInt16Be & 0x4000) != 0;
			int sectionLength = readUInt16Be & 0x0fff;

			byte[] macAddress = new byte[6];
			macAddress[5] = ms.ReadUInt8();
			macAddress[4] = ms.ReadUInt8();

			byte readUInt8 = ms.ReadUInt8();
			int payloadScramblingControl = (readUInt8 & 0x30) >> 4;
			if (payloadScramblingControl != 0)
				return; //We don't know how to unscramble these.
			int addressScramblingControl = (readUInt8 & 0x0c) >> 2;
			bool llcSnap = (readUInt8 & 0x02) != 0;
			bool currentNextIndicator = (readUInt8 & 0x01) != 0;

			byte sectionNumber = ms.ReadUInt8();
			byte lastSectionNumber = ms.ReadUInt8();
			macAddress[3] = ms.ReadUInt8();
			macAddress[2] = ms.ReadUInt8();
			macAddress[1] = ms.ReadUInt8();
			macAddress[0] = ms.ReadUInt8();

			long length = ms.GetAvailableBytes() - 4;
			if (length <= 0)
			{
				return;
			}

			EnsureVirtualNetworkIdentifier(sourcePid);
			byte[] payload = ms.ReadBytes(length);
			if (llcSnap)
			{
				HandleLlcSnap(sourcePid, macAddress, payload);
			}
			else
			{
				HandleIpDatagram(sourcePid, payload);
			}
		}

        private Thread _actionQueueThread;
        private Queue<Action> _actionQueue;
        private bool restartRequired;
        private void SeperatedThreadFunction()
        {
            while (true)
            {
                int remainingElements = 0;
                lock (_actionQueue)
                {
                    remainingElements = _actionQueue.Count;
                }
                if (remainingElements == 0)
                {
                    break;
                }

                Action action = null;
                lock (_actionQueue)
                {
                    action = _actionQueue.Dequeue();
                }
                action();
            }
            restartRequired = true;
        }
        
        private bool announcedLlcTraffic;
        private void HandleLlcSnap(int sourcePid, byte[] macAddress, byte[] payload)
        {
	        if (!announcedLlcTraffic)
	        {
		        //logger.InfoFormat("Detected LLC/SNAP traffic on PID {0:x4}", sourcePid);
		        announcedLlcTraffic = true;
	        }

	        PhysicalAddress physicalAddress = new PhysicalAddress(macAddress);
	        EventHandler.OnLlcFrame(networkIdentifier, PhysicalAddress.None, physicalAddress, (ushort)payload.Length, payload);
        }
        
        private void HandleIpDatagram(int sourcePid, byte[] payload)
        {
            EventHandler.OnIpDatagram(networkIdentifier, payload);
        }

        private PidNetworkIdentifier networkIdentifier;
        private void EnsureVirtualNetworkIdentifier(int sourcePid)
        {
	        if (networkIdentifier == null)
	        {
		        networkIdentifier = new PidNetworkIdentifier((ushort)sourcePid);
	        }
	        else
	        {
		        if (networkIdentifier.PID != sourcePid)
		        {
			        throw new DataBroadcastingException("Re-using the same MPE Decoder for multiple PIDs is not supported yet.");
		        }
	        }
        }
    }
}