Synnefo » Pithos MS Client

#region

/* -----------------------------------------------------------------------

 * <copyright file="Agent.cs" company="GRNet">

 * 

 * Copyright 2011-2012 GRNET S.A. All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or

 * without modification, are permitted provided that the following

 * conditions are met:

 *

 *   1. Redistributions of source code must retain the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer.

 *

 *   2. Redistributions in binary form must reproduce the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer in the documentation and/or other materials

 *      provided with the distribution.

 *

 *

 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

 * POSSIBILITY OF SUCH DAMAGE.

 *

 * The views and conclusions contained in the software and

 * documentation are those of the authors and should not be

 * interpreted as representing official policies, either expressed

 * or implied, of GRNET S.A.

 * </copyright>

 * -----------------------------------------------------------------------

 */

#endregion

using System;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Diagnostics.Contracts;

using System.Threading;

using System.Threading.Async;

using System.Threading.Tasks;

using Pithos.Core.Agents;

namespace Pithos.Core

{

    public class Agent<TMessage> : IDisposable

    {

        private readonly ConcurrentQueue<TMessage> _queue;

        //private readonly AsyncCollection<TMessage> _messages;

        private readonly AsyncCollection<TMessage> _messages;

        private readonly CancellationTokenSource _cancelSource = new CancellationTokenSource();

        public CancellationToken CancellationToken;

        private readonly Action<Agent<TMessage>> _process;

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

        {

            _queue=new ConcurrentQueue<TMessage>();

            _messages = new AsyncCollection<TMessage>(_queue);            

            _process = action;

            CancellationToken = _cancelSource.Token;

        }

        public bool IsEmpty

        {

            get { return _queue.IsEmpty; }

        }

        public void Post(TMessage message)

        {

            _messages.Add(message);

        }

        ConcurrentDictionary<TMessage,TaskCompletionSource<object>> _awaiters=new ConcurrentDictionary<TMessage,TaskCompletionSource<object>>();

        public Task PostAndAwait(TMessage message)

        {            

            var tcs = new TaskCompletionSource<object>();

            _awaiters[message] = tcs;

            Post(message);

            return tcs.Task;

        }

        /// <summary>

        /// Receives a message asynchronously, optionally with a timeout. Receive throws a TimeoutException if the timeout expires

        /// </summary>

        /// <returns>A Task that will return the message asynchronously</returns>

        public  Task<TMessage> Receive()

        {

            return _messages.Take();

        }

        public void NotifyComplete(TMessage message)

        {

            TaskCompletionSource<object> tcs;

            if (_awaiters.TryRemove(message,out tcs))

                tcs.SetResult(null);

        }

        /// <summary>

        /// Start the agent

        /// </summary>

        public void Start()

        {

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

        }

        /// <summary>

        /// Create and start a new agent for the specified type of message

        /// </summary>

        /// <param name="action">The message processing action</param>

        /// <returns>A started Agent</returns>

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

        {

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

            agent.Start();

            return agent;

        }

        /// <summary>

        /// Stops the agent 

        /// </summary>

        public void Stop()

        {

            //Stop the message queue

            //_messages.CompleteAdding();

            //And signal the cancellation

            _cancelSource.Cancel();

        }

        /// <summary>

        /// Execute an action asynchronously, using the agent's cancellation source

        /// </summary>

        /// <param name="action">The action to execute</param>

        public void DoAsync(Action action)

        {

            if(action==null)

                throw new ArgumentNullException("action");

            Contract.EndContractBlock();

            Task.Factory.StartNew(action, CancellationToken);

        }

        ~Agent()

        {

            Dispose(false);

        }

        public void Dispose()

        {

            Dispose(true);

            GC.SuppressFinalize(this);

        }

        protected void Dispose(bool disposing)

        {

            if (disposing)

            {

                Stop();

                _cancelSource.Dispose();

            }

        }

        public IEnumerable<TMessage> GetEnumerable()

        {

            return _queue;

        }

        /// <summary>

        /// Remove the first message that matches the predicate

        /// </summary>

        /// <param name="predicate">The condition to match</param>

        /// <remarks>Removes the first message that matches the predicate by dequeing all 

        /// messages and re-enqueing all except the first matching message</remarks>

        public void Remove(Func<TMessage,bool> predicate)

        {

            //Can this work? Dequeue all items 

            //and then enqueue everything except the filtered items

            _queue.RemoveFirst(predicate);

        }

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

        {

            if(process==null)

                throw new ArgumentNullException("process");

            if(loop==null)

                throw new ArgumentNullException("loop");

            Contract.EndContractBlock();

            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);

                }

            },CancellationToken);

        }

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

        {

            return process.ContinueWith(t =>

            {   

                //Spawn the Loop immediatelly

                Task.Factory.StartNew(loop,CancellationToken);

                //Then process possible exceptions

                if (t.IsFaulted)

                {

                    var ex = t.Exception.InnerException;

                    if (ex is OperationCanceledException)

                        Stop();

                    if (onError != null)

                        onError(ex);

                }

                return default(T);

            });

        }

    }

}