ProducerConsumerQueue.cs

ProducerConsumerQueue.cs

   public class ProducerConsumerQueue : IDisposable
    {
      /// <summary>
      /// thread blocking collection 
      /// </summary>
      BlockingCollection<WorkItem> taskCollection = new BlockingCollection<WorkItem>();


      /// <summary>
      /// task continuations and 
      /// </summary>
      class WorkItem
      {
        public readonly TaskCompletionSource<object> TaskSource;
        public readonly Action Action;
        public readonly CancellationToken? CancelToken;

        public WorkItem( TaskCompletionSource<object> taskSource, Action action, CancellationToken? cancelToken = null)
        {
          TaskSource = taskSource;
          Action = action;
          CancelToken = cancelToken ?? CancellationToken.None;
        }
      }

     
      private ProducerConsumerQueue( int workerCount )
      {
        // Create and start a separate Task for each consumer:
        for (int i = 0; i < workerCount; i++) Task.Run(() => Consume());
      }

      public void Dispose() { taskCollection.CompleteAdding(); }

      public Task EnqueueTask( Action action_in )
      {
        return EnqueueTask(action_in, null);
      }

      /// <summary>
      /// background the messaging handler from intense work
      /// </summary>
      /// <param name="action"></param>
      /// <param name="cancelToken"></param>
      /// <returns></returns>
      public Task EnqueueTask( Action action, CancellationToken? cancelToken )
      {
        TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
        taskCollection.Add(new WorkItem(tcs, action, cancelToken));
        return tcs.Task;
      }

      /// <summary>
      /// the separate thread of tasks consumer
      /// </summary>
      void Consume()
      {
        foreach (WorkItem item in taskCollection.GetConsumingEnumerable())
        {
          if (item.CancelToken.HasValue && item.CancelToken.Value.IsCancellationRequested)
          {
            item.TaskSource.SetCanceled();
          }
          else
          {
            try
            {
              item.Action();
              item.TaskSource.SetResult(null);   // Indicate completion
            }
            catch (OperationCanceledException ex)
            {
              if (ex.CancellationToken == item.CancelToken)
              {
                item.TaskSource.SetCanceled();
              }
              else
              {
                item.TaskSource.SetException(ex);
              }
            }
              // ReSharper disable once CatchAllClause
            catch (Exception ex)
            {
              item.TaskSource.SetException(ex);
            }
          }
        }
      }
    }