We’re rewarding the question askers & reputations are being recalculated! Read more.
7 added 842 characters in body
source | link

As per comments this doesn't have As per comments this doesn't have the desired effects either. The best thing to do that occurred to me would be then to use an ObservableCollection. This would be a reliable to get hold of all added items. We might need to do something different regarding completing the work of a queue serially though.

Putting the desired effects either. The best thing to do that occurred to me would be then to useitems into an ObservableCollection. Thisobservable queue would beleak memory like the OP suggested. Hence we can go for a reliable to get hold of all added itemssolution where we simply have an event. We might need to do something different regarding completing the work of a queue serially thoughcan also create an Observable from that event.

Per instance, we could try So there's little change from my last previous solution (if you include the suggestion I made about using the observable collection add event to acquire and release a mutex per each item such asget an observable).

public class ParallelTaskQueueRx<TReturnValue>DataEventArgs<T> : IParallelTaskQueueRx<TReturnValue>EventArgs
{
    privatepublic ParallelTaskQueueRxDataEventArgs(T data) {
        ObservableResultsData = new ObservableCollection<TReturnValue>();data;
    }
 
    public ObservableCollection<TReturnValue>T ObservableResultsData { get; private set; }
}

public class ParallelTaskQueueRx<TReturnValue>
{
    public event EventHandler<DataEventArgs<TReturnValue>> ItemProcessed;

    public IObservable<TReturnValue> ObservableResults
    {
        get
        {
            return Observable.FromEventPattern<DataEventArgs<TReturnValue>>(
                h=> ItemProcessed += h,
                h => ItemProcessed -= h
            ).Select(e => e.EventArgs.Data);
        }
    }

    private readonly IDictionary<string, Mutex>SemaphoreSlim> _queueDirectory =
        new ConcurrentDictionary<string, Mutex>SemaphoreSlim>();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            var mutexsemaphore = new MutexSemaphoreSlim(false);
           1, mutex.ReleaseMutex(1);
            _queueDirectory.Add(queueId, mutexsemaphore);
        }
        ThreadPool.QueueUserWorkItem(async ctx => {
            _queueDirectory[queueId].WaitOneWait();
            var result = await myTask();
            _queueDirectory[queueId].ReleaseMutexRelease();
            lock ItemProcessed?.Invoke(ObservableResults)
            {
              this, new ObservableResults.AddDataEventArgs<TReturnValue>(result));
            });
    }
}

I tested this one and it seemed to work just fine. Usage

var parallellqueue = new }ParallelTaskQueueRx<string>();
  parallellqueue.ObservableResults.ForEachAsync(valuue => }
}Console.WriteLine(value));

As per comments this doesn't have the desired effects either. The best thing to do that occurred to me would be then to use an ObservableCollection. This would be a reliable to get hold of all added items. We might need to do something different regarding completing the work of a queue serially though.

Per instance, we could try to acquire and release a mutex per each item such as

public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>
{
    private ParallelTaskQueueRx() {
        ObservableResults = new ObservableCollection<TReturnValue>();
    }
 
    public ObservableCollection<TReturnValue> ObservableResults { get; private set; }
    private readonly IDictionary<string, Mutex> _queueDirectory =
        new ConcurrentDictionary<string, Mutex>();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            var mutex = new Mutex(false);
            mutex.ReleaseMutex();
            _queueDirectory.Add(queueId, mutex);
        }
        ThreadPool.QueueUserWorkItem(async ctx => {
            _queueDirectory[queueId].WaitOne();
            var result = await myTask();
            _queueDirectory[queueId].ReleaseMutex();
            lock (ObservableResults)
            {
                ObservableResults.Add(result);
            }
        });
    }
}

As per comments this doesn't have the desired effects either. The best thing to do that occurred to me would be then to use an ObservableCollection. This would be a reliable to get hold of all added items. We might need to do something different regarding completing the work of a queue serially though.

Putting the items into an observable queue would leak memory like the OP suggested. Hence we can go for a solution where we simply have an event. We can also create an Observable from that event. So there's little change from my last previous solution (if you include the suggestion I made about using the observable collection add event to get an observable).

public class DataEventArgs<T> : EventArgs
{
    public DataEventArgs(T data) {
        Data = data;
    }
    public T Data { get; private set; }
}

public class ParallelTaskQueueRx<TReturnValue>
{
    public event EventHandler<DataEventArgs<TReturnValue>> ItemProcessed;

    public IObservable<TReturnValue> ObservableResults
    {
        get
        {
            return Observable.FromEventPattern<DataEventArgs<TReturnValue>>(
                h=> ItemProcessed += h,
                h => ItemProcessed -= h
            ).Select(e => e.EventArgs.Data);
        }
    }

    private readonly IDictionary<string, SemaphoreSlim> _queueDirectory =
        new ConcurrentDictionary<string, SemaphoreSlim>();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            var semaphore = new SemaphoreSlim(1, 1);
            _queueDirectory.Add(queueId, semaphore);
        }
        ThreadPool.QueueUserWorkItem(async ctx => {
            _queueDirectory[queueId].Wait();
            var result = await myTask();
            _queueDirectory[queueId].Release();
             ItemProcessed?.Invoke(this, new DataEventArgs<TReturnValue>(result));
        });
    }
}

I tested this one and it seemed to work just fine. Usage

var parallellqueue = new ParallelTaskQueueRx<string>();
parallellqueue.ObservableResults.ForEachAsync(valuue => Console.WriteLine(value));
6 added 226 characters in body
source | link
public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>
{
    private ParallelTaskQueueRx() {
        ObservableResults = new ObservableCollection<TReturnValue>();
    }

    public ObservableCollection<TReturnValue> ObservableResults { get; private set; }
    private readonly IDictionary<string, Mutex> _queueDirectory =
        new ConcurrentDictionary<string, Mutex>();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            var mutex = new Mutex(false);
            mutex.ReleaseMutex();
            _queueDirectory.Add(queueId, mutex);
        }
        ThreadPool.QueueUserWorkItem(async ctx => {
            _queueDirectory[queueId].WaitOne();
            ObservableResults.Add(var result = await myTask());
            _queueDirectory[queueId].ReleaseMutex();
            lock (ObservableResults)
            {
                ObservableResults.Add(result);
            }
        });
    }
}
public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>
{
    private ParallelTaskQueueRx() {
        ObservableResults = new ObservableCollection<TReturnValue>();
    }

    public ObservableCollection<TReturnValue> ObservableResults { get; private set; }
    private readonly IDictionary<string, Mutex> _queueDirectory =
        new ConcurrentDictionary<string, Mutex>();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            var mutex = new Mutex(false);
            mutex.ReleaseMutex();
            _queueDirectory.Add(queueId, mutex);
        }
        ThreadPool.QueueUserWorkItem(async ctx => {
            _queueDirectory[queueId].WaitOne();
            ObservableResults.Add(await myTask());
            _queueDirectory[queueId].ReleaseMutex();
        });
    }
}
public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>
{
    private ParallelTaskQueueRx() {
        ObservableResults = new ObservableCollection<TReturnValue>();
    }

    public ObservableCollection<TReturnValue> ObservableResults { get; private set; }
    private readonly IDictionary<string, Mutex> _queueDirectory =
        new ConcurrentDictionary<string, Mutex>();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            var mutex = new Mutex(false);
            mutex.ReleaseMutex();
            _queueDirectory.Add(queueId, mutex);
        }
        ThreadPool.QueueUserWorkItem(async ctx => {
            _queueDirectory[queueId].WaitOne();
            var result = await myTask();
            _queueDirectory[queueId].ReleaseMutex();
            lock (ObservableResults)
            {
                ObservableResults.Add(result);
            }
        });
    }
}
5 added 1534 characters in body
source | link

As per comments this doesn't have the desired effects either. The best thing to do that occurred to me would be then to use an ObservableCollection. This would be a reliable to get hold of all added items. We might need to do something different regarding completing the work of a queue serially though.

Per instance, we could try to acquire and release a mutex per each item such as

public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>
{
    private ParallelTaskQueueRx() {
        ObservableResults = new ObservableCollection<TReturnValue>();
    }

    public ObservableCollection<TReturnValue> ObservableResults { get; private set; }
    private readonly IDictionary<string, Mutex> _queueDirectory =
        new ConcurrentDictionary<string, Mutex>();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            var mutex = new Mutex(false);
            mutex.ReleaseMutex();
            _queueDirectory.Add(queueId, mutex);
        }
        ThreadPool.QueueUserWorkItem(async ctx => {
            _queueDirectory[queueId].WaitOne();
            ObservableResults.Add(await myTask());
            _queueDirectory[queueId].ReleaseMutex();
        });
    }
}

As per comments this doesn't have the desired effects either. The best thing to do that occurred to me would be then to use an ObservableCollection. This would be a reliable to get hold of all added items. We might need to do something different regarding completing the work of a queue serially though.

Per instance, we could try to acquire and release a mutex per each item such as

public class ParallelTaskQueueRx<TReturnValue> : IParallelTaskQueueRx<TReturnValue>
{
    private ParallelTaskQueueRx() {
        ObservableResults = new ObservableCollection<TReturnValue>();
    }

    public ObservableCollection<TReturnValue> ObservableResults { get; private set; }
    private readonly IDictionary<string, Mutex> _queueDirectory =
        new ConcurrentDictionary<string, Mutex>();

    public void ProcessTaskOnSpecificQueue(
        Func<Task<TReturnValue>> myTask,
        string queueId)
    {
        if (!_queueDirectory.ContainsKey(queueId))
        {
            var mutex = new Mutex(false);
            mutex.ReleaseMutex();
            _queueDirectory.Add(queueId, mutex);
        }
        ThreadPool.QueueUserWorkItem(async ctx => {
            _queueDirectory[queueId].WaitOne();
            ObservableResults.Add(await myTask());
            _queueDirectory[queueId].ReleaseMutex();
        });
    }
}
4 deleted 106 characters in body
source | link
3 Added a version without subject
source | link
2 added 2 characters in body
source | link
1
source | link