using System;
    using System.Collections.Concurrent;
    using System.Threading;
    using System.Threading.Tasks;
    using System.Windows.Forms;
    
    namespace PCQueueDemo
    {
        public partial class Form1 : Form
        {
            // to consume results that are depedent
            private ConcurrentDictionary<int, WorkItem> _dict = new ConcurrentDictionary<int, WorkItem>();

        private readonly object _sync = new object();
        private BindingSource _bindingSource = new BindingSource();
        private TaskScheduler _ui; 
        private ProducerConsumerQueue _producers;
        private BlockingCollection<WorkItem> _workQueue;
        public Form1()
        {
            InitializeComponent();
            _workQueue = new BlockingCollection<WorkItem>();

            // create 3 producers which will produce work item to the work queue.
            _producers = new ProducerConsumerQueue(3, _workQueue);

            // capture the UI context in order to update data grid.
            _ui = TaskScheduler.FromCurrentSynchronizationContext();

            // bind the data grid
            dataGridView.DataSource = _bindingSource;
            dataGridView.AutoSize = true;
            dataGridView.Dock = DockStyle.Fill;
        }

        // Run button click handler
        private void button1_Click(object sender, EventArgs e)
        {
            _producers.Produce(); // start producing on each producer thread.
            var ignore = Task.Run(() => updateGrid()); // consume the producer result to update the grid.
        }

        private void updateGrid()
        {
            foreach (var item in _workQueue.GetConsumingEnumerable())
            {
                WorkItem copy = item;

                //? Is it a waste to use a thread for each result?
                
                Task.Factory.StartNew(() =>
                {
                    WorkItem localItem;
                    if (_dict.TryGetValue(item.Number, out localItem))
                    {
                        //? how to avoid locking if the update here takes longer than simply incrementing the counter?
                        lock (_sync) localItem.Counter++;
                    }
                    else if (_dict.TryAdd(copy.Number, copy))
                    {
                        //? Is this negligible since the lock should be very short?
                        lock(_sync) _bindingSource.Add(copy);
                    }
                    dataGridView.Refresh();
                }, CancellationToken.None, TaskCreationOptions.None, _ui);
            }
        }
    }

    // a demo work item
    public class WorkItem
    {
        public WorkItem(int number)
        {
            Number = number;
            Counter = 1; // 1st appearance
        }
        public int Number { get; private set; }
        public int Counter { get; set; }
    }

    //? This is just a producer queue; how to encapsulate the consumer together with producer while allowing it to update the UI?
    public class ProducerConsumerQueue
    {
        private readonly int _producerCount;
        private Task[] _producers;
        private BlockingCollection<WorkItem> _workQueue;

        public ProducerConsumerQueue(int producerCount, BlockingCollection<WorkItem> workQueue)
        {
            _workQueue = workQueue;
            _producerCount = producerCount;
            _producers = new Task[_producerCount];
        }

        public void Produce()
        {
            for (int i = 0; i < _producerCount; i++)
                _producers[i] = Task.Run(() => runProducer());

            Task.WhenAll(_producers).ContinueWith(_ => _workQueue.CompleteAdding());
        }

        private void runProducer()
        {
            // to demo just run for 5 seconds on each producer
            var fiveSecondsLater = DateTime.Now.AddSeconds(15);
            var random = new Random();
            while (!_workQueue.IsAddingCompleted && DateTime.Now <= fiveSecondsLater)
            {
                _workQueue.Add(new WorkItem(random.Next(21)));
                Thread.Sleep(100);
            }
        }
    }
}






 


public class MarketQuote
{
    private SortedDictionary<decimal, int> _bids;
    private SortedDictionary<decimal, int> _asks = new SortedDictionary<decimal, int>();
    public MarketQuote(string symbol, decimal bid, decimal ask, int askSize, int bidSize)
    {
        Symbol = symbol;
        _bids = new SortedDictionary<decimal, int> { { bid, bidSize } };
        _asks = new SortedDictionary<decimal, int> { { ask, askSize } };
    }

    public void AddBid(decimal bid, int quantity)
    {
        if (_bids.ContainsKey(bid))
            _bids[bid] += quantity;
        else
            _bids.Add(bid, quantity);
    }

    public void AddAsk(decimal ask, int quantity)
    {
        if (_asks.ContainsKey(ask)) _asks[ask] += quantity;
        else _asks.Add(ask, quantity);
    }

    public void ReduceAsk(decimal ask, int quantity)
    {
        _asks[ask] -= quantity;
        if (_asks[ask] == 0)
        {
            _asks.Remove(ask);
        }
        Last = ask;
    }

    public void ReduceBid(decimal bid, int quantity)
    {
        _bids[bid] -= quantity;
        if (_bids[bid] == 0)
        {
            _bids.Remove(bid);
        }
        Last = bid;
    }

    private bool hasBid() { return _bids.Count > 0; }
    private bool hasAsk() { return _asks.Count > 0; }

    public decimal Bid { get { return hasBid() ? _bids.Last().Key : 0; } }
    public int BidQuantity { get { return hasBid() ? _bids.Last().Value : 0; } }

    public decimal Ask { get { return hasAsk() ? _asks.First().Key : 0; } }
    public int AskQuantity { get { return hasAsk() ? _asks.First().Value : 0; } }

    public string Symbol { get; private set;}
    public decimal Last { get; private set; }
}

 // should be encapsualted in an interface.
 private void checkCrossTrade(MarketQuote quote)
    {
        MarketQuote currentQuote;
        if (_dict.TryGetValue(quote.Symbol, out currentQuote))
        {
            if (currentQuote.Ask == quote.Bid)
            {
                //currentQuote.Last = currentQuote.Ask; //? update last price first before reducing the ask!
                int diff = quote.BidQuantity - currentQuote.AskQuantity;
                if (diff > 0)
                    currentQuote.AddBid(quote.Bid, diff); // bid size is more than ask size; so add remaining bid.
                currentQuote.ReduceAsk(currentQuote.Ask, diff > 0 ? currentQuote.AskQuantity : quote.BidQuantity);

                //? Add the ask even though the bid has crossed over e.g. traded.
                currentQuote.AddAsk(quote.Ask, quote.AskQuantity);
            }
            else if (currentQuote.Bid == quote.Ask)
            {
                //currentQuote.Last = currentQuote.Bid;
                int diff = quote.AskQuantity - currentQuote.BidQuantity;
                if (diff > 0)
                    currentQuote.AddAsk(quote.Ask, diff);
                currentQuote.ReduceBid(currentQuote.Bid, diff > 0 ? currentQuote.BidQuantity : quote.AskQuantity);

                currentQuote.AddBid(quote.Bid, quote.BidQuantity);
            }
            else
            {
                currentQuote.AddAsk(quote.Ask, quote.AskQuantity);
                currentQuote.AddBid(quote.Bid, quote.BidQuantity);
            }
        }
        else
        {
            _dict.Add(quote.Symbol, quote);
            _bindingSource.Add(quote);
        }
        gridView.Refresh();
    }
    
 private void button1_Click(object sender, EventArgs e)
    {
        buttonA.Enabled = false;
        // var ignore = Task.Factory.StartNew(() => updateGrid2());
        //, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.FromCurrentSynchronizationContext());
        //IQuoteManager quoteManager = new QuoteManager(2);

        //var tasks = _filePaths.Select(file => 
        //                quoteManager.EnqueueQuote(() => new FileBroker(file, _quoteParser).GetQuotes()));

        //var processingTasks = (from t in tasks select updateGrid(t)).ToArray();
        //await Task.WhenAll(processingTasks);
        var brokerProducer = new ProducerQueue<MarketQuote>(2);

        _quoteQueue = brokerProducer.Produce(() =>
            _filePaths.Select(file => new FileBroker(file, _quoteParser).GetQuotes()).ToArray());

        foreach (var quote in _quoteQueue.GetConsumingEnumerable())
        {
            checkCrossTrade(quote);
        }

        Debug.WriteLine("Broker A is done");
        buttonA.Enabled = true;
    }

    public class QuoteParse : IQuoteParser
{
    private readonly char _separator;
    public QuoteParse(char separator)
    {
        _separator = separator;
    }
    public MarketQuote Parse(string line)
    {
        if (string.IsNullOrEmpty(line))
            throw new ArgumentException("Input cannot be empty");
        string[] fields = line.Split(_separator);
        string symbol = fields[0];
        if (string.IsNullOrEmpty(symbol))
            throw new InvalidDataException("Invalid symbol");

        decimal bid;
        if (!decimal.TryParse(fields[1], out bid))
            throw new InvalidDataException("Unable to parse Bid price");

        decimal ask;
        if (!decimal.TryParse(fields[2], out ask))
            throw new InvalidDataException("Unable to parse Ask price");

        int bidSize;
        if (!int.TryParse(fields[3], out bidSize))
            throw new InvalidDataException("Unable to parse Bid Quantity");

        int askSize;
        if (!int.TryParse(fields[4], out askSize))
            throw new InvalidDataException("Unable to parse Ask Quantity");

        return new MarketQuote(symbol, bid, ask, askSize, bidSize);
    }
}

/// <summary>
/// A dummy broker whose data is sent via a file.
/// </summary>
public class FileBroker : IBroker
{
    private readonly string _filePath;
    private readonly IQuoteParser _quoteParser;
    public FileBroker(string filePath, IQuoteParser quoteParse)
    {
        if (!File.Exists(filePath))
            throw new ArgumentException("invalid file path: " + filePath);
        _filePath = filePath;
        _quoteParser = quoteParse;
    }

    public IEnumerable<MarketQuote> GetQuotes()
    {
        // skip the header row
        return File.ReadLines(_filePath).Skip(1)
                                           .Where(line => !string.IsNullOrEmpty(line))
                                           .Select(line => _quoteParser.Parse(line));
    }
}

    using System;
    using System.Collections.Concurrent;
    using System.Threading;
    using System.Threading.Tasks;
    using System.Windows.Forms;
    
    namespace PCQueueDemo
    {
        public partial class Form1 : Form
        {
            // to consume results that are depedent
            private ConcurrentDictionary<int, WorkItem> _dict = new ConcurrentDictionary<int, WorkItem>();

        private readonly object _sync = new object();
        private BindingSource _bindingSource = new BindingSource();
        private TaskScheduler _ui; 
        private ProducerConsumerQueue _producers;
        private BlockingCollection<WorkItem> _workQueue;
        public Form1()
        {
            InitializeComponent();
            _workQueue = new BlockingCollection<WorkItem>();

            // create 3 producers which will produce work item to the work queue.
            _producers = new ProducerConsumerQueue(3, _workQueue);

            // capture the UI context in order to update data grid.
            _ui = TaskScheduler.FromCurrentSynchronizationContext();

            // bind the data grid
            dataGridView.DataSource = _bindingSource;
            dataGridView.AutoSize = true;
            dataGridView.Dock = DockStyle.Fill;
        }

        // Run button click handler
        private void button1_Click(object sender, EventArgs e)
        {
            _producers.Produce(); // start producing on each producer thread.
            var ignore = Task.Run(() => updateGrid()); // consume the producer result to update the grid.
        }

        private void updateGrid()
        {
            foreach (var item in _workQueue.GetConsumingEnumerable())
            {
                WorkItem copy = item;

                //? Is it a waste to use a thread for each result?
                
                Task.Factory.StartNew(() =>
                {
                    WorkItem localItem;
                    if (_dict.TryGetValue(item.Number, out localItem))
                    {
                        //? how to avoid locking if the update here takes longer than simply incrementing the counter?
                        lock (_sync) localItem.Counter++;
                    }
                    else if (_dict.TryAdd(copy.Number, copy))
                    {
                        //? Is this negligible since the lock should be very short?
                        lock(_sync) _bindingSource.Add(copy);
                    }
                    dataGridView.Refresh();
                }, CancellationToken.None, TaskCreationOptions.None, _ui);
            }
        }
    }

    // a demo work item
    public class WorkItem
    {
        public WorkItem(int number)
        {
            Number = number;
            Counter = 1; // 1st appearance
        }
        public int Number { get; private set; }
        public int Counter { get; set; }
    }

    //? This is just a producer queue; how to encapsulate the consumer together with producer while allowing it to update the UI?
    public class ProducerConsumerQueue
    {
        private readonly int _producerCount;
        private Task[] _producers;
        private BlockingCollection<WorkItem> _workQueue;

        public ProducerConsumerQueue(int producerCount, BlockingCollection<WorkItem> workQueue)
        {
            _workQueue = workQueue;
            _producerCount = producerCount;
            _producers = new Task[_producerCount];
        }

        public void Produce()
        {
            for (int i = 0; i < _producerCount; i++)
                _producers[i] = Task.Run(() => runProducer());

            Task.WhenAll(_producers).ContinueWith(_ => _workQueue.CompleteAdding());
        }

        private void runProducer()
        {
            // to demo just run for 5 seconds on each producer
            var fiveSecondsLater = DateTime.Now.AddSeconds(15);
            var random = new Random();
            while (!_workQueue.IsAddingCompleted && DateTime.Now <= fiveSecondsLater)
            {
                _workQueue.Add(new WorkItem(random.Next(21)));
                Thread.Sleep(100);
            }
        }
    }
}



    

    using System;
    using System.Collections.Concurrent;
    using System.Threading;
    using System.Threading.Tasks;
    using System.Windows.Forms;
    
    namespace PCQueueDemo
    {
        public partial class Form1 : Form
        {
            // to consume results that are depedent
            private ConcurrentDictionary<int, WorkItem> _dict = new ConcurrentDictionary<int, WorkItem>();

        private readonly object _sync = new object();
        private BindingSource _bindingSource = new BindingSource();
        private TaskScheduler _ui; 
        private ProducerConsumerQueue _producers;
        private BlockingCollection<WorkItem> _workQueue;
        public Form1()
        {
            InitializeComponent();
            _workQueue = new BlockingCollection<WorkItem>();

            // create 3 producers which will produce work item to the work queue.
            _producers = new ProducerConsumerQueue(3, _workQueue);

            // capture the UI context in order to update data grid.
            _ui = TaskScheduler.FromCurrentSynchronizationContext();

            // bind the data grid
            dataGridView.DataSource = _bindingSource;
            dataGridView.AutoSize = true;
            dataGridView.Dock = DockStyle.Fill;
        }

        // Run button click handler
        private void button1_Click(object sender, EventArgs e)
        {
            _producers.Produce(); // start producing on each producer thread.
            var ignore = Task.Run(() => updateGrid()); // consume the producer result to update the grid.
        }

        private void updateGrid()
        {
            foreach (var item in _workQueue.GetConsumingEnumerable())
            {
                WorkItem copy = item;

                //? Is it a waste to use a thread for each result?
                
                Task.Factory.StartNew(() =>
                {
                    WorkItem localItem;
                    if (_dict.TryGetValue(item.Number, out localItem))
                    {
                        //? how to avoid locking if the update here takes longer than simply incrementing the counter?
                        lock (_sync) localItem.Counter++;
                    }
                    else if (_dict.TryAdd(copy.Number, copy))
                    {
                        //? Is this negligible since the lock should be very short?
                        lock(_sync) _bindingSource.Add(copy);
                    }
                    dataGridView.Refresh();
                }, CancellationToken.None, TaskCreationOptions.None, _ui);
            }
        }
    }

    // a demo work item
    public class WorkItem
    {
        public WorkItem(int number)
        {
            Number = number;
            Counter = 1; // 1st appearance
        }
        public int Number { get; private set; }
        public int Counter { get; set; }
    }

    //? This is just a producer queue; how to encapsulate the consumer together with producer while allowing it to update the UI?
    public class ProducerConsumerQueue
    {
        private readonly int _producerCount;
        private Task[] _producers;
        private BlockingCollection<WorkItem> _workQueue;

        public ProducerConsumerQueue(int producerCount, BlockingCollection<WorkItem> workQueue)
        {
            _workQueue = workQueue;
            _producerCount = producerCount;
            _producers = new Task[_producerCount];
        }

        public void Produce()
        {
            for (int i = 0; i < _producerCount; i++)
                _producers[i] = Task.Run(() => runProducer());

            Task.WhenAll(_producers).ContinueWith(_ => _workQueue.CompleteAdding());
        }

        private void runProducer()
        {
            // to demo just run for 5 seconds on each producer
            var fiveSecondsLater = DateTime.Now.AddSeconds(15);
            var random = new Random();
            while (!_workQueue.IsAddingCompleted && DateTime.Now <= fiveSecondsLater)
            {
                _workQueue.Add(new WorkItem(random.Next(21)));
                Thread.Sleep(100);
            }
        }
    }
}



    

    using System;
    using System.Collections.Concurrent;
    using System.Threading;
    using System.Threading.Tasks;
    using System.Windows.Forms;
    
    namespace PCQueueDemo
    {
        public partial class Form1 : Form
        {
            // to consume results that are depedent
            private ConcurrentDictionary<int, WorkItem> _dict = new ConcurrentDictionary<int, WorkItem>();

        private readonly object _sync = new object();
        private BindingSource _bindingSource = new BindingSource();
        private TaskScheduler _ui; 
        private ProducerConsumerQueue _producers;
        private BlockingCollection<WorkItem> _workQueue;
        public Form1()
        {
            InitializeComponent();
            _workQueue = new BlockingCollection<WorkItem>();

            // create 3 producers which will produce work item to the work queue.
            _producers = new ProducerConsumerQueue(3, _workQueue);

            // capture the UI context in order to update data grid.
            _ui = TaskScheduler.FromCurrentSynchronizationContext();

            // bind the data grid
            dataGridView.DataSource = _bindingSource;
            dataGridView.AutoSize = true;
            dataGridView.Dock = DockStyle.Fill;
        }

        // Run button click handler
        private void button1_Click(object sender, EventArgs e)
        {
            _producers.Produce(); // start producing on each producer thread.
            var ignore = Task.Run(() => updateGrid()); // consume the producer result to update the grid.
        }

        private void updateGrid()
        {
            foreach (var item in _workQueue.GetConsumingEnumerable())
            {
                WorkItem copy = item;

                //? Is it a waste to use a thread for each result?
                
                Task.Factory.StartNew(() =>
                {
                    WorkItem localItem;
                    if (_dict.TryGetValue(item.Number, out localItem))
                    {
                        //? how to avoid locking if the update here takes longer than simply incrementing the counter?
                        lock (_sync) localItem.Counter++;
                    }
                    else if (_dict.TryAdd(copy.Number, copy))
                    {
                        //? Is this negligible since the lock should be very short?
                        lock(_sync) _bindingSource.Add(copy);
                    }
                    dataGridView.Refresh();
                }, CancellationToken.None, TaskCreationOptions.None, _ui);
            }
        }
    }

    // a demo work item
    public class WorkItem
    {
        public WorkItem(int number)
        {
            Number = number;
            Counter = 1; // 1st appearance
        }
        public int Number { get; private set; }
        public int Counter { get; set; }
    }

    //? This is just a producer queue; how to encapsulate the consumer together with producer while allowing it to update the UI?
    public class ProducerConsumerQueue
    {
        private readonly int _producerCount;
        private Task[] _producers;
        private BlockingCollection<WorkItem> _workQueue;

        public ProducerConsumerQueue(int producerCount, BlockingCollection<WorkItem> workQueue)
        {
            _workQueue = workQueue;
            _producerCount = producerCount;
            _producers = new Task[_producerCount];
        }

        public void Produce()
        {
            for (int i = 0; i < _producerCount; i++)
                _producers[i] = Task.Run(() => runProducer());

            Task.WhenAll(_producers).ContinueWith(_ => _workQueue.CompleteAdding());
        }

        private void runProducer()
        {
            // to demo just run for 5 seconds on each producer
            var fiveSecondsLater = DateTime.Now.AddSeconds(15);
            var random = new Random();
            while (!_workQueue.IsAddingCompleted && DateTime.Now <= fiveSecondsLater)
            {
                _workQueue.Add(new WorkItem(random.Next(21)));
                Thread.Sleep(100);
            }
        }
    }
}
 








public class MarketQuote
{
    private SortedDictionary<decimal, int> _bids;
    private SortedDictionary<decimal, int> _asks = new SortedDictionary<decimal, int>();
    public MarketQuote(string symbol, decimal bid, decimal ask, int askSize, int bidSize)
    {
        Symbol = symbol;
        _bids = new SortedDictionary<decimal, int> { { bid, bidSize } };
        _asks = new SortedDictionary<decimal, int> { { ask, askSize } };
    }

    public void AddBid(decimal bid, int quantity)
    {
        if (_bids.ContainsKey(bid))
            _bids[bid] += quantity;
        else
            _bids.Add(bid, quantity);
    }

    public void AddAsk(decimal ask, int quantity)
    {
        if (_asks.ContainsKey(ask)) _asks[ask] += quantity;
        else _asks.Add(ask, quantity);
    }

    public void ReduceAsk(decimal ask, int quantity)
    {
        _asks[ask] -= quantity;
        if (_asks[ask] == 0)
        {
            _asks.Remove(ask);
        }
        Last = ask;
    }

    public void ReduceBid(decimal bid, int quantity)
    {
        _bids[bid] -= quantity;
        if (_bids[bid] == 0)
        {
            _bids.Remove(bid);
        }
        Last = bid;
    }

    private bool hasBid() { return _bids.Count > 0; }
    private bool hasAsk() { return _asks.Count > 0; }

    public decimal Bid { get { return hasBid() ? _bids.Last().Key : 0; } }
    public int BidQuantity { get { return hasBid() ? _bids.Last().Value : 0; } }

    public decimal Ask { get { return hasAsk() ? _asks.First().Key : 0; } }
    public int AskQuantity { get { return hasAsk() ? _asks.First().Value : 0; } }

    public string Symbol { get; private set;}
    public decimal Last { get; private set; }
}

 // should be encapsualted in an interface.
 private void checkCrossTrade(MarketQuote quote)
    {
        MarketQuote currentQuote;
        if (_dict.TryGetValue(quote.Symbol, out currentQuote))
        {
            if (currentQuote.Ask == quote.Bid)
            {
                //currentQuote.Last = currentQuote.Ask; //? update last price first before reducing the ask!
                int diff = quote.BidQuantity - currentQuote.AskQuantity;
                if (diff > 0)
                    currentQuote.AddBid(quote.Bid, diff); // bid size is more than ask size; so add remaining bid.
                currentQuote.ReduceAsk(currentQuote.Ask, diff > 0 ? currentQuote.AskQuantity : quote.BidQuantity);

                //? Add the ask even though the bid has crossed over e.g. traded.
                currentQuote.AddAsk(quote.Ask, quote.AskQuantity);
            }
            else if (currentQuote.Bid == quote.Ask)
            {
                //currentQuote.Last = currentQuote.Bid;
                int diff = quote.AskQuantity - currentQuote.BidQuantity;
                if (diff > 0)
                    currentQuote.AddAsk(quote.Ask, diff);
                currentQuote.ReduceBid(currentQuote.Bid, diff > 0 ? currentQuote.BidQuantity : quote.AskQuantity);

                currentQuote.AddBid(quote.Bid, quote.BidQuantity);
            }
            else
            {
                currentQuote.AddAsk(quote.Ask, quote.AskQuantity);
                currentQuote.AddBid(quote.Bid, quote.BidQuantity);
            }
        }
        else
        {
            _dict.Add(quote.Symbol, quote);
            _bindingSource.Add(quote);
        }
        gridView.Refresh();
    }
    
 private void button1_Click(object sender, EventArgs e)
    {
        buttonA.Enabled = false;
        // var ignore = Task.Factory.StartNew(() => updateGrid2());
        //, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.FromCurrentSynchronizationContext());
        //IQuoteManager quoteManager = new QuoteManager(2);

        //var tasks = _filePaths.Select(file => 
        //                quoteManager.EnqueueQuote(() => new FileBroker(file, _quoteParser).GetQuotes()));

        //var processingTasks = (from t in tasks select updateGrid(t)).ToArray();
        //await Task.WhenAll(processingTasks);
        var brokerProducer = new ProducerQueue<MarketQuote>(2);

        _quoteQueue = brokerProducer.Produce(() =>
            _filePaths.Select(file => new FileBroker(file, _quoteParser).GetQuotes()).ToArray());

        foreach (var quote in _quoteQueue.GetConsumingEnumerable())
        {
            checkCrossTrade(quote);
        }

        Debug.WriteLine("Broker A is done");
        buttonA.Enabled = true;
    }

    public class QuoteParse : IQuoteParser
{
    private readonly char _separator;
    public QuoteParse(char separator)
    {
        _separator = separator;
    }
    public MarketQuote Parse(string line)
    {
        if (string.IsNullOrEmpty(line))
            throw new ArgumentException("Input cannot be empty");
        string[] fields = line.Split(_separator);
        string symbol = fields[0];
        if (string.IsNullOrEmpty(symbol))
            throw new InvalidDataException("Invalid symbol");

        decimal bid;
        if (!decimal.TryParse(fields[1], out bid))
            throw new InvalidDataException("Unable to parse Bid price");

        decimal ask;
        if (!decimal.TryParse(fields[2], out ask))
            throw new InvalidDataException("Unable to parse Ask price");

        int bidSize;
        if (!int.TryParse(fields[3], out bidSize))
            throw new InvalidDataException("Unable to parse Bid Quantity");

        int askSize;
        if (!int.TryParse(fields[4], out askSize))
            throw new InvalidDataException("Unable to parse Ask Quantity");

        return new MarketQuote(symbol, bid, ask, askSize, bidSize);
    }
}

/// <summary>
/// A dummy broker whose data is sent via a file.
/// </summary>
public class FileBroker : IBroker
{
    private readonly string _filePath;
    private readonly IQuoteParser _quoteParser;
    public FileBroker(string filePath, IQuoteParser quoteParse)
    {
        if (!File.Exists(filePath))
            throw new ArgumentException("invalid file path: " + filePath);
        _filePath = filePath;
        _quoteParser = quoteParse;
    }

    public IEnumerable<MarketQuote> GetQuotes()
    {
        // skip the header row
        return File.ReadLines(_filePath).Skip(1)
                                           .Where(line => !string.IsNullOrEmpty(line))
                                           .Select(line => _quoteParser.Parse(line));
    }
}

1. Can explicit locking be avoided?

2. How to encapsulate consumer and producer within a single object e.g. ProducerConsumerQueue which currently is really a producer queue.

3. Currently I use a separate task to process each result; is this a waste?

    using System;
    using System.Collections.Concurrent;
    using System.Threading;
    using System.Threading.Tasks;
    using System.Windows.Forms;
    
    namespace PCQueueDemo
    {
        public partial class Form1 : Form
        {
            // to consume results that are depedent
            private ConcurrentDictionary<int, WorkItem> _dict = new ConcurrentDictionary<int, WorkItem>();

        private readonly object _sync = new object();
        private BindingSource _bindingSource = new BindingSource();
        private TaskScheduler _ui; 
        private ProducerConsumerQueue _producers;
        private BlockingCollection<WorkItem> _workQueue;
        public Form1()
        {
            InitializeComponent();
            _workQueue = new BlockingCollection<WorkItem>();

            // create 3 producers which will produce work item to the work queue.
            _producers = new ProducerConsumerQueue(3, _workQueue);

            // capture the UI context in order to update data grid.
            _ui = TaskScheduler.FromCurrentSynchronizationContext();

            // bind the data grid
            dataGridView.DataSource = _bindingSource;
            dataGridView.AutoSize = true;
            dataGridView.Dock = DockStyle.Fill;
        }

        // Run button click handler
        private void button1_Click(object sender, EventArgs e)
        {
            _producers.Produce(); // start producing on each producer thread.
            var ignore = Task.Run(() => updateGrid()); // consume the producer result to update the grid.
        }

        private void updateGrid()
        {
            foreach (var item in _workQueue.GetConsumingEnumerable())
            {
                WorkItem copy = item;

                //? Is it a waste to use a thread for each result?
                
                Task.Factory.StartNew(() =>
                {
                    WorkItem localItem;
                    if (_dict.ContainsKey(item.Number) && _dict.TryGetValue(item.Number, out localItem))
                    {
                        //? how to avoid locking if the update here takes longer than simply incrementing the counter?
                        lock (_sync) localItem.Counter++;
                    }
                    else if (_dict.TryAdd(copy.Number, copy))
                    {
                        //? Is this negligible since the lock should be very short?
                        lock(_sync) _bindingSource.Add(copy);
                    }
                    dataGridView.Refresh();
                }, CancellationToken.None, TaskCreationOptions.None, _ui);
            }
        }
    }

    // a demo work item
    public class WorkItem
    {
        public WorkItem(int number)
        {
            Number = number;
            Counter = 1; // 1st appearance
        }
        public int Number { get; private set; }
        public int Counter { get; set; }
    }

    //? This is just a producer queue; how to encapsulate the consumer together with producer while allowing it to update the UI?
    public class ProducerConsumerQueue
    {
        private readonly int _producerCount;
        private Task[] _producers;
        private BlockingCollection<WorkItem> _workQueue;

        public ProducerConsumerQueue(int producerCount, BlockingCollection<WorkItem> workQueue)
        {
            _workQueue = workQueue;
            _producerCount = producerCount;
            _producers = new Task[_producerCount];
        }

        public void Produce()
        {
            for (int i = 0; i < _producerCount; i++)
                _producers[i] = Task.Run(() => runProducer());

            Task.WhenAll(_producers).ContinueWith(_ => _workQueue.CompleteAdding());
        }

        private void runProducer()
        {
            // to demo just run for 5 seconds on each producer
            var fiveSecondsLater = DateTime.Now.AddSeconds(15);
            var random = new Random();
            while (!_workQueue.IsAddingCompleted && DateTime.Now <= fiveSecondsLater)
            {
                _workQueue.Add(new WorkItem(random.Next(21)));
                Thread.Sleep(100);
            }
        }
    }
}



    

1. Can explicit locking be avoided?

2. How to encapsulate consumer and producer within a single object e.g. ProducerConsumerQueue which currently is really a producer queue.

3. Currently I use a separate task to process each result especially given it ultimately has to post back to the UI thread; is this a waste?

    using System;
    using System.Collections.Concurrent;
    using System.Threading;
    using System.Threading.Tasks;
    using System.Windows.Forms;
    
    namespace PCQueueDemo
    {
        public partial class Form1 : Form
        {
            // to consume results that are depedent
            private ConcurrentDictionary<int, WorkItem> _dict = new ConcurrentDictionary<int, WorkItem>();

        private readonly object _sync = new object();
        private BindingSource _bindingSource = new BindingSource();
        private TaskScheduler _ui; 
        private ProducerConsumerQueue _producers;
        private BlockingCollection<WorkItem> _workQueue;
        public Form1()
        {
            InitializeComponent();
            _workQueue = new BlockingCollection<WorkItem>();

            // create 3 producers which will produce work item to the work queue.
            _producers = new ProducerConsumerQueue(3, _workQueue);

            // capture the UI context in order to update data grid.
            _ui = TaskScheduler.FromCurrentSynchronizationContext();

            // bind the data grid
            dataGridView.DataSource = _bindingSource;
            dataGridView.AutoSize = true;
            dataGridView.Dock = DockStyle.Fill;
        }

        // Run button click handler
        private void button1_Click(object sender, EventArgs e)
        {
            _producers.Produce(); // start producing on each producer thread.
            var ignore = Task.Run(() => updateGrid()); // consume the producer result to update the grid.
        }

        private void updateGrid()
        {
            foreach (var item in _workQueue.GetConsumingEnumerable())
            {
                WorkItem copy = item;

                //? Is it a waste to use a thread for each result?
                
                Task.Factory.StartNew(() =>
                {
                    WorkItem localItem;
                    if (_dict.TryGetValue(item.Number, out localItem))
                    {
                        //? how to avoid locking if the update here takes longer than simply incrementing the counter?
                        lock (_sync) localItem.Counter++;
                    }
                    else if (_dict.TryAdd(copy.Number, copy))
                    {
                        //? Is this negligible since the lock should be very short?
                        lock(_sync) _bindingSource.Add(copy);
                    }
                    dataGridView.Refresh();
                }, CancellationToken.None, TaskCreationOptions.None, _ui);
            }
        }
    }

    // a demo work item
    public class WorkItem
    {
        public WorkItem(int number)
        {
            Number = number;
            Counter = 1; // 1st appearance
        }
        public int Number { get; private set; }
        public int Counter { get; set; }
    }

    //? This is just a producer queue; how to encapsulate the consumer together with producer while allowing it to update the UI?
    public class ProducerConsumerQueue
    {
        private readonly int _producerCount;
        private Task[] _producers;
        private BlockingCollection<WorkItem> _workQueue;

        public ProducerConsumerQueue(int producerCount, BlockingCollection<WorkItem> workQueue)
        {
            _workQueue = workQueue;
            _producerCount = producerCount;
            _producers = new Task[_producerCount];
        }

        public void Produce()
        {
            for (int i = 0; i < _producerCount; i++)
                _producers[i] = Task.Run(() => runProducer());

            Task.WhenAll(_producers).ContinueWith(_ => _workQueue.CompleteAdding());
        }

        private void runProducer()
        {
            // to demo just run for 5 seconds on each producer
            var fiveSecondsLater = DateTime.Now.AddSeconds(15);
            var random = new Random();
            while (!_workQueue.IsAddingCompleted && DateTime.Now <= fiveSecondsLater)
            {
                _workQueue.Add(new WorkItem(random.Next(21)));
                Thread.Sleep(100);
            }
        }
    }
}