using System; using System.Collections.Generic; using DotNetCore.CAP.Messages; using Microsoft.Extensions.Logging; namespace DotNetCore.CAP.Test.FakeInMemoryQueue { internal class InMemoryQueue { private readonly ILogger _logger; private static readonly object Lock = new object(); private readonly Dictionary, List)> _groupTopics; public Dictionary Messages { get; } public InMemoryQueue(ILogger logger) { _logger = logger; _groupTopics = new Dictionary, List)>(); Messages = new Dictionary(); } public void Subscribe(string groupId, Action received, string topic) { lock (Lock) { if (_groupTopics.ContainsKey(groupId)) { var topics = _groupTopics[groupId]; if (!topics.Item2.Contains(topic)) { topics.Item2.Add(topic); } } else { _groupTopics.Add(groupId, (received, new List { topic })); } } } public void ClearSubscriber() { _groupTopics.Clear(); } public void Send(string topic, TransportMessage message) { Messages.Add(topic, message); foreach (var groupTopic in _groupTopics) { if (groupTopic.Value.Item2.Contains(topic)) { try { groupTopic.Value.Item1?.Invoke(message); } catch (Exception e) { _logger.LogError(e, $"Consumption message raises an exception. Group-->{groupTopic.Key} Name-->{topic}"); } } } } } }