ProtocolStack/MessageAssembly.cs

using System;
using System.Collections;
using System.Collections.Specialized;
using System.Threading;
using Common;

namespace ProtocolStack
{
        /*************************************************************************/
        public class MessageAssembly
        {
                AssemblyLine assemblyLine; // segment storage and processing
                protected IncomingMessageQueue messageQueueOutput; // message destination
                protected StackInterface thisStack;
                ChunkAssembler chunkAssembler;
                Thread processingThread;






                public MessageAssembly(RemoteHostComms remoteHost) {
                        thisStack = remoteHost.thisStack;
                        assemblyLine = new AssemblyLine();
                        chunkAssembler = new ChunkAssembler();
                        messageQueueOutput = thisStack.ReceivedMessages;
                }

                public void Process(Segment s) {
                        // insert segment into assembly line
                        lock(assemblyLine.SyncRoot) {
                                assemblyLine.Add(s);
                                if (s.IsRetransmission) {
                                        Console.WriteLine("Just added retransmissoin {0} to assembly line", s.Headers.SequenceNumber);
                                }
                                Monitor.PulseAll(assemblyLine.SyncRoot);
                        }
                }

                public void Start() {
                        if (processingThread == null || !processingThread.IsAlive) {
                                processingThread = new Thread(new ThreadStart(Run));                            
                                assemblyLine.Clear();
                                chunkAssembler.Init();
                                processingThread.Start();
                        } else {
                                Console.Error.WriteLine("*** Error starting Message Assembly thread: Thread already started! ***");
                        }
                }

                public void Stop() {
                        if (processingThread != null && processingThread.IsAlive) {
                                processingThread.Abort();
                                processingThread.Join();
                        }
                }

                void Run() {
                        Console.WriteLine("Started Assembly line assembler");
                        Message msg = null;
                        try {
                                while(true) {
                                        /* check for completed assembly lines */
                                        msg = null;
                                        lock (assemblyLine.SyncRoot) {
                                                while (!assemblyLine.HoldsCompleteMessage) {
                                                        Monitor.Wait(assemblyLine.SyncRoot);
                                                }
                                                // syphon off the messages
                                                msg = assemblyLine.ExtractCompleteMessage();
                                                Monitor.PulseAll(assemblyLine.SyncRoot);
                                        }
                                        // Send this message somewhere for action.
                                        // check for retransmission message
                                        if (msg.Type == MessageType.Retransmission) {
                                                RetransmissionMessage retrans = (RetransmissionMessage)msg;
                                                Segment retransSegment = retrans.RawSegment;
                                                retransSegment.IsRetransmission = true;
                                                retransSegment.Headers.TimeReceived = DateTime.Now;
                                                Console.WriteLine("Just got retransmission of {0}. Adding to assembly. Retrans flag {1}", retrans.RawSegment.Headers.SequenceNumber, retrans.RawSegment.IsRetransmission);
                                                thisStack.GetRemoteHostByID(retransSegment.Headers.SourceDeviceID).AcceptIncomingSegment(retransSegment);
                                        } else if (msg.Type == MessageType.Chunk) {
                                                Console.WriteLine("  > Got Chunk");
                                                chunkAssembler.Add(msg);
                                                while (chunkAssembler.ContainsCompleteMessage()) {
                                                        Message reChunkedMsg = chunkAssembler.ExtractCompleteMessage();
                                                        if (reChunkedMsg != null) {     
                                                                Console.WriteLine(">>> Assembled Chunked Message");
                                                                messageQueueOutput.EnqueueBlocking(reChunkedMsg);
                                                        } else {
                                                                throw new ApplicationException("Extracted a Null chunked message");
                                                        }
                                                }
                                        } else {
                                                messageQueueOutput.EnqueueBlocking(msg);
                                        }
                                }
                        } catch (ThreadAbortException) {
                                Console.WriteLine("Shutting down message assembly");
                        }
                        
                }
        }


        /*************************************************************************/
        class AssemblyLine : SortedList {
                // TODO: Include comparator that deals correctly with wraparound




                private int startCount; 
                private int finishCount;




                public AssemblyLine() : base() {
                        startCount = 0;
                        finishCount = 0;
                }





                public void Add(Segment seg) {
                        this.Add(seg.Headers.SequenceNumber, seg);
                        if (seg.Headers.IsFirstInMessage) { startCount++; }
                        if (seg.Headers.IsLastInMessage) { finishCount++; }
                //      PrintContents();
                }


                public bool HoldsCompleteMessage {
                        get { 
                                int i, j = -1;
                                bool result = false;
                                if (startCount > 0  && finishCount > 0) { 
                                        /* if we have the same (non-zero) number of starts and finishes */

                                        this.IdentifyCompleteMessage(out i, out j);
                                        if (i == -1 || j == -1) { result = false; } else { result = true; }
                                }
                                return result;
                        }
                }

                public Message ExtractCompleteMessage() {
                        Message result;
                        int startIndex, finishIndex;
                        SegmentData[] segmentData;
                        uint[] seqNumbers;
                        // run through the list, looking for contigious sequencing betwen a start and finish
                        IdentifyCompleteMessage(out startIndex, out finishIndex);
                        segmentData = new SegmentData[finishIndex - startIndex + 1];
                        seqNumbers = new uint[finishIndex - startIndex + 1];
                        for (int i = startIndex; i <= finishIndex; i++) {
                                segmentData[i-startIndex] = ((Segment)this.GetByIndex(i)).Data;
                                seqNumbers[i-startIndex] = ((Segment)this.GetByIndex(i)).Headers.SequenceNumber;
                        }
                        foreach(uint seqNumber in seqNumbers) {
                                Remove(seqNumber);
                        }
                        result = CreateTypedMessage(segmentData);
                        startCount--;
                        finishCount--;
                        return result;
                }

                public void PrintContents() {
                        Console.Write("StartCount: {0}, FinishCouunt: {1}. Contents:", startCount, finishCount);
                        for (int i = 0; i < this.Count; i++) {
                                Segment s = (Segment)this.GetByIndex(i);
                                Console.Write("[{0}] {1} {2}{3},", i, s.Headers.SequenceNumber, s.Headers.IsFirstInMessage ? "F" : "_", s.Headers.IsLastInMessage ? "L" : "_");
                        }
                        Console.WriteLine();
                }
                private void IdentifyCompleteMessage(out int startIndex, out int finishIndex) {
                        // run through the list, looking for contigious sequencing betwen a start and finish segment;
                        startIndex = -1;
                        finishIndex = -1;
                        bool foundIt = false;
                        int index = 0;
                        int potentialStartIndex;
                        uint lastSeqSeen;
                        Segment currentSegment;

                        // scan through until we find a start node. 
                        // once we have found a start node, scan through until we get a gap or a finish node.
                        // if we hit a gap, find another start node
                        // if we hit a finish, return start and finish.
                        while (!foundIt && index < Count) {
                                currentSegment = (Segment)this.GetByIndex(index);
                                if (!currentSegment.Headers.IsFirstInMessage) {
                                        index++; // cycle through until we hit 1st start segment
                                } else { // we've hit a start segment @ index.
                                        potentialStartIndex = index;
                                        // quick special case test (kludge, could be more graceful)
                                        if (!currentSegment.Headers.IsLastInMessage) { // if we don't have a 1st and last same seg.
                                                // start off lastSeqSeen....
                                                lastSeqSeen = currentSegment.Headers.SequenceNumber;
                                                index++;
                                                if (index >= Count) { break; } // kludge
                                                currentSegment = (Segment)this.GetByIndex(index);
                                                while ((!currentSegment.Headers.IsLastInMessage) && (currentSegment.Headers.SequenceNumber == (lastSeqSeen +1))) {
                                                        // while we haven't hit a finish node, and there is not a gap in the sequence
                                                        lastSeqSeen = currentSegment.Headers.SequenceNumber;
                                                        index++;
                                                        if (index >= Count) { break; } // kludge
                                                        currentSegment = (Segment)this.GetByIndex(index);
                                                        // TODO: Exception handling if run off end of list.
                                                } 
                                                /* we need to check for a gap before we check for last in message, because we 
                                                 * could miss the penultimate segment */
                                                // here when hit finish node or segment muckup
                                                if (currentSegment.Headers.SequenceNumber == (lastSeqSeen + 1)) { // if we're still in sequence
                                                        // must have dropped out due to finding a finish
                                                        startIndex = potentialStartIndex;
                                                        finishIndex = index;
                                                        foundIt = true;
                                                } else {
                                                        Console.WriteLine("Gap In sequence detected");
                                                        //index++; /* we hit a gap, so we move on to the next one and fall out into the
                                                        //                * main while !foundIt loop */
                                                }
                                        } else { // we do have first and last sequence on same segment
                                                startIndex = potentialStartIndex;
                                                finishIndex = index;
                                                foundIt = true;
                                        }
                                }
                        }
                }       

                public Message CreateTypedMessage(SegmentData[] src) {
                        return CreateTypedMessageStatic(src);
                }

                public static Message CreateTypedMessageStatic(SegmentData[] src) {
                        byte[] data = SegmentData.ArrayToByteArray(src);
                        MessageType msgType = (MessageType)src[0].Data[0];
                        Message result;
                        switch(msgType) {
                                case MessageType.Noop: { result = new EmptyMessage(); break; }
                                case MessageType.Retransmission: { result = new RetransmissionMessage(data); break; }
                                case MessageType.CacheIndex: { result = new CacheIndexMessage(data); break; }
                                case MessageType.NoChange: { result = new NoChangeMessage(data); break; }
                                case MessageType.HTTPRequest: { result = new HTTPRequestMessage(data); break; }
                                case MessageType.HTTPResponse: { result = new HTTPResponseMessage(data); break; }
                                case MessageType.CacheUpdateHTTPResponse: { result = new CacheUpdateHTTPResponseMsg(data); break; }
                                case MessageType.CacheUpdateNoChange: {result = new CacheUpdateNoChangeMsg(data); break; }
                                case MessageType.Chunk: { result = new MessageChunk(data); break; }
                                case MessageType.CacheIndexRequest : { result = new CacheIndexRequestMessage(data); break; }
                                default: { throw new NotSupportedException(String.Format("Message type of {0} is not recognised", msgType)); }
                        }
                        return result;
                }
        }
        /*************************************************************************/
        public class ChunkAssembler {
                /*********************************************************************/
                class AssemblyContainer {
                        /*****************************************************************/
                        class ChunkedMessageAssembler {
                                private int first;
                                private int last;
                                private int count;
                                MessageChunk[] chunkArray;

                                public int ChunksOutstanding {
                                        get { return ((last - first) + 1) - count; }
                                }

                                public bool BelongsHere(MessageChunk chunk) {
                                        return (chunk.offset >= first && chunk.offset <= last);
                                }

                                public ChunkedMessageAssembler(MessageChunk initialChunk) {
                                        first = initialChunk.firstInMessage;
                                        last = initialChunk.lastInMessage;
                                        chunkArray = new MessageChunk[(last - first) + 1];
                                        count = 0;
                                        Add(initialChunk);
                                }

                                public void Add(MessageChunk chunk) {
                                        if (chunk.firstInMessage == first && chunk.lastInMessage == last) {
                                                if (chunkArray[chunk.offset - first] == null) { // -first to return to zero based.
                                                        chunkArray[chunk.offset - first] = chunk;
                                                        count++;
                                                } else {
                                                        throw new ApplicationException("Attempting to assign chunk to somewhere full");
                                                }
                                        } else {
                                                throw new ApplicationException("Attempting to add a chunk to a ChunkedMessageAssember that doesn't match first and last");
                                        }
                                }

                                public Message ExtractMessage() {
                                        if (ChunksOutstanding == 0) {
                                                return AssemblyLine.CreateTypedMessageStatic(MessageChunk.GetMessageFromChunks(chunkArray));
                                        } else {
                                                throw new ApplicationException("Still chunks outstanding");
                                        }
                                }

                        }
                        /*****************************************************************/
                        public DateTime timestamp;
                        private ArrayList chunkedMessageAssemblers;
                                                
                        public AssemblyContainer(DateTime timestamp) {
                                this.timestamp = timestamp;
                                chunkedMessageAssemblers = new ArrayList();
                        }

                        public void AddChunk(MessageChunk chunk) {
                                bool foundHome = false;
                                foreach(ChunkedMessageAssembler assembler in chunkedMessageAssemblers) {
                                        if (assembler.BelongsHere(chunk)) {
                                                assembler.Add(chunk);
                                        }
                                        foundHome = true;
                                        break;
                                }
                                if (!foundHome) {
                                        chunkedMessageAssemblers.Add(new ChunkedMessageAssembler(chunk));
                                }
                        }

                        public bool ContainsCompleteMessage() {
                                bool result = false;
                                foreach (ChunkedMessageAssembler chunkedAssembler in chunkedMessageAssemblers) {
                                        result = (chunkedAssembler.ChunksOutstanding == 0);
                                        if (result) { break; }
                                }
                                return result;
                        }

                        public Message ExtractCompleteMessage() {
                                Message result = null;
                                foreach (ChunkedMessageAssembler chunkedAssembler in chunkedMessageAssemblers) {
                                        if (chunkedAssembler.ChunksOutstanding == 0) {
                                                result = chunkedAssembler.ExtractMessage();
                                        }
                                        if (result != null) { 
                                                chunkedMessageAssemblers.Remove(chunkedAssembler);
                                                break; }
                                }
                                return result;
                        }
                }
                /*********************************************************************/
                private ListDictionary assemblyContainers;
                
                public ChunkAssembler() {
                        assemblyContainers = new ListDictionary();
                }


                public void Add(Message msg) {
                        if (msg.Type != MessageType.Chunk) {
                                throw new ApplicationException("Unable to assemble non-chunked message in message chunk assembly");
                        } else { // we do have a message chunk.
                                MessageChunk chunk = (MessageChunk)msg;
                                AssemblyContainer currentContainer;
                                if (assemblyContainers.Contains(chunk.timestamp)) {
                                        currentContainer = (AssemblyContainer)assemblyContainers[chunk.timestamp];
                                } else {
                                        currentContainer = new AssemblyContainer(chunk.timestamp);
                                        assemblyContainers.Add(chunk.timestamp, currentContainer);
                                }
                                currentContainer.AddChunk(chunk);
                        }
                }

                public void Init() {
                        assemblyContainers.Clear();
                }

                public bool ContainsCompleteMessage() {
                        bool result = false;
                        foreach (AssemblyContainer container in assemblyContainers.Values) {
                                result = container.ContainsCompleteMessage();
                                if (result) {break;}
                        }
                        return result;
                }

                public Message ExtractCompleteMessage() {
                        Message result = null;
                        foreach (AssemblyContainer container in assemblyContainers.Values) { 
                                result = container.ExtractCompleteMessage();
                                if (result != null) { break; }
                        }
                        return result;
                }


        }
}

---