public Engine(IWorkItemRepository repository, IActivityRunner activityRunner,
            IStateMachineProvider stateMachineProvider)
        {
            if (repository == null) throw new ArgumentNullException("repository");
            if (activityRunner == null) throw new ArgumentNullException("activityRunner");
            if (stateMachineProvider == null) throw new ArgumentNullException("stateMachineProvider");

            _repository = repository;
            _activityRunner = activityRunner;
            _stateMachineProvider = stateMachineProvider;

            _stateQueue = new ActionBlock<int>(id => UpdateState(id),
                new ExecutionDataflowBlockOptions
                {
                    MaxDegreeOfParallelism = 1
                });

            _workerQueue = new ActionBlock<int>(id => RunActivity(id),
                new ExecutionDataflowBlockOptions
                {
                    MaxDegreeOfParallelism = int.MaxValue
                });

            _stateQueue.Completion.ContinueWith(t => { _workerQueue.Complete(); }, TaskContinuationOptions.OnlyOnFaulted);

            _workerQueue.Completion.ContinueWith(t => { ((IDataflowBlock) _stateQueue).Fault(t.Exception); },
                TaskContinuationOptions.OnlyOnFaulted);
        }

 public EventMapper(BleSystemConfiguration systemConfiguration)
 {
     var receiverPaths = systemConfiguration.ReceiverPaths;
     _useWeightedPath = systemConfiguration.UseWeightedPaths;
     _receiverPaths = receiverPaths.ToDictionary(rp => rp);
     _eventGroupHandler = new ActionBlock<List<SignalEventDetails>>(deg => HandleEventGroup(deg), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 });
     _eventsByReceiver = new Dictionary<BleReceiver, Dictionary<string, SignalEventDetails>>(systemConfiguration.BleReceivers.Count);
     _monitoredReceivers = systemConfiguration.BleReceivers.Values.ToList();
     _monitoredTransmitters = systemConfiguration.BleTransmitters.Values.ToList();
     _receiverPathsTree = new Dictionary<BleReceiver, Dictionary<ReceiverPath, ReceiverPath>>();
     foreach (var bleReceiver in _monitoredReceivers)
     {
         if (_eventsByReceiver.ContainsKey(bleReceiver))
             continue;
         var pathDictionary = new Dictionary<ReceiverPath, ReceiverPath>();
         _receiverPathsTree.Add(bleReceiver, pathDictionary);
         foreach (var receiverPath in receiverPaths)
         {
             if (Equals(receiverPath.From, bleReceiver) || Equals(receiverPath.To, bleReceiver))
                 pathDictionary.Add(receiverPath, receiverPath);
         }
         var receiverTransmitters = new Dictionary<string, SignalEventDetails>(_monitoredTransmitters.Count);
         _monitoredTransmitters.ForEach(t =>
         {
             if (receiverTransmitters.ContainsKey(t.MacAddress))
                 return;
             receiverTransmitters.Add(t.MacAddress,
                 new SignalEventDetails { Transmitter = t, BleReceiver = bleReceiver });
         });
         _eventsByReceiver.Add(bleReceiver, receiverTransmitters);
     }
     _scanTimer = new Timer(ScanInterval);
     _scanTimer.Elapsed += ScanIntervalElapsed;
     _scanTimer.Enabled = true;
 }

        public async Task TestDataBroadcaster2()
        {
            var random = new Random();
            var buffer = new BufferBlock<int>();
            
            int sum1 = 0;
            int sum2 = 0;
            int sum3 = 0;

            var action1 = new ActionBlock<int>(i => sum1 = sum1 + i);
            var action2 = new ActionBlock<int>(i => sum2 = sum2 + i);
            var action3 = new ActionBlock<int>(i => sum3 = sum3 + i);

            buffer.ToDataflow().LinkToMultiple(new []{action1, action2, action3}.Select(a => a.ToDataflow()).ToArray());
            
            for (int j = 0; j < 1000; j++)
            {
                buffer.Post((int)(random.NextDouble() * 10000));
            }

            buffer.Complete();

            await TaskEx.AwaitableWhenAll(action1.Completion, action2.Completion, action3.Completion);

            Console.WriteLine("sum1 = {0} , sum2 = {1}", sum1, sum2);
            Assert.AreEqual(sum1, sum2);
            Assert.AreEqual(sum1, sum3);
        }

		public void BasicUsageTest ()
		{
			int[] array = null;
			var evt = new ManualResetEventSlim (false);

			var buffer = new BatchBlock<int> (10);
			var block = new ActionBlock<int[]> (i =>
			{
				array = i;
				evt.Set ();
			});
			buffer.LinkTo<int[]> (block);

			for (int i = 0; i < 9; i++)
				Assert.IsTrue (buffer.Post (i));

			Assert.IsFalse (evt.Wait (100));

			Assert.IsNull (array);

			Assert.IsTrue (buffer.Post (42));
			Assert.IsTrue (evt.Wait (1000));

			Assert.IsNotNull (array);
			CollectionAssert.AreEqual (new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 42 }, array);
		}

        public void Run()
        {
            Console.WriteLine("Generating first {0} powers of 2.", MaxItems);
            var bufferBlock = new BufferBlock<int>();
            Enumerable.Range(1, MaxItems)
                .ToList()
                .ForEach(i => bufferBlock.Post(i));

            Console.WriteLine("Signaling completion to the source block.");
            bufferBlock.Complete();
            Console.WriteLine("Done.");

            Console.WriteLine("Creating and linking the remaing blocks to the network.");
            var transformBlock = new TransformBlock<int, double>(i =>
            {
                Thread.Sleep(200);
                return Math.Pow(2, i);
            }, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });

            var actionBlock = new ActionBlock<double>(async i =>
            {
                await Task.Delay(500);
                Console.WriteLine(i);
            }, new ExecutionDataflowBlockOptions { BoundedCapacity = 10 });

            bufferBlock.LinkTo(transformBlock, new DataflowLinkOptions { PropagateCompletion = true });
            transformBlock.LinkTo(actionBlock, new DataflowLinkOptions { PropagateCompletion = true });

            Console.WriteLine("Waiting for the completion to be propagated through the network...");
            actionBlock.Completion.ContinueWith(t =>
            {
                Console.WriteLine("Finished processing.");
                Console.WriteLine(string.Format("Completion status: {0}.", t.Status));
            }).Wait();
        }

        /// <summary>
        /// Initialize a new instance of the <see cref="DataflowWebHookSender"/> with the given retry policy, <paramref name="options"/>,
        /// and <paramref name="httpClient"/>. This constructor is intended for unit testing purposes.
        /// </summary>
        internal DataflowWebHookSender(
            ILogger logger,
            IEnumerable<TimeSpan> retryDelays,
            ExecutionDataflowBlockOptions options,
            HttpClient httpClient,
            Action<WebHookWorkItem> onWebHookSuccess,
            Action<WebHookWorkItem> onWebHookFailure)
            : base(logger)
        {
            retryDelays = retryDelays ?? DefaultRetries;

            options = options ?? new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = DefaultMaxConcurrencyLevel
            };

            _httpClient = httpClient ?? new HttpClient();

            // Create the launch processors with the given retry delays
            _launchers = new ActionBlock<WebHookWorkItem>[1 + retryDelays.Count()];

            int offset = 0;
            _launchers[offset++] = new ActionBlock<WebHookWorkItem>(async item => await LaunchWebHook(item), options);
            foreach (TimeSpan delay in retryDelays)
            {
                _launchers[offset++] = new ActionBlock<WebHookWorkItem>(async item => await DelayedLaunchWebHook(item, delay));
            }

            string msg = string.Format(CultureInfo.CurrentCulture, CustomResources.Manager_Started, typeof(DataflowWebHookSender).Name, _launchers.Length);
            Logger.Info(msg);

            // Set handlers for testing purposes
            _onWebHookSuccess = onWebHookSuccess;
            _onWebHookFailure = onWebHookFailure;
        }

        static Publisher()
        {
            _trackerBlock = new ActionBlock<ContainerBase>(async tc =>
            {
                try
                {
                    ContainerBase trackerContainer = tc;
                    if (_firstTC == null)
                        _firstTC = trackerContainer;
                    else if (_firstTC == trackerContainer &&
                             !_trackerBlockCancellationTokenSource.IsCancellationRequested)
                    {
                        try
                        {
                            await
                                Task.Delay(TimeSpan.FromSeconds(15), _trackerBlockCancellationTokenSource.Token)
                                    .ConfigureAwait(false);
                        }
                        catch (OperationCanceledException)
                        {
                            Configurator.Configuration.Logger.Info("Task.Delay exiting");
                        }
                    }

                    foreach (
                        TrackerData td in
                            trackerContainer.GetTrackerData(_trackerBlockCancellationTokenSource.IsCancellationRequested)
                        )
                    {
                        _publisherBlock.Post(td);
                    }
                }
                catch (Exception ex)
                {
                    Configurator.Configuration.Logger.Error(ex.Message, ex);
                }
                finally
                {
                    if (!_trackerBlockCancellationTokenSource.IsCancellationRequested)
                        _trackerBlock.Post(tc);
                    else if (_trackerBlock.InputCount == 0)
                        _trackerBlock.Complete();
                }
            });

            _publisherBlock = new ActionBlock<TrackerData>(tc =>
            {
                try
                {
                    _lastPersistanceComplete = false;
                    Configurator.Configuration.Persister.Persist(tc);
                    _lastPersistanceComplete = true;
                }
                catch (Exception ex)
                {
                    Configurator.Configuration.Logger.Error(ex.Message, ex);
                    _lastPersistanceComplete = true;
                }
            }, new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = 4});
        }

 private void GetLastEventProcessedForHandlers()
 {
     //TODO calculate the lowest numbered LEP of all the handlers and use that for the start position 
     // ask each registered handler to get there last processed event and hold on to it.
     var actionBlock = new ActionBlock<IHandler>(x => x.GetLastPositionProcessed(), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 8 });
     _eventHandlers.ForEach(async x=> await actionBlock.SendAsync(x));
 }

        public static IPropagatorBlock<File, FileLine> GetFileLinesEnumeratorBlock()
        {
            var resultsBlock = new BufferBlock<FileLine>();
            var actionBlock = new ActionBlock<File>(
                async file =>
                {
                    using (var reader = new System.IO.StreamReader(new System.IO.FileStream(
                        file.FullPath,
                        System.IO.FileMode.Open,
                        System.IO.FileAccess.Read,
                        System.IO.FileShare.Read,
                        bufferSize: 4096,
                        useAsync: true)))
                    {
                        string line;
                        var row = 1;
                        while ((line = await reader.ReadLineAsync()) != null)
                        {
                            if (!string.IsNullOrWhiteSpace(line))
                            {
                                resultsBlock.Post(new FileLine(file, row, line));
                            }

                            row++;
                        }
                    }
                },
                new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Utils.GlobalMaxDegreeOfParallelism });

            actionBlock.PropagateCompleted(resultsBlock);
            return DataflowBlock.Encapsulate(actionBlock, resultsBlock);
        }

        public void Run()
        {
            var cts = new CancellationTokenSource();

            Console.WriteLine("Generating first {0} powers of 2.", MaxItems);
            var bufferBlock = new BufferBlock<int>(new DataflowBlockOptions { CancellationToken = cts.Token });
            Enumerable.Range(1, MaxItems)
                .ToList()
                .ForEach(i => bufferBlock.Post(i));
            Console.WriteLine("Scheduling cancellation after 5 seconds.");
            cts.CancelAfter(TimeSpan.FromSeconds(5));

            Console.WriteLine("Creating and linking the remaing blocks to the network.");
            var transformBlock = new TransformBlock<int, double>(i =>
            {
                Thread.Sleep(500);
                return Math.Pow(2, i);
            }, new ExecutionDataflowBlockOptions { BoundedCapacity = 1, CancellationToken = cts.Token });

            var actionBlock = new ActionBlock<double>(async i =>
            {
                await Task.Delay(1000);
                Console.WriteLine(i);
            }, new ExecutionDataflowBlockOptions { BoundedCapacity = 10, CancellationToken = cts.Token });

            bufferBlock.LinkTo(transformBlock, new DataflowLinkOptions { PropagateCompletion = true });
            transformBlock.LinkTo(actionBlock, new DataflowLinkOptions { PropagateCompletion = true });

            var t1 = bufferBlock.Completion.ContinueWith(t => Console.WriteLine("Buffer block status: {0}", t.Status));
            var t2 = actionBlock.Completion.ContinueWith(t => Console.WriteLine("Action block status: {0}", t.Status));
            Console.WriteLine("Waiting for the network to finish.");
            Task.WaitAll(t1, t2);
        }

		public void BasicUsageTest ()
		{
			bool act1 = false, act2 = false;
			var evt = new CountdownEvent (2);

			var block = new WriteOnceBlock<int> (null);
			var action1 = new ActionBlock<int> (i =>
			{
				act1 = i == 42;
				evt.Signal ();
			});
			var action2 = new ActionBlock<int> (i =>
			{
				act2 = i == 42;
				evt.Signal ();
			});

			block.LinkTo (action1);
			block.LinkTo (action2);

			Assert.IsTrue (block.Post (42), "#1");
			Assert.IsFalse (block.Post (43), "#2");

			Assert.IsTrue (evt.Wait (1000), "#3");

			Assert.IsTrue (act1, "#4");
			Assert.IsTrue (act2, "#5");
		}

        public static ITargetBlock<string> SetupPipeline()
        {
            var fileNamesForPath = new TransformBlock<string, IEnumerable<string>>(
              path =>
              {
                  return GetFileNames(path);
              });

            var lines = new TransformBlock<IEnumerable<string>, IEnumerable<string>>(
              fileNames =>
              {
                  return LoadLines(fileNames);
              });

            var words = new TransformBlock<IEnumerable<string>, IEnumerable<string>>(
              lines2 =>
              {
                  return GetWords(lines2);
              });

            var display = new ActionBlock<IEnumerable<string>>(
              coll =>
              {
                  foreach (var s in coll)
                  {
                      WriteLine(s);
                  }
              });

       
            fileNamesForPath.LinkTo(lines);
            lines.LinkTo(words);
            words.LinkTo(display);
            return fileNamesForPath;
        }

        public void Run()
        {
            Console.WriteLine("Generating first {0} powers of 2.", MaxItems);
            var bufferBlock = new BufferBlock<int>();
            var transformBlock = new TransformBlock<int, double>(i =>
            {
                Thread.Sleep(500);
                return Math.Pow(2, i);
            }, new ExecutionDataflowBlockOptions { BoundedCapacity = 10 });

            var actionBlock = new ActionBlock<double>(async i =>
            {
                await Task.Delay(1000);
                Console.WriteLine(i);
                _waitHandle.Signal();
            }, new ExecutionDataflowBlockOptions { BoundedCapacity = 10 });

            bufferBlock.LinkTo(transformBlock);
            transformBlock.LinkTo(actionBlock);

            Enumerable.Range(1, MaxItems)
                .ToList()
                .ForEach(i => bufferBlock.Post(i));

            _waitHandle.Wait();
        }

        public RobotStateDashboard()
        {
            dataBlockState = new ActionBlock<IRobotState>(v => { Dispatcher.RunAsync(CoreDispatcherPriority.Normal, () => { StateDataText = v.ToString(); }).AsTask().Wait(); });
            dataBlockPose = new ActionBlock<IRobotPose>(v => { Dispatcher.RunAsync(CoreDispatcherPriority.Normal, () => { PoseDataText = v.ToString(); }).AsTask().Wait(); });

            this.InitializeComponent();
        }

        public ExecutionPipeline(Kernel kernel)
        {
            _kernel = kernel;
            _commandQueue = new BufferBlock<CommandRequest[]>();
            _queryQueue = new BatchBlock<QueryRequest>(MaxConcurrentQueries);

            var transactionHandler = new ActionBlock<object>(t =>
            {
                if (t is QueryRequest[])
                {
                    var queries = t as QueryRequest[];
                    Task[] tasks = queries.Select(q => Task.Factory.StartNew(_ => ExecuteQuery(q), null)).ToArray();
                    Task.WaitAll(tasks);
                }
                else if (t is CommandRequest[])
                {
                    var commands = t as CommandRequest[];
                    foreach (var commandContext in commands)
                    {
                        var result = _kernel.Execute(commandContext.Command);
                        commandContext.Response.Post(result);
                    }
                }

            });
            _commandQueue.LinkTo(transactionHandler);
            _queryQueue.LinkTo(transactionHandler);
            _timer = new Timer(_ => _queryQueue.TriggerBatch());
            _timer.Change(Interval, Interval);
        }

        public MatchPipeline(RiotApi riotApi, RiotQuerySettings querySettings, IMatchDetailProcessor matchDetailProcessor)
        {
            api = riotApi;
            this.querySettings = querySettings;
            this.matchDetailProcessor = matchDetailProcessor;

            queryQueues.Add(querySettings.Queue);

            // Create match producer
            if (!querySettings.NoDownload)
            {
                playerMatchProducer = new PlayerMatchProducer(api, querySettings, queryQueues, testSynchronizer);

                // Create blocks
                ConsumeMatchBlock = new TransformBlock<MatchSummary, MatchDetail>(
                    async match => await ConsumeMatch(match),
                    new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 5 });
            }
            ConsumeMatchDetailBlock = new ActionBlock<MatchDetail>(
                async match => await matchDetailProcessor.ConsumeMatchDetail(match),
                new ExecutionDataflowBlockOptions() {
                    MaxDegreeOfParallelism = matchDetailProcessor.MaxDegreeOfParallelism
                });

            // Link blocks
            if (!querySettings.NoDownload)
            {
                playerMatchProducer.MatchProducerBlock.LinkTo(ConsumeMatchBlock, new DataflowLinkOptions() { PropagateCompletion = false });
                ConsumeMatchBlock.LinkTo(ConsumeMatchDetailBlock, new DataflowLinkOptions() { PropagateCompletion = false }, match => match != null);
                ConsumeMatchBlock.LinkTo(DataflowBlock.NullTarget<MatchDetail>(), new DataflowLinkOptions() { PropagateCompletion = false });
            }
        }

        public CrawlScheduler(IUrlHasher urlHasher, IConfiguration configuration, ISinglePageCrawler pageCrawler, ICrawlUrlRepository crawlUrlRepository)
        {
            this.urlHasher = urlHasher;
            this.configuration = configuration;
            this.pageCrawler = pageCrawler;
            this.crawlUrlRepository = crawlUrlRepository;

            schedulingBlock = new ActionBlock<PageCrawlResult>(result =>
                {
                    var websiteDefinition = result.CrawlUrl.WebsiteDefinition;

                    Interlocked.Increment(ref websiteDefinition.ProcessedUrlsCount);

                    RaisePageCrawled(result);

                    if (result.Links.Any())
                    {
                        Schedule(result.Links);
                    }

                    if (websiteDefinition.UrlsToProcessCount == websiteDefinition.ProcessedUrlsCount)
                    {
                        websiteProcessingDefinitions[websiteDefinition].Complete();
                    }

                    ScheduleNext();
                });
        }

        public void ProduceLogs(int count, int buffSize)
        {
            var bufferOptions = new DataflowBlockOptions() { BoundedCapacity = buffSize };
            var writerOptions = new ExecutionDataflowBlockOptions() { BoundedCapacity = 10, MaxDegreeOfParallelism = 1, MaxMessagesPerTask = 10, SingleProducerConstrained = true };

            LogGenerator g = new LogGenerator();

            var file = new StreamWriter("basic.async.buff.log", false);

            BufferBlock<string> buffer = new BufferBlock<string>(bufferOptions);
            ActionBlock<string> writer = new ActionBlock<string>(s => file.WriteLine(s), writerOptions);

            buffer.LinkTo(writer, new DataflowLinkOptions() { PropagateCompletion = true });

            for (int i = 0; i < count; i++)
            {
                g.Next();

                var line = string.Format(g.FormatStr, g.Param1, g.Param2, g.Param3, g.Param4, g.Param5, g.Param6);
                writer.SendAsync(line).Wait();
            }

            buffer.Complete();

            Completed = writer.Completion.ContinueWith(t => file.Close());
        }

		public void BasicUsageTest ()
		{
			Tuple<IList<int>, IList<int>, IList<string>> result = null;
			var evt = new ManualResetEventSlim (false);

			var actionBlock =
				new ActionBlock<Tuple<IList<int>, IList<int>, IList<string>>> (r =>
				{
					result = r;
					evt.Set ();
				});
			var block = new BatchedJoinBlock<int, int, string> (3);

			block.LinkTo (actionBlock);

			// all targets once
			Assert.IsTrue (block.Target1.Post (1));
			Assert.IsTrue (block.Target2.Post (2));

			Assert.IsFalse (evt.Wait (100));
			Assert.IsNull (result);

			Assert.IsTrue (block.Target3.Post ("foo"));

			Assert.IsTrue (evt.Wait (1000));

			Assert.IsNotNull (result);
			CollectionAssert.AreEqual (new[] { 1 }, result.Item1);
			CollectionAssert.AreEqual (new[] { 2 }, result.Item2);
			CollectionAssert.AreEqual (new[] { "foo" }, result.Item3);
		}

        private FileLogWriter(string filePath) {
            _filePath = filePath;
            _messages = new ActionBlock<string>(new Func<string, Task>(WriteToFile));

            AppDomain.CurrentDomain.ProcessExit += OnProcessExit;
            AppDomain.CurrentDomain.UnhandledException += OnUnhandledException;
        }