docker files fixes from liaqat

src/Managing.Infrastructure.Exchanges/ExchangeStream.cs

@@ -0,0 +1,41 @@
+using Binance.Net.Interfaces.Clients;
+using Managing.Application.Abstractions.Services;
+using Managing.Application.Shared;
+using Managing.Domain.Candles;
+using Managing.Infrastructure.Exchanges.Helpers;
+using Microsoft.Extensions.Logging;
+using static Managing.Common.Enums;
+
+namespace Managing.Infrastructure.Exchanges;
+
+public class ExchangeStream : IExchangeStream
+{
+    private readonly ILogger<StreamService> logger;
+    private readonly IBinanceSocketClient _binanceSocketClient;
+
+    public ExchangeStream(IBinanceSocketClient binanceSocketClient, ILogger<StreamService> logger)
+    {
+        _binanceSocketClient = binanceSocketClient;
+        this.logger = logger;
+    }
+
+    public async Task StartBinanceWorker(Ticker ticker, Func<Candle, Task> action)
+    {
+        logger.LogInformation($"Starting binance worker for {ticker}");
+
+        await _binanceSocketClient.SpotApi.ExchangeData.SubscribeToKlineUpdatesAsync(BinanceHelpers.ToBinanceTicker(ticker), Binance.Net.Enums.KlineInterval.OneSecond, candle =>
+        {
+            if (candle.Data.Data?.Final == true)
+            {
+                //action((candle) => { CandleHelpers.Map(candle.Data.Data, ticker, Timeframe.FiveMinutes)});
+                action(CandleHelpers.Map(candle.Data.Data, ticker, Timeframe.FiveMinutes));
+            }
+        });
+    }
+
+    public async Task StopBinanceWorker()
+    {
+        logger.LogInformation($"Stoping all Binance worker subscription");
+        await _binanceSocketClient.UnsubscribeAllAsync();
+    }
+}