Take the 2-minute tour ×
Code Review Stack Exchange is a question and answer site for peer programmer code reviews. It's 100% free, no registration required.

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

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();

        lock (_observableSyncLock)
        {
            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
                _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;
            }
        }
        throw new InvalidOperationException("There are no entries in the queue in order to dequeue");
    }

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

    public Int32 Count
    {
        get
        {

            lock (_queueSyncLock)
            {
                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();
        }
    }
}

11/02/2013 14:25 - Updated with comments from svick

share|improve this question
add comment

1 Answer

  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.
share|improve this answer
    
Thanks for the comments. I will review these and fix accordingly. As a note, I am not using Rx everything is out of the BCL. –  Stuart Blackler Feb 11 '13 at 13:59
    
I have confirmed the need for seperate locks here: stackoverflow.com/questions/14813780/… –  Stuart Blackler Feb 11 '13 at 14:20
    
@StuartBlackler Well, I think IObservable<T> is much less useful without Rx, so you might consider using it. –  svick Feb 11 '13 at 14:40
1  
@StuartBlackler And you really don't need the separate lock objects, I have added my own answer to that SO question. –  svick Feb 11 '13 at 14:44
add comment

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

Not the answer you're looking for? Browse other questions tagged or ask your own question.