How do I add lock/async/await or threading concepts in this code

See more:

C#

threads

I want to call

(1) GetMessages() this will keep bringing the messages in separate thread

(2) GetOrderData() this will fetch the order details and insert them in datagrid and the its task is finished here

(3) UpdateOrderBook() this will now update the existing datagrid (which already has some data from step 2). Here in this method I want to check the queue and have to iterate the concurrentqueue then do some processing and then insert that record in the existing grid.

The two process (1,3) would run asynchronously but for the first time it should process in above order.

I got stuck in making them run asynchronously.


This works perfectly for the first time. But when I change the product and reconnect the things mess up and the data is not processed properly.

I am new to multi-threading so not sure if I need to use lock or async/await or something else. What am I doing wrong in the above code?

What I have tried:

void OnReceivingFeedMessage(message)
    {
         concurrentQueue.Enqueue(message);
      
         if (!messageStreamStarted)   // only first time get order book
         {
             messageStreamStarted = true;
             GetOrderBookData();
         }
    }

    private void GetOrderBookData()
    {
        MarketData m = new MarketData();
        ProductOrderBook p = m.GetProductOrderBook(productId);           
        bidsList = p.bids;
        asksList = p.asks;
        isOrderBookUpdated = true;

        Task task3 = Task.Run(() => KickStartToProcessQueue());         
    }

    private void KickStartToProcessQueue()
    {
        while (threadProcessQueueExist)
        {
            int recordCountNew = concurrentQueue.Count();
            if (recordCountNew != 0)
            {
                if (isOrderBookUpdated)
                {
                    ProcessQueueMessages();
                }
            }
        }
    }

    private void ProcessQueueMessages()
    {
        if (!concurrentQueue.IsEmpty)
        {
            string jsonString;
            while (concurrentQueue.TryDequeue(out jsonString))
            {
              // have to insert the record in existing order book
            }
         }
    }

// The code written on product selectedindex change

    private void CloseAndReconnectToGetWebsocketFeed()
    { 
        w.CloseWebsocketConnection();                 
        messageStreamStarted = false;            
        isOrderBookUpdated = false;
        ConcurrentQueue<string> wssMessagesQueue = new ConcurrentQueue<string>();
        concurrentQueue = wssMessagesQueue;

        ConnectAndGetWebsocketFeedMessages(); // this calls OnReceivingFeedMessage
    }