/
BlockingQueue.cs
227 lines (199 loc) · 8.09 KB
/
BlockingQueue.cs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
//-----------------------------------------------------------------------
// <copyright file="BlockingQueue.cs" company="Akka.NET Project">
// Copyright (C) 2009-2021 Lightbend Inc. <http://www.lightbend.com>
// Copyright (C) 2013-2021 .NET Foundation <https://github.com/akkadotnet/akka.net>
// </copyright>
//-----------------------------------------------------------------------
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
namespace Akka.TestKit.Internal
{
/// <summary>
/// This class represents a queue with the same characteristics of a <see cref="BlockingCollection{T}"/>.
/// The queue can enqueue items at either the front (FIFO) or the end (LIFO) of the collection.
/// <remarks>Note! Part of internal API. Breaking changes may occur without notice. Use at own risk.</remarks>
/// </summary>
/// <typeparam name="T">The type of item to store.</typeparam>
public class BlockingQueue<T>
{
private readonly BlockingCollection<Positioned> _collection = new BlockingCollection<Positioned>(new QueueWithAddFirst());
/// <summary>
/// The number of items that are currently in the queue.
/// </summary>
public int Count { get { return _collection.Count; } }
/// <summary>
/// Adds the specified item to the end of the queue.
/// </summary>
/// <param name="item">The item to add to the queue.</param>
public void Enqueue(T item)
{
if (!_collection.TryAdd(new Positioned(item)))
throw new InvalidOperationException("Failed to enqueue item into the queue.");
}
/// <summary>
/// Adds the specified item to the front of the queue.
/// </summary>
/// <param name="item">The item to add to the queue.</param>
public void AddFirst(T item)
{
if(!_collection.TryAdd(new Positioned(item, first:true)))
throw new InvalidOperationException("Failed to enqueue item into the head of the queue.");
}
/// <summary>
/// Tries to add the specified item to the end of the queue within the specified time period.
/// A token can be provided to cancel the operation if needed.
/// </summary>
/// <param name="item">The item to add to the queue.</param>
/// <param name="millisecondsTimeout">The number of milliseconds to wait for the add to complete.</param>
/// <param name="cancellationToken">The cancellation token that can be used to cancel the operation.</param>
/// <returns><c>true</c> if the add completed within the specified timeout; otherwise, <c>false</c>.</returns>
public bool TryEnqueue(T item, int millisecondsTimeout, CancellationToken cancellationToken)
{
return _collection.TryAdd(new Positioned(item), millisecondsTimeout, cancellationToken);
}
/// <summary>
/// Tries to remove the specified item from the queue.
/// </summary>
/// <param name="item">The item to remove from the queue.</param>
/// <returns><c>true</c> if the item was removed; otherwise, <c>false</c>.</returns>
public bool TryTake(out T item)
{
if(_collection.TryTake(out var p))
{
item = p.Value;
return true;
}
item = default;
return false;
}
/// <summary>
/// Tries to remove the specified item from the queue within the specified time period.
/// A token can be provided to cancel the operation if needed.
/// </summary>
/// <param name="item">The item to remove from the queue.</param>
/// <param name="millisecondsTimeout">The number of milliseconds to wait for the remove to complete.</param>
/// <param name="cancellationToken">The cancellation token that can be used to cancel the operation.</param>
/// <returns><c>true</c> if the remove completed within the specified timeout; otherwise, <c>false</c>.</returns>
public bool TryTake(out T item, int millisecondsTimeout, CancellationToken cancellationToken)
{
if(_collection.TryTake(out var p, millisecondsTimeout, cancellationToken))
{
item = p.Value;
return true;
}
item = default;
return false;
}
/// <summary>
/// Removes an item from the collection.
/// </summary>
/// <param name="cancellationToken">The cancellation token that can be used to cancel the operation.</param>
/// <exception cref="OperationCanceledException">
/// This exception is thrown when the operation is canceled.
/// </exception>
/// <returns>The item removed from the collection.</returns>
public T Take(CancellationToken cancellationToken)
{
var p = _collection.Take(cancellationToken);
return p.Value;
}
/// <summary>
/// Copies the items from the <see cref="BlockingQueue{T}"/> instance into a new <see cref="List{T}"/>.
/// </summary>
/// <returns>A <see cref="List{T}"/> containing copies of the elements of the collection</returns>
public List<T> ToList()
{
var positionArray = _collection.ToArray();
return positionArray.Select(positioned => positioned.Value).ToList();
}
private class Positioned
{
private readonly T _value;
private readonly bool _first;
public Positioned(T value, bool first = false)
{
_value = value;
_first = first;
}
public T Value { get { return _value; } }
public bool First { get { return _first; } }
}
private class QueueWithAddFirst : IProducerConsumerCollection<Positioned>
{
private readonly LinkedList<Positioned> _list = new LinkedList<Positioned>();
public int Count {
get
{
lock (SyncRoot)
{
return _list.Count;
}
}
}
public bool TryAdd(Positioned item)
{
lock (SyncRoot)
{
if(item.First)
_list.AddFirst(item);
else
_list.AddLast(item);
return true;
}
}
public bool TryTake(out Positioned item)
{
lock(SyncRoot)
{
if(_list.Count == 0)
{
item = null;
return false;
}
item = _list.First.Value;
_list.RemoveFirst();
return true;
}
}
public void CopyTo(Positioned[] array, int index)
{
lock(SyncRoot)
{
_list.CopyTo(array, index);
}
}
public void CopyTo(Array array, int index)
{
lock(SyncRoot)
{
((ICollection)_list).CopyTo(array, index);
}
}
public Positioned[] ToArray()
{
lock(SyncRoot)
{
return _list.ToArray();
}
}
public IEnumerator<Positioned> GetEnumerator()
{
lock(SyncRoot)
{
//We must create a copy
return new List<Positioned>(_list).GetEnumerator();
}
}
IEnumerator IEnumerable.GetEnumerator()
{
return GetEnumerator();
}
public object SyncRoot { get; } = new object();
public bool IsSynchronized => true;
}
}
}