mcs/class/System.Threading.Tasks.Dataflow/System.Threading.Tasks.Dataflow/BufferBlock.cs

   1 // BufferBlock.cs
   2 //
   3 // Copyright (c) 2011 Jérémie "garuma" Laval
   4 //
   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:
  11 //
  12 // The above copyright notice and this permission notice shall be included in
  13 // all copies or substantial portions of the Software.
  14 //
  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
  21 // THE SOFTWARE.
  22
  23 using System.Collections.Generic;
  24 using System.Collections.Concurrent;
  25
  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> ();
  33
  34                 public BufferBlock () : this (DataflowBlockOptions.Default)
  35                 {
  36                 }
  37
  38                 public BufferBlock (DataflowBlockOptions dataflowBlockOptions)
  39                 {
  40                         if (dataflowBlockOptions == null)
  41                                 throw new ArgumentNullException ("dataflowBlockOptions");
  42
  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);
  50                 }
  51
  52                 DataflowMessageStatus ITargetBlock<T>.OfferMessage (
  53                         DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source,
  54                         bool consumeToAccept)
  55                 {
  56                         return messageBox.OfferMessage (messageHeader, messageValue, source, consumeToAccept);
  57                 }
  58
  59                 public IDisposable LinkTo (ITargetBlock<T> target, DataflowLinkOptions linkOptions)
  60                 {
  61                         return outgoing.AddTarget (target, linkOptions);
  62                 }
  63
  64                 T ISourceBlock<T>.ConsumeMessage (DataflowMessageHeader messageHeader,
  65                                                   ITargetBlock<T> target,
  66                                                   out bool messageConsumed)
  67                 {
  68                         return outgoing.ConsumeMessage (messageHeader, target, out messageConsumed);
  69                 }
  70
  71                 bool ISourceBlock<T>.ReserveMessage (DataflowMessageHeader messageHeader,
  72                                                      ITargetBlock<T> target)
  73                 {
  74                         return outgoing.ReserveMessage (messageHeader, target);
  75                 }
  76
  77                 void ISourceBlock<T>.ReleaseReservation (DataflowMessageHeader messageHeader,
  78                                                          ITargetBlock<T> target)
  79                 {
  80                         outgoing.ReleaseReservation (messageHeader, target);
  81                 }
  82
  83                 public bool TryReceive (Predicate<T> filter, out T item)
  84                 {
  85                         return outgoing.TryReceive (filter, out item);
  86                 }
  87
  88                 public bool TryReceiveAll (out IList<T> items)
  89                 {
  90                         return outgoing.TryReceiveAll (out items);
  91                 }
  92
  93                 /// <summary>
  94                 /// Moves items from the input queue to the output queue.
  95                 /// </summary>
  96                 void ProcessQueue ()
  97                 {
  98                         T item;
  99                         while (messageQueue.TryTake (out item))
 100                                 outgoing.AddData (item);
 101                 }
 102
 103                 public void Complete ()
 104                 {
 105                         messageBox.Complete ();
 106                         outgoing.Complete ();
 107                 }
 108
 109                 void IDataflowBlock.Fault (Exception exception)
 110                 {
 111                         compHelper.RequestFault (exception);
 112                 }
 113
 114                 public Task Completion {
 115                         get {
 116                                 return compHelper.Completion;
 117                         }
 118                 }
 119
 120                 public int Count {
 121                         get {
 122                                 return outgoing.Count;
 123                         }
 124                 }
 125
 126                 public override string ToString ()
 127                 {
 128                         return NameHelper.GetName (this, dataflowBlockOptions);
 129                 }
 130         }
 131 }
 132