ProtocolStack/TransmissionQueue.cs

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

namespace ProtocolStack {
        public class TransmissionQueue : Queue {
                private Queue lowPriorityQueue;

                public TransmissionQueue() : base() {
                        lowPriorityQueue = new Queue();
                }

                public void ResetConnection() {
                        lock(this.SyncRoot) {
                                Clear();
                        }
                        lock (lowPriorityQueue.SyncRoot) {
                                lowPriorityQueue.Clear();
                        }
                }

                public void Enqueue(TransmissionQueueEntry tqe, bool isLowPriority) {
                        lock (this.SyncRoot) {
                                if (!isLowPriority) {
                                        base.Enqueue(tqe);
                                } else {
                                        lock(lowPriorityQueue.SyncRoot) {
                                                lowPriorityQueue.Enqueue(tqe);
                                                Monitor.PulseAll(lowPriorityQueue.SyncRoot);
                                        }
                                }
                                Monitor.PulseAll(this.SyncRoot);
                        }
                }
                
                public void Enqueue(TransmissionQueueEntry[] tqes, bool isLowPriority) {
                        lock(this.SyncRoot) {
                                if (!isLowPriority) {
                                        foreach (TransmissionQueueEntry tqe in tqes) {
                                                base.Enqueue(tqe);
                                        }
                                } else {
                                        lock(lowPriorityQueue.SyncRoot) {
                                                foreach (TransmissionQueueEntry tqe in tqes) {
                                                        lowPriorityQueue.Enqueue(tqe);
                                                }
                                                Monitor.PulseAll(lowPriorityQueue.SyncRoot);
                                        }
                                }
                                Monitor.PulseAll(this.SyncRoot);
                        }
                }

                public void EnqueueUnprotected(TransmissionQueueEntry[] tqes, bool isLowPriority) {
                        Queue target = (isLowPriority ? lowPriorityQueue : this);
                        foreach (TransmissionQueueEntry tqe in tqes) {
                                target.Enqueue(tqe);
                        }
                }

                public TransmissionQueueEntry DequeueBlocking() {
                        TransmissionQueueEntry result;
                        lock(this.SyncRoot) {
                                while (Count == 0 && lowPriorityQueue.Count == 0) {
                                        Monitor.Wait(this.SyncRoot);
                                }
                                if (Count != 0) {
                                        result = (TransmissionQueueEntry)base.Dequeue();
                                        //Console.WriteLine("Selected Normal queue. {0} items in main Q, {1} in low pri", Count, lowPriorityQueue.Count);
                                } else { // must be low priority queue that woke us up
                                        lock(lowPriorityQueue.SyncRoot) {
                                                result = (TransmissionQueueEntry)lowPriorityQueue.Dequeue();
                                        //      Console.WriteLine("Selected low priority queue. {0} items in main Q, {1} in low pri", Count, lowPriorityQueue.Count);
                                                Monitor.PulseAll(lowPriorityQueue.SyncRoot);
                                        }
                                }
                                Monitor.PulseAll(this.SyncRoot);
                        }
                        return result;
                }


                public bool IsEmpty {
                        get { 
                                return (Count == 0); 
                        }
                }
        }

        /*********************************************************************************/

        public class TransmissionQueueEntry {
                public SegmentData segmentData;
                public ushort destinationDeviceID;
                public bool isMessageStart;
                public bool isMessageEnd;

                public TransmissionQueueEntry() {}

                public TransmissionQueueEntry(SegmentData segmentData, ushort destinationDeviceID, bool isMessageStart, bool isMessageEnd) {
                        this.segmentData = segmentData; // could have problems with not getting a copy here
                        this.destinationDeviceID = destinationDeviceID;
                        this.isMessageStart = isMessageStart;
                        this.isMessageEnd = isMessageEnd;
                }

                public static TransmissionQueueEntry[] FromMessage(Message msg) {
                        return TransmissionQueueEntry.FromMessage(msg, false);
                }

                public static TransmissionQueueEntry[] FromMessage(Message msg, bool singleSegmentOnly) {
                        /* this.ToSegmentData() 
                         * foreach(SegmentData in result) 
                         *              add TransmissonQueueEntry wrapper
                         *              place in result array
                         * */
                        SegmentData[] result = SegmentData.FromMessage(msg);
                        if (singleSegmentOnly && (result.Length != 1)) {
                                throw new ApplicationException("Single segment only was specified, and we have " + result.Length + " segments");
                        }
                        TransmissionQueueEntry[] tqes = new TransmissionQueueEntry[result.Length];
                        for(int i = 0; i < result.Length; i++) {
                                tqes[i] = new TransmissionQueueEntry(result[i], msg.Destination, i == 0, i == (result.Length -1));
                        }
                        return tqes;
                }
        }
}

---