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

   1 // PredicateBlock.cs
   2 //
   3 // Copyright (c) 2012 Petr Onderka
   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 namespace System.Threading.Tasks.Dataflow {
  24         /// <summary>
  25         /// This block is used by the version of <see cref="DataflowBlock.LinkTo"/>
  26         /// that has a predicate to wrap the target block,
  27         /// so that the predicate can be checked.
  28         /// </summary>
  29         class PredicateBlock<T> : ITargetBlock<T> {
  30                 /// <summary>
  31                 /// Wraps the source block of the link.
  32                 /// This is necessary so that the communication from target to source works correctly.
  33                 /// </summary>
  34                 class SourceBlock : ISourceBlock<T> {
  35                         readonly ISourceBlock<T> actualSource;
  36                         readonly PredicateBlock<T> predicateBlock;
  37
  38                         public SourceBlock (ISourceBlock<T> actualSource,
  39                                             PredicateBlock<T> predicateBlock)
  40                         {
  41                                 this.actualSource = actualSource;
  42                                 this.predicateBlock = predicateBlock;
  43                         }
  44
  45                         public Task Completion
  46                         {
  47                                 get { return actualSource.Completion; }
  48                         }
  49
  50                         public void Complete ()
  51                         {
  52                                 actualSource.Complete ();
  53                         }
  54
  55                         public void Fault (Exception exception)
  56                         {
  57                                 actualSource.Fault (exception);
  58                         }
  59
  60                         public T ConsumeMessage (DataflowMessageHeader messageHeader,
  61                                                  ITargetBlock<T> target, out bool messageConsumed)
  62                         {
  63                                 return actualSource.ConsumeMessage (messageHeader, predicateBlock,
  64                                         out messageConsumed);
  65                         }
  66
  67                         public IDisposable LinkTo (ITargetBlock<T> target,
  68                                                    DataflowLinkOptions linkOptions)
  69                         {
  70                                 return actualSource.LinkTo (target, linkOptions);
  71                         }
  72
  73                         public void ReleaseReservation (DataflowMessageHeader messageHeader,
  74                                                         ITargetBlock<T> target)
  75                         {
  76                                 actualSource.ReleaseReservation (messageHeader, predicateBlock);
  77                         }
  78
  79                         public bool ReserveMessage (DataflowMessageHeader messageHeader,
  80                                                     ITargetBlock<T> target)
  81                         {
  82                                 return actualSource.ReserveMessage (messageHeader, predicateBlock);
  83                         }
  84                 }
  85
  86                 readonly ITargetBlock<T> actualTarget;
  87                 readonly Predicate<T> predicate;
  88                 readonly SourceBlock sourceBlock;
  89
  90                 public PredicateBlock (ISourceBlock<T> actualSource,
  91                                        ITargetBlock<T> actualTarget, Predicate<T> predicate)
  92                 {
  93                         this.actualTarget = actualTarget;
  94                         this.predicate = predicate;
  95                         sourceBlock = new SourceBlock (actualSource, this);
  96                 }
  97
  98                 public DataflowMessageStatus OfferMessage (
  99                         DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source,
 100                         bool consumeToAccept)
 101                 {
 102                         if (!messageHeader.IsValid)
 103                                 throw new ArgumentException ("The messageHeader is not valid.",
 104                                         "messageHeader");
 105                         if (consumeToAccept && source == null)
 106                                 throw new ArgumentException (
 107                                         "consumeToAccept may only be true if provided with a non-null source.",
 108                                         "consumeToAccept");
 109
 110                         if (!predicate(messageValue))
 111                                 return DataflowMessageStatus.Declined;
 112
 113                         return actualTarget.OfferMessage (messageHeader, messageValue, sourceBlock,
 114                                 consumeToAccept);
 115                 }
 116
 117                 public Task Completion {
 118                         get { return actualTarget.Completion; }
 119                 }
 120
 121                 public void Complete ()
 122                 {
 123                         actualTarget.Complete ();
 124                 }
 125
 126                 public void Fault (Exception exception)
 127                 {
 128                         actualTarget.Fault (exception);
 129                 }
 130         }
 131 }