src/Coast.RabbitMQ/EventBusRabbitMQ.cs

这里只显示每个文件的开头 60 行。登录后可以解锁完整代码。

微信扫码关注公众号，代码整理好后第一时间通知你。

namespace Coast.RabbitMQ
{
    using System;
    using System.Net.Sockets;
    using System.Text;
    using System.Text.Json;
    using System.Threading;
    using System.Threading.Tasks;
    using Coast.Core;
    using Coast.Core.DataLayer;
    using Coast.Core.EventBus;
    using Coast.Core.Util;
    using global::RabbitMQ.Client;
    using global::RabbitMQ.Client.Events;
    using global::RabbitMQ.Client.Exceptions;
    using Microsoft.Extensions.DependencyInjection;
    using Microsoft.Extensions.Logging;
    using Polly;
    using Polly.Retry;

    public class EventBusRabbitMQ : IEventBus, IDisposable
    {
        const string BROKER_NAME = "coast_event_bus";

        private readonly IRabbitMQPersistentConnection _persistentConnection;
        private readonly ILogger<EventBusRabbitMQ> _logger;
        private readonly IEventBusSubscriptionsManager _subsManager;
        private readonly IServiceProvider _serviceProvider;
        private readonly int _retryCount;
        private readonly IProcessSagaEvent _processSagaEvent;
        private readonly IRepositoryFactory _repositoryFactory;
        private readonly string _callBackEventName;

        private IModel _consumerChannel;
        private string _queueName;

        public EventBusRabbitMQ(
            IRabbitMQPersistentConnection persistentConnection,
            ILogger<EventBusRabbitMQ> logger,
            IServiceProvider serviceProvider,
            IEventBusSubscriptionsManager subsManager,
            IProcessSagaEvent processSagaEvent,
            IRepositoryFactory repositoryFactory,
            CoastOptions coastOptions,
            string queueName = null,
            int retryCount = 5)
        {
            _serviceProvider = serviceProvider;
            _persistentConnection = persistentConnection ?? throw new ArgumentNullException(nameof(persistentConnection));
            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
            _subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager();
            _callBackEventName = coastOptions.DomainName + CoastConstant.CallBackEventSuffix;
            _queueName = string.IsNullOrWhiteSpace(queueName) ? coastOptions.DomainName + CoastConstant.QueueNameSuffix : queueName + CoastConstant.QueueNameSuffix;
            _consumerChannel = CreateConsumerChannel();
            _retryCount = retryCount;
            _subsManager.OnEventRemoved += SubsManager_OnEventRemoved;
            _processSagaEvent = processSagaEvent;
            _repositoryFactory = repositoryFactory;
        }