Fixed disposing of streams which fixed bugs of live streams still pub…

…lishing data upon exit
bleunguts · Oct 4, 2023 · d367e75 · d367e75
1 parent cf1ff27
commit d367e75
Showing 1 changed file with 13 additions and 2 deletions.
diff --git a/ProjectX.GatewayAPI/BackgroundServices/FXPricingService.cs b/ProjectX.GatewayAPI/BackgroundServices/FXPricingService.cs
@@ -15,6 +15,8 @@ public class FXPricingService : BackgroundService
         private readonly FXTasksChannel _fxTaskChannel;
         private readonly IFXMarketService _fxMarketService;
 
+        private Dictionary<string, IDisposable> _disposables = new Dictionary<string, IDisposable>(); 
+
         public FXPricingService(ILogger<FXPricingService> logger,
             FXTasksChannel fXTasksChannel,
             IFXMarketService fxMarketService,           
@@ -43,12 +45,21 @@ await foreach (var request in _fxTaskChannel.ReadAllAsync(stoppingToken))
                     switch(request.Mode)
                     {
                         case FXRateMode.Subscribe:
-                            _fxMarketService.StreamSpotPricesFor(request)
+                            var disposable = _fxMarketService.StreamSpotPricesFor(request)
                                             .Subscribe(priceResponse => hubContext.Clients.All.PushFxRate(new SpotPriceResult(request.ClientName, priceResponse.Timestamp, priceResponse.Value)));
+                            if(!_disposables.TryAdd(request.CurrencyPair, disposable))
+                            {
+                                _logger.LogWarning($"Disposable stream already added {request.CurrencyPair}");
+                            }
                             break;
                        case FXRateMode.Unsubscribe:
                             _fxMarketService.UnStream(request.CurrencyPair);
-                            hubContext.Clients.All.StopFxRate(request.CurrencyPair);
+                            if(_disposables.TryGetValue(request.CurrencyPair, out IDisposable d))
+                            {
+                                d.Dispose();
+                                _disposables.Remove(request.CurrencyPair);
+                            }
+                            await hubContext.Clients.All.StopFxRate(request.CurrencyPair);
                             break;
                         default:
                             throw new NotSupportedException($"SpotRequest type {request.Mode} is not supported");