Statistics
| Branch: | Revision:

root / trunk / Pithos.Core / Agents / Agent.cs @ 174bbb6e

History | View | Annotate | Download (7.5 kB)

1
#region
2
/* -----------------------------------------------------------------------
3
 * <copyright file="Agent.cs" company="GRNet">
4
 * 
5
 * Copyright 2011-2012 GRNET S.A. All rights reserved.
6
 *
7
 * Redistribution and use in source and binary forms, with or
8
 * without modification, are permitted provided that the following
9
 * conditions are met:
10
 *
11
 *   1. Redistributions of source code must retain the above
12
 *      copyright notice, this list of conditions and the following
13
 *      disclaimer.
14
 *
15
 *   2. Redistributions in binary form must reproduce the above
16
 *      copyright notice, this list of conditions and the following
17
 *      disclaimer in the documentation and/or other materials
18
 *      provided with the distribution.
19
 *
20
 *
21
 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
22
 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
23
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
24
 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
25
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
26
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
27
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
28
 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
29
 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
30
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
31
 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
32
 * POSSIBILITY OF SUCH DAMAGE.
33
 *
34
 * The views and conclusions contained in the software and
35
 * documentation are those of the authors and should not be
36
 * interpreted as representing official policies, either expressed
37
 * or implied, of GRNET S.A.
38
 * </copyright>
39
 * -----------------------------------------------------------------------
40
 */
41
#endregion
42
using System;
43
using System.Collections.Concurrent;
44
using System.Collections.Generic;
45
using System.Diagnostics.Contracts;
46
using System.Threading;
47
using System.Threading.Async;
48
using System.Threading.Tasks;
49
using Pithos.Core.Agents;
50

    
51
namespace Pithos.Core
52
{
53
    public class Agent<TMessage> : IDisposable
54
    {
55
        private readonly ConcurrentQueue<TMessage> _queue;
56
        //private readonly AsyncCollection<TMessage> _messages;
57
        private readonly AsyncCollection<TMessage> _messages;
58
        private readonly CancellationTokenSource _cancelSource = new CancellationTokenSource();
59
        public CancellationToken CancellationToken;
60

    
61
        private readonly Action<Agent<TMessage>> _process;
62

    
63

    
64
        public Agent(Action<Agent<TMessage>> action)
65
        {
66
            _queue=new ConcurrentQueue<TMessage>();
67
            _messages = new AsyncCollection<TMessage>(_queue);            
68
            _process = action;
69
            CancellationToken = _cancelSource.Token;
70
        }
71

    
72
        public bool IsEmpty
73
        {
74
            get { return _queue.IsEmpty; }
75
        }
76

    
77
        public void Post(TMessage message)
78
        {
79
            _messages.Add(message);
80
        }
81

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

    
84
        public Task PostAndAwait(TMessage message)
85
        {            
86
            var tcs = new TaskCompletionSource<object>();
87
            _awaiters[message] = tcs;
88
            Post(message);
89
            return tcs.Task;
90
        }
91

    
92
        /// <summary>
93
        /// Receives a message asynchronously, optionally with a timeout. Receive throws a TimeoutException if the timeout expires
94
        /// </summary>
95
        /// <returns>A Task that will return the message asynchronously</returns>
96
        public  Task<TMessage> Receive()
97
        {
98
            return _messages.Take();
99
        }
100

    
101
        public void NotifyComplete(TMessage message)
102
        {
103
            TaskCompletionSource<object> tcs;
104
            if (_awaiters.TryRemove(message,out tcs))
105
                tcs.SetResult(null);
106
        }
107

    
108

    
109

    
110
        /// <summary>
111
        /// Start the agent
112
        /// </summary>
113
        public void Start()
114
        {
115
            Task.Factory.StartNew(() => _process(this), CancellationToken);            
116
        }
117

    
118

    
119
        /// <summary>
120
        /// Create and start a new agent for the specified type of message
121
        /// </summary>
122
        /// <param name="action">The message processing action</param>
123
        /// <returns>A started Agent</returns>
124
        public static Agent<TMessage> Start(Action<Agent<TMessage>> action)
125
        {
126
            var agent = new Agent<TMessage>(action);
127
            agent.Start();
128
            return agent;
129
        }
130

    
131
        /// <summary>
132
        /// Stops the agent 
133
        /// </summary>
134
        public void Stop()
135
        {
136
            //Stop the message queue
137
            //_messages.CompleteAdding();
138
            //And signal the cancellation
139
            _cancelSource.Cancel();
140
        }
141

    
142
        /// <summary>
143
        /// Execute an action asynchronously, using the agent's cancellation source
144
        /// </summary>
145
        /// <param name="action">The action to execute</param>
146
        public void DoAsync(Action action)
147
        {
148
            Contract.Requires(action!=null);
149
            Task.Factory.StartNew(action, CancellationToken);
150
        }
151

    
152

    
153
        ~Agent()
154
        {
155
            Dispose(false);
156
        }
157

    
158
        public void Dispose()
159
        {
160
            Dispose(true);
161
            GC.SuppressFinalize(this);
162
        }
163

    
164
        protected void Dispose(bool disposing)
165
        {
166
            if (disposing)
167
            {
168
                Stop();
169
                _cancelSource.Dispose();
170
            }
171
        }
172

    
173
        public IEnumerable<TMessage> GetEnumerable()
174
        {
175
            return _queue;
176
        }
177

    
178
        /// <summary>
179
        /// Remove the first message that matches the predicate
180
        /// </summary>
181
        /// <param name="predicate">The condition to match</param>
182
        /// <remarks>Removes the first message that matches the predicate by dequeing all 
183
        /// messages and re-enqueing all except the first matching message</remarks>
184
        public void Remove(Func<TMessage,bool> predicate)
185
        {
186
            //Can this work? Dequeue all items 
187
            //and then enqueue everything except the filtered items
188

    
189
            _queue.RemoveFirst(predicate);
190
        }
191

    
192
        public Task LoopAsync(Task process, Action loop,Action<Exception> onError=null)
193
        {
194
            Contract.Requires(process!=null);
195
            Contract.Requires(loop!=null);
196

    
197
            return process.ContinueWith(t =>
198
            {   
199
             
200
                Task.Factory.StartNew(loop, CancellationToken);
201

    
202
                if (t.IsFaulted)
203
                {
204
                    var ex = t.Exception.InnerException;
205
                    if (ex is OperationCanceledException)
206
                        Stop();
207
                    if (onError != null)
208
                        onError(ex);
209
                }
210
            },CancellationToken);
211
        }
212

    
213
        public Task<T> LoopAsync<T>(Task<T> process, Action loop,Action<Exception> onError=null)
214
        {
215
            return process.ContinueWith(t =>
216
            {   
217
                //Spawn the Loop immediatelly
218
                Task.Factory.StartNew(loop,CancellationToken);
219
                //Then process possible exceptions
220
                if (t.IsFaulted)
221
                {
222
                    var ex = t.Exception.InnerException;
223
                    if (ex is OperationCanceledException)
224
                        Stop();
225
                    if (onError != null)
226
                        onError(ex);
227
                }
228
                return default(T);
229
            });
230
        }
231
    }
232
}