using KLHZ.Trader.Core.Common.Messaging.Contracts; using KLHZ.Trader.Core.Common.Messaging.Contracts.Messages; using System.Collections.Concurrent; using System.Threading.Channels; namespace KLHZ.Trader.Core.Common.Messaging.Services { public class DataBus : IDataBus { private readonly ConcurrentDictionary> _candlesChannels = new(); private readonly ConcurrentDictionary> _priceChannels = new(); private readonly ConcurrentDictionary> _commandChannels = new(); private readonly ConcurrentDictionary> _chatMessages = new(); public bool AddChannel(Channel channel) { return _chatMessages.TryAdd(Guid.NewGuid().ToString(), channel); } public bool AddChannel(string key, Channel channel) { return _priceChannels.TryAdd(key, channel); } public bool AddChannel(string key, Channel channel) { return _candlesChannels.TryAdd(key, channel); } public bool AddChannel(string key, Channel channel) { return _commandChannels.TryAdd(key, channel); } public async Task BroadcastNewPrice(INewPriceMessage newPriceMessage) { foreach (var channel in _priceChannels.Values) { await channel.Writer.WriteAsync(newPriceMessage); } } public async Task BroadcastNewCandle(INewCandle newPriceMessage) { foreach (var channel in _candlesChannels.Values) { await channel.Writer.WriteAsync(newPriceMessage); } } public async Task BroadcastCommand(TradeCommand command) { foreach (var channel in _commandChannels.Values) { await channel.Writer.WriteAsync(command); } } public async Task BroadcastCommand(MessageForAdmin message) { foreach (var channel in _chatMessages.Values) { await channel.Writer.WriteAsync(message); } } } }