ObservablePriorityQueue<T> Implementation

Question

I have a requirement in my current project that will need a Prioritised Queue that supports the IObservable interface. Please notify me of any problems with the implementation that I currently have:

ObservablePriorityQueue<T>

public sealed class ObservablePriorityQueue<T> : IQueue<T>, IObservable<T> where T : IPrioritised
{
    #region "IObservable<T> Implementation"
    // A list of the subscribers for the IObservable implementation
    List<IObserver<T>> _subscribers = new List<IObserver<T>>(10);

    Object _observableSyncLock = new Object();

    #region "Interface specific"
    public IDisposable Subscribe(IObserver<T> observer)
    {
        if (observer == null)
        {
            throw new ArgumentNullException("The observer cannot be null.");
        }

        if (!_subscribers.Contains(observer))
        {
            lock (_observableSyncLock)
            {
                _subscribers.Add(observer);
            }
        }

        return new Disposable(() =>
        {
            this.Unsubscribe(observer);
        });
    }
    #endregion

    public void Unsubscribe(IObserver<T> observer)
    {
        if (observer == null)
        {
            throw new ArgumentNullException("The observer cannot be null.");
        }

        observer.OnCompleted();

        if (_subscribers.Contains(observer))
        {
            // remove the entry, but don't dispose it just 
            // in case they want to re-subscribe with the same observer later
            lock (_observableSyncLock)
            {
                _subscribers.Remove(observer);
            }
        }
    }
    #endregion

    #region "IQueue<T> Implementation"

    readonly List<T> _data = new List<T>(100);
    Object _queueSyncLock = new Object();

    public void Enqueue(T value)
    {
        if (value == null)
        {
            throw new ArgumentException("The item to be enqueued cannot be null");
        }

        lock (_queueSyncLock)
        {
            _data.Add(value);
        }

        // now that the entry has been added, notify everyone
        Task.Run(() =>
        {
            lock (_observableSyncLock)
            {
                foreach (IObserver<T> subscriber in _subscribers)
                {
                    subscriber.OnNext(value);
                }
            }
        });
    }

    public T Dequeue()
    {
        if (_data.Count > 0)
        {
            lock (_queueSyncLock)
            {
                var result = _data.OrderByDescending(element => element.Priority).ThenBy(element => element.TimeStamp).First();
                _data.Remove(result);
                return result;
            }
        }
        return default(T);
    }

    public T Peek()
    {
        if (_data.Count > 0)
        {
            return _data.OrderByDescending(element => element.Priority).ThenBy(element => element.TimeStamp).First();
        }
        return default(T);
    }

    public Int32 Count { get { return _data.Count; } }
    #endregion
}

IQueue<T>

public interface IQueue<T>
{

    void Enqueue(T value);
    T Dequeue();
    T Peek();

    Int32 Count { get; }
}

IPrioritised<T>

public interface IPrioritised
{
    QueuePriority Priority { get; }
    DateTime TimeStamp { get; }
}

QueuePriority

public enum QueuePriority
{
    None = 0,
    Lowest = 1,
    Low = 2,
    Normal = 3,
    High = 4,
    Highest = 5
}

Disposable

public sealed class Disposable : IDisposable
{
    readonly Action _action;
    public Disposable(Action action)
    {
        _action = action;
    }
    public void Dispose()
    {
        if (_action != null)
        {
            _action();
        }
    }
}

Answer 1

1. At first, I was confused what exactly did the implementation of IObservable<T> mean. You should properly document that.
2. You don't need to have a separate lock objects. If you're locking on a specific object, I think you should use that object in the lock.
3. If you're just calling a single method in your lambda, you can use more succinct syntax: new Disposable(() => this.Unsubscribe(observer)).
4. If you want to make your code thread-safe, you need to lock all your reads too. In Unsubscribe() you don't do that, which means someone could write to the list while you're reading it. The same problem is in Peek().
5. Dequeue() shouldn't return default(T) if the queue is empty. This can be problematic especially with value types. Instead, you should have a method like bool TryDequeue(out T result).
6. You're accessing Count outside of a lock. I wouldn't rely on the fact that doing this is safe, I think you should lock before accessing it too.
7. Since you're using Rx, you might as well use their Disposable.Create() instead of creating your own.

Answer 2

In addition to svick's points, I noticed that in your Enqueue method you're checking queued values for null:

public void Enqueue(T value)
{
    if (value == null)
    {
        throw new ArgumentException("The item to be enqueued cannot be null");
    }

However, there is no constraint on the generic type, T. If the user intends to queue value types (such as structs), this will fail.