3 // Copyright (c) 2011 Jérémie "garuma" Laval
5 // Permission is hereby granted, free of charge, to any person obtaining a copy
6 // of this software and associated documentation files (the "Software"), to deal
7 // in the Software without restriction, including without limitation the rights
8 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 // copies of the Software, and to permit persons to whom the Software is
10 // furnished to do so, subject to the following conditions:
12 // The above copyright notice and this permission notice shall be included in
13 // all copies or substantial portions of the Software.
15 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
23 using System.Collections.Generic;
24 using System.Collections.Concurrent;
26 namespace System.Threading.Tasks.Dataflow {
27 public sealed class BufferBlock<T> : IPropagatorBlock<T, T>, IReceivableSourceBlock<T> {
28 readonly DataflowBlockOptions dataflowBlockOptions;
29 readonly CompletionHelper compHelper;
30 readonly MessageBox<T> messageBox;
31 readonly OutgoingQueue<T> outgoing;
32 readonly BlockingCollection<T> messageQueue = new BlockingCollection<T> ();
34 public BufferBlock () : this (DataflowBlockOptions.Default)
38 public BufferBlock (DataflowBlockOptions dataflowBlockOptions)
40 if (dataflowBlockOptions == null)
41 throw new ArgumentNullException ("dataflowBlockOptions");
43 this.dataflowBlockOptions = dataflowBlockOptions;
44 this.compHelper = CompletionHelper.GetNew (dataflowBlockOptions);
45 this.messageBox = new PassingMessageBox<T> (this, messageQueue, compHelper,
46 () => outgoing.IsCompleted, _ => ProcessQueue (), dataflowBlockOptions);
47 this.outgoing = new OutgoingQueue<T> (this, compHelper,
48 () => messageQueue.IsCompleted, messageBox.DecreaseCount,
49 dataflowBlockOptions);
52 DataflowMessageStatus ITargetBlock<T>.OfferMessage (
53 DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source,
56 return messageBox.OfferMessage (messageHeader, messageValue, source, consumeToAccept);
59 public IDisposable LinkTo (ITargetBlock<T> target, DataflowLinkOptions linkOptions)
61 return outgoing.AddTarget (target, linkOptions);
64 T ISourceBlock<T>.ConsumeMessage (DataflowMessageHeader messageHeader,
65 ITargetBlock<T> target,
66 out bool messageConsumed)
68 return outgoing.ConsumeMessage (messageHeader, target, out messageConsumed);
71 bool ISourceBlock<T>.ReserveMessage (DataflowMessageHeader messageHeader,
72 ITargetBlock<T> target)
74 return outgoing.ReserveMessage (messageHeader, target);
77 void ISourceBlock<T>.ReleaseReservation (DataflowMessageHeader messageHeader,
78 ITargetBlock<T> target)
80 outgoing.ReleaseReservation (messageHeader, target);
83 public bool TryReceive (Predicate<T> filter, out T item)
85 return outgoing.TryReceive (filter, out item);
88 public bool TryReceiveAll (out IList<T> items)
90 return outgoing.TryReceiveAll (out items);
94 /// Moves items from the input queue to the output queue.
99 while (messageQueue.TryTake (out item))
100 outgoing.AddData (item);
103 public void Complete ()
105 messageBox.Complete ();
106 outgoing.Complete ();
109 void IDataflowBlock.Fault (Exception exception)
111 compHelper.RequestFault (exception);
114 public Task Completion {
116 return compHelper.Completion;
122 return outgoing.Count;
126 public override string ToString ()
128 return NameHelper.GetName (this, dataflowBlockOptions);