Fix parallel processing.

Author: mikasoukhov

Connectors/Tinkoff/TinkoffMessageAdapter.cs

@@ -3,6 +3,8 @@ namespace StockSharp.Tinkoff;
 using Ecng.ComponentModel;
 using Ecng.Net;
 
+using Nito.AsyncEx;
+
 public partial class TinkoffMessageAdapter
 {
 	private GrpcChannel _channel;
@@ -16,6 +18,8 @@ public partial class TinkoffMessageAdapter
 	private static TimeSpan GetCurrentDelay(TimeSpan currentDelay)
 		=> currentDelay.Multiply(2).Min(_maxDelay);
 
+	private readonly AsyncLock _lock = new();
+
 	/// <summary>
 	/// Initializes a new instance of the <see cref="TinkoffMessageAdapter"/>.
 	/// </summary>

Connectors/Tinkoff/TinkoffMessageAdapter_MarketData.cs

@@ -536,7 +536,7 @@ protected override async ValueTask OnTFCandlesSubscriptionAsync(MarketDataMessag
 			if (!TryGetAndRemove(mdMsg.OriginalTransactionId, out var t))
 				return;
 
-			await _mdStream.RequestStream.WriteAsync(new()
+			await WriteMdRequest(new()
 			{
 				SubscribeCandlesRequest = new()
 				{
@@ -555,7 +555,7 @@ await _mdStream.RequestStream.WriteAsync(new()
 	}
 
 	private Task SubscribeCandles(MarketDataMessage mdMsg, CancellationToken cancellationToken)
-		=> _mdStream.RequestStream.WriteAsync(new()
+		=> WriteMdRequest(new()
 		{
 			SubscribeCandlesRequest = new()
 			{
@@ -673,7 +673,7 @@ protected override async ValueTask OnMarketDepthSubscriptionAsync(MarketDataMess
 			if (!TryGetAndRemove(mdMsg.OriginalTransactionId, out var t))
 				return;
 
-			await _mdStream.RequestStream.WriteAsync(new()
+			await WriteMdRequest(new()
 			{
 				SubscribeOrderBookRequest = new()
 				{
@@ -698,7 +698,7 @@ await _mdStream.RequestStream.WriteAsync(new()
 	}
 
 	private Task SubscribeMarketDepth(MarketDataMessage mdMsg, CancellationToken cancellationToken)
-		=> _mdStream.RequestStream.WriteAsync(new()
+		=> WriteMdRequest(new()
 		{
 			SubscribeOrderBookRequest = new()
 			{
@@ -741,7 +741,7 @@ protected override async ValueTask OnTicksSubscriptionAsync(MarketDataMessage md
 			if (!TryGetAndRemove(mdMsg.OriginalTransactionId, out var t))
 				return;
 
-			await _mdStream.RequestStream.WriteAsync(new()
+			await WriteMdRequest(new()
 			{
 				SubscribeTradesRequest = new()
 				{
@@ -765,7 +765,7 @@ await _mdStream.RequestStream.WriteAsync(new()
 	}
 
 	private Task SubscribeTicks(MarketDataMessage mdMsg, CancellationToken cancellationToken)
-		=> _mdStream.RequestStream.WriteAsync(new()
+		=> WriteMdRequest(new()
 		{
 			SubscribeTradesRequest = new()
 			{
@@ -807,7 +807,7 @@ protected override async ValueTask OnLevel1SubscriptionAsync(MarketDataMessage m
 			if (!TryGetAndRemove(mdMsg.OriginalTransactionId, out var t))
 				return;
 
-			await _mdStream.RequestStream.WriteAsync(new()
+			await WriteMdRequest(new()
 			{
 				SubscribeLastPriceRequest = new()
 				{
@@ -836,7 +836,7 @@ await _mdStream.RequestStream.WriteAsync(new()
 	}
 
 	private Task SubscribeLevel1(MarketDataMessage mdMsg, CancellationToken cancellationToken)
-		=> _mdStream.RequestStream.WriteAsync(new()
+		=> WriteMdRequest(new()
 		{
 			SubscribeLastPriceRequest = new()
 			{
@@ -861,4 +861,10 @@ private Task SubscribeLevel1(MarketDataMessage mdMsg, CancellationToken cancella
 				SubscriptionAction = SubscriptionAction.Subscribe,
 			},
 		}, cancellationToken);
+
+	private async Task WriteMdRequest(MarketDataRequest message, CancellationToken cancellationToken)
+	{
+		using var _ = await _lock.LockAsync(cancellationToken);
+		await _mdStream.RequestStream.WriteAsync(message, cancellationToken);
+	}
 }