Synnefo » Pithos MS Client

using System;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

namespace Pithos.Core

{

    public class Agent<TMessage> : IDisposable

    {        

        private readonly BlockingCollection<TMessage> _messages = new BlockingCollection<TMessage>();

        private readonly CancellationTokenSource _cancelSource = new CancellationTokenSource();

        public CancellationToken CancellationToken;

        private readonly Action<Agent<TMessage>> _process;

        public Agent(Action<Agent<TMessage>> action)

        {

            _process = action;

            CancellationToken = _cancelSource.Token;

        }

        public void Post(TMessage message)

        {

            _messages.Add(message);

        }

        public Task<TMessage> Receive(int timeout = -1)

        {

            return Task<TMessage>.Factory.StartNew(() =>

            {

                TMessage item;

                if (!_messages.TryTake(out item, timeout, CancellationToken))

                    throw new TimeoutException();

                return item;

            });

        }

        public Task<TMessage> TryReceive(int timeout = -1)

        {

            return Task<TMessage>.Factory.StartNew(() =>

            {

                TMessage item;

                _messages.TryTake(out item, timeout, CancellationToken);

                return item;

            });

        }

        public void Start()

        {

            Task.Factory.StartNew(() => _process(this), CancellationToken);            

        }

        public static Agent<TMessage> Start(Action<Agent<TMessage>> action)

        {

            var agent = new Agent<TMessage>(action);

            agent.Start();

            return agent;

        }

        public void Stop()

        {

            _messages.CompleteAdding();

            _cancelSource.Cancel();

        }

        public void DoAsync(Action action)

        {

            Task.Factory.StartNew(action, CancellationToken);

        }

        ~Agent()

        {

            Dispose(false);

        }

        public void Dispose()

        {

            Dispose(true);

            GC.SuppressFinalize(this);

        }

        protected void Dispose(bool disposing)

        {

            if (disposing)

            {

                Stop();

                _messages.Dispose();

                _cancelSource.Dispose();

            }

        }

/*

        public void AddFromEnumerable(IEnumerable<TMessage> enumerable)

        {

            foreach (var message in enumerable)

            {

                Post(message);

            }

        }

*/

        public IEnumerable<TMessage> GetEnumerable()

        {

            return _messages;

        }

        public Task LoopAsync(Task process, Action loop,Action<Exception> onError=null)

        {

            return process.ContinueWith(t =>

            {   

                Task.Factory.StartNew(loop, CancellationToken);

                if (t.IsFaulted)

                {

                    var ex = t.Exception.InnerException;

                    if (ex is OperationCanceledException)

                        Stop();

                    if (onError != null)

                        onError(ex);

                }

            });

        }

    }

}