anchor.diff:                  let cts = new CancellationTokenSource()
                   return 
                       { new Collector<'T, 'R option> with
  -                        member self.DegreeOfParallelism = stream.DegreeOfParallelism
  +                        member self.DegreeOfParallelism = flow.DegreeOfParallelism
                           member self.Iterator() = 
                               {   Index = ref -1; 
                                   Func = (fun value -> match chooser ctx value with Some value' -> resultRef := Some value'; cts.Cancel() | None -> ());
  
positive.code:                 let cts =  new CancellationTokenSource()
                  return
                      { new Collector<'T, 'T option> with
                          member self.DegreeOfParallelism = stream.DegreeOfParallelism 
                          member self.Iterator() = 
                              {   Index = ref -1; 
                                  Func = (fun value -> if predicate ctx value then resultRef := Some value; cts.Cancel() else ());
  
negative.code:                             if s = e then
                                  return! projection collector.Result
                              else
                                  use! flow = CloudFile.Read(path, (fun flow -> async { return flow }), true)
                                  let parStream = ParStream.ofSeq (rangeBasedReadLines s e flow) 
                                  let collectorResult = parStream.Apply (toParStreamCollector collector)
                                  return! projection collectorResult
                          }
  