Analogue of Queue.Peek() for BlockingCollection when listening to consuming IEnumerable<T>
.net-4.0, c#, multithreading, task-parallel-library, wcf
Solution
You should consider intermediate queue.
`BlockingCollection<T>` can't "peek" items because of its nature - there can be more than one consumer. One of them can peek an item, and another one can take it - hence, the first one will try to take item, that already has been taken.
Problem
I'm using Pipelines pattern implementation to decouple messages consumer from a producer to avoid slow-consumer issue. In case of any exception on a message processing stage `[1]` it will be lost and not dispatched to an other service/layer `[2]`. How can I handle such issue in `[3]` so message will not be lost and what is important! order of messages will not be mixed up so upper service/layer will get messages in the order they came in. I have an idea which involves an other intermediate `Queue` but it seems complex? Unfortunately `BlockingCollection<T>` does not expose any analogue of `Queue.Peek()` method so I can just read next available message and in case of successfull processing do `Dequeue()` ``` private BlockingCollection<IMessage> messagesQueue; // TPL Task does following: // Listen to new messages and as soon as any comes in - process it foreach (var cachedMessage in messagesQueue.GetConsumingEnumerable(cancellation)) { const int maxRetries = 3; int retriesCounter = 0; bool isSent = false; // On this point a message already is removed from messagesQueue while (!isSent && retriesCounter++ <= maxRetries) { try { // [1] Preprocess a message // [2] Dispatch to an other service/layer clientProxyCallback.SendMessage(cachedMessage); isSent = true; } catch(Exception exception) { // [3] // logging if (!isSent && retriesCounter < maxRetries) { Thread.Sleep(NSeconds); } } if (!isSent && retriesCounter == maxRetries) { // just log, message is lost on this stage! } } } ``` EDIT: Forgot to say this is IIS hosted WCF service which dispatches messages back to Silverlight client WCF Proxy via client callback contract. EDIT2: Below is how I would do this using `Peek()`, Am I missing something? ``` bool successfullySent = true; try { var item = queue.Peek(); PreProcessItem(item); SendItem(item); } catch(Exception exception) { successfullySent = false; } finally { if (successfullySent) { // just remove already sent item from the queue queue.Dequeue(); } } ``` EDIT3: Surely I can use old style approach using while loop, bool flag, `Queue` and `AutoResetEvent`, but I just wondering whether the same is possible using `BlockingCollection` and `GetConsumingEnumerable()` I think facility like `Peek` would be very helpful when using together with consuming enumerable, since otherwise all Pipeline pattern implementation examples new stuff like `BlockingCollection` and `GetConsumingEnumerable()` looks not durable and I have to move back to the old approach.