This is a continuation of Parallel Task Queue that runs in sequence.Using Reactive Extensions I want to create a generic class a that process a task and executes it on a specific queue returning a TReturnValue on an observable stream.To specify which queue to be used to process the task, a queue id is specified.If two tasks are running on different queues (i.e. with different queue ids) then the two tasks should be run in parallel. Tasks on the same queue should be run in sequence.All tasks no matter which queue they run on, should return TReturnValue on the same one and only observable stream.A ConcurrentDirectory of BlockingCollections using the queue id as key is used to keep track of each queue.To prevent memory leaks, the BlockingCollection and Key is removed from the Directory when the queue is empty (i.e. all work has completed on the queue). Should new task emerge with the same queue id, a new BlockingCollection will be added again and put in the dictionary with the queue id as key.A Subject is used to create the observer for the TReturnValue's. I've read that Subjects should be used with care, and herein lies my key question:Is the following the right approach, specifically in terms of using a Subject? Will I run into any threading/blocking issues with this approach? Any other considerations?Interface:public interface IParallelTaskQueueRx<TReturnValue> : IDisposable{    IObservable<TReturnValue> ObservableResults { get; }    void ProcessTaskOnSpecificQueue(        Func<Task<TReturnValue>> myTask,        string queueId);}Implementation:public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>{    private readonly ISubject<TReturnValue> _subjectReturnValue = new Subject<TReturnValue>();    private IObserver<TReturnValue> ObserverReturnValue => _subjectReturnValue.AsObserver();    private readonly IDictionary<string, BlockingCollection<Func<Task<TReturnValue>>>> _queueDirectory =        new ConcurrentDictionary<string, BlockingCollection<Func<Task<TReturnValue>>>>();    public IObservable<TReturnValue> ObservableResults => _subjectReturnValue.AsObservable();    public void ProcessTaskOnSpecificQueue(        Func<Task<TReturnValue>> myTask,         string queueId)    {        if (!_queueDirectory.ContainsKey(queueId))        {            Debug.WriteLine($Creating Queue: {queueId});            _queueDirectory.Add(queueId, new BlockingCollection<Func<Task<TReturnValue>>>());            _queueDirectory[queueId].Add(myTask);            ProcessQueue(queueId);        }        else        {            Debug.WriteLine($Adding value to queue: {queueId});            _queueDirectory[queueId].Add(myTask);        }    }    private void ProcessQueue(string queueId)    {        Task.Run(async () =>        {            while (_queueDirectory[queueId].Count > 0)            {                try                {                    var task = _queueDirectory[queueId].Take();                    var result = await task();                    lock (ObserverReturnValue)                    {                        ObserverReturnValue.OnNext(result);                    }                }                catch (Exception ex)                {                    lock (ObserverReturnValue)                    {                        ObserverReturnValue.OnError(ex);                    }                }            }            _queueDirectory[queueId].CompleteAdding();            lock (_queueDirectory)            {                _queueDirectory.Remove(queueId);            }            Debug.WriteLine($Removing Queue: {queueId});        });    }    public void Dispose()    {        _queueDirectory.Clear();        ObserverReturnValue.OnCompleted();    }}Simple test:class Program{    static void Main(string[] args)    {        var parallelTaskQueue = new ParallelTaskQueueRx<string>();        var disp = parallelTaskQueue.ObservableResults.Subscribe(            System.Console.WriteLine,            ex => {System.Console.WriteLine($Error: {ex.Message});},            () => {System.Console.WriteLine(Compleded);});        Start(parallelTaskQueue);        System.Console.ReadKey();    }    private static async void Start(ParallelTaskQueueRx<string> parallelTaskQueue)    {        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>            {                await Task.Delay(TimeSpan.FromSeconds(1));                return Queue1: #1 (1 sec delay);            },Queue1);        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>            {                await Task.Delay(TimeSpan.FromMilliseconds(250));                return Queue2: #1 (0,25 sec delay);            },Queue2);        System.Console.WriteLine(----Waiting 0,25 sec----);        await Task.Delay(TimeSpan.FromMilliseconds(250));        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>            {                await Task.Delay(TimeSpan.FromMilliseconds(250));                return Queue1: #2 (0,25 sec delay);            },Queue1);        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>            {                await Task.Delay(TimeSpan.FromMilliseconds(1250));                return Queue2 2 (1,25 sec delay);            },Queue2);        parallelTaskQueue.ProcessTaskOnSpecificQueue(() => Task.FromResult(Queue2 #3 (no delay)),Queue2);        parallelTaskQueue.ProcessTaskOnSpecificQueue(() => Task.FromResult(Queue2 #4 (no delay)),Queue2);        System.Console.WriteLine(----Waiting 5 sec----);        await Task.Delay(TimeSpan.FromSeconds(5));        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>            {                await Task.Delay(TimeSpan.FromSeconds(1));                return Queue1 #3 (1 sec delay);            },Queue1);        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>            {                await Task.Delay(TimeSpan.FromMilliseconds(250));                return Queue2 #5 (0,25 sec delay);            },Queue2);        System.Console.WriteLine(----Waiting 0,25 sec----);        await Task.Delay(TimeSpan.FromMilliseconds(250));        parallelTaskQueue.ProcessTaskOnSpecificQueue(async () =>            {                await Task.Delay(TimeSpan.FromMilliseconds(250));                return Queue1 #4 (0,25 sec delay);            },Queue1);    }}
Generic parallel task queue returning an observable sequence of return value
c#;system.reactive
Based on the great input received above, I've settled on the slightly modified solution below. The solution is still using a Subject to keep the Observable hot.The code is also on GitHub and have been released as a NuGet.In terms of using a Subject this article was helpful, although a bit long and complex to read: To Use Subject Or Not To Use Subject?public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>{    private readonly ISubject<TReturnValue> _subjectReturnValue = new Subject<TReturnValue>();    private IObserver<TReturnValue> ObserverReturnValue => _subjectReturnValue.AsObserver();    private readonly IDictionary<string, BlockingCollection<Func<Task<TReturnValue>>>> _queueDirectory =        new ConcurrentDictionary<string, BlockingCollection<Func<Task<TReturnValue>>>>();    public IObservable<TReturnValue> ObservableResults => _subjectReturnValue.AsObservable();    public void ProcessTaskOnSpecificQueue(        Func<Task<TReturnValue>> myTask,         string queueId)    {        if (!_queueDirectory.ContainsKey(queueId))        {            Debug.WriteLine($Creating Queue: {queueId});            _queueDirectory.Add(queueId, new BlockingCollection<Func<Task<TReturnValue>>>());            _queueDirectory[queueId].Add(myTask);            ProcessQueue(queueId);        }        else        {            Debug.WriteLine($Adding value to queue: {queueId});            _queueDirectory[queueId].Add(myTask);        }    }    private void ProcessQueue(string queueId)    {        Task.Run(async () =>        {            while (_queueDirectory[queueId].Count > 0)            {                var task = _queueDirectory[queueId].Take();                TReturnValue result;                try                {                    result = await task();                }                catch (Exception ex)                {                    lock (ObserverReturnValue)                    {                        ObserverReturnValue.OnError(ex);                        continue;                    }                }                lock (ObserverReturnValue)                {                    ObserverReturnValue.OnNext(result);                }            }            _queueDirectory[queueId].CompleteAdding();            _queueDirectory.Remove(queueId);            Debug.WriteLine($Removing Queue: {queueId});        });    }    public void Dispose()    {        _queueDirectory.Clear();        ObserverReturnValue.OnCompleted();    }}Some charateristcs of this implementations are, that if a new task arrives at a specific queue before the previous task has completed, then this new task is added to the queue and processed next. The queue is not deleted before all tasks on it have been processed. So, in situations where there are very few tasks coming in and the tasks are quickyly completed, a new queue is created with every new task. This adds a bit of overhead.On the other hand, if the number of tasks coming in are high and/or they take long time to process, then the queue is maintained and only created with the first task.These characteristics make this solution well suited for managing constant high-load and/or burst scenarios.