ProtocolStack/ReceiveQueue.cs

using System;
using System.Threading;
using System.Collections;

namespace ProtocolStack
{
        public class ReceiveQueue : Queue
        {
                RemoteHostComms thisRemoteHost;
                Segment workingSegment;
                Thread processingThread;

                public ReceiveQueue(RemoteHostComms thisRemoteHost) : base() {
                        this.thisRemoteHost = thisRemoteHost;
                }

                public void EnqueueBlocking(Segment seg) {
                        lock(this) {
                                Enqueue(seg);
                                Monitor.Pulse(this);
                        }
                }
                
                public Segment DequeueBlocking() {
                        Segment result;
                        lock(this) {
                                while (Count == 0) {
                                        Monitor.Wait(this);
                                }
                                result = (Segment)Dequeue();
                                Monitor.Pulse(this);
                        }
                        return result;
                }

                public void Start() {
                        if (processingThread == null || !processingThread.IsAlive) {
                                processingThread = new Thread(new ThreadStart(ProcessQueue));
                                processingThread.Start();
                        }
                }

                public void Stop() {
                        if (processingThread.IsAlive) {
                                processingThread.Abort();
                                processingThread.Join();
                        }
                }


                void ProcessQueue() {
                        while(true) {
                                workingSegment = this.DequeueBlocking(); // blocks
                                // Discard any duplicate or old segments
                                if (thisRemoteHost.SeqManager.IsProcessable(workingSegment.Headers)) {
                                        ProcessSegmentHeaders(workingSegment.Headers, workingSegment.Data.DataLength);
                                        // send it for processing
                                        thisRemoteHost.MessageAssembler.Process(workingSegment);
                                } // else drop and noop;        
                        }
                }

                private void ProcessSegmentHeaders(SegmentHeaders myHeaders, uint dataLength) {
                        /* updates protocol sequence number state, issues NACKs for missing segments */
                        thisRemoteHost.SeqManager.ProcessReceivedSegmentHeaders(ref myHeaders);
                        /* update ACK state*/
                        thisRemoteHost.AckManager.ProcessReceivedSegmentHeaders(ref myHeaders);
                        /* retransmit if we need to */
                        thisRemoteHost.NackManager.ProcessReceivedSegmentHeaders(ref myHeaders);
                        /* make new rate calculations */
                        thisRemoteHost.RateManager.ProcessReceivedSegmentHeaders(ref myHeaders, dataLength);
                }
        }



}

---