code.grnet.gr Git - pithos-ms-client/blob - trunk/Pithos.Core/JobQueue.cs

   1 // -----------------------------------------------------------------------
   2 // <copyright file="JobQueue.cs" company="Microsoft">
   3 // TODO: Update copyright text.
   4 // </copyright>
   5 // -----------------------------------------------------------------------
   6
   7 using System.Collections.Concurrent;
   8 using System.Threading;
   9 using System.Threading.Tasks;
  10
  11 namespace Pithos.Core
  12 {
  13     using System;
  14     using System.Collections.Generic;
  15     using System.Linq;
  16     using System.Text;
  17
  18     /// <summary>
  19     /// TODO: Update summary.
  20     /// </summary>
  21     public class JobQueue
  22     {
  23         private readonly BlockingCollection<Action> _statusUpdateQueue = new BlockingCollection<Action>();
  24         private CancellationToken _cancellationToken;
  25
  26
  27         public void Start(CancellationToken token)
  28         {
  29             _cancellationToken = token;
  30             Task.Factory.StartNew(ProcessUpdates, _cancellationToken);
  31         }
  32
  33         private void ProcessUpdates()
  34         {
  35             foreach (var action in _statusUpdateQueue.GetConsumingEnumerable())
  36             {
  37                 action();
  38             }
  39         }
  40
  41         public void Add(Action action)
  42         {
  43             _statusUpdateQueue.Add(action);
  44         }
  45
  46         public void Stop()
  47         {
  48             _statusUpdateQueue.CompleteAdding();
  49         }
  50
  51     }
  52
  53     public class JobAgent:Agent<Action>
  54     {
  55         protected JobAgent(Action<Agent<Action>>  action)
  56             :base(action)
  57         {
  58
  59         }
  60
  61         public static JobAgent Create()
  62         {
  63             return (JobAgent)Start(queue =>
  64             {
  65                 Action loop = null;
  66                 loop = () =>
  67                 {
  68                     var job = queue.Receive();
  69                     job.ContinueWith(t =>
  70                     {
  71                         t.IgnoreExceptions();
  72                         var action = job.Result;
  73                         action();
  74                         queue.DoAsync(loop);
  75                     });
  76                 };
  77                 loop();
  78             });
  79         }
  80
  81     }
  82 }