EasyCachingAbstractBus.cs

namespace EasyCaching.Core
{
    using EasyCaching.Core.Bus;
    using EasyCaching.Core.Diagnostics;
    using System;
    using System.Diagnostics;
    using System.Threading;
    using System.Threading.Tasks;

    public abstract class EasyCachingAbstractBus : IEasyCachingBus
    {
        protected static readonly DiagnosticListener s_diagnosticListener =
                    new DiagnosticListener(EasyCachingDiagnosticListenerExtensions.DiagnosticListenerName);

        public abstract void BasePublish(string topic, EasyCachingMessage message);
        public abstract Task BasePublishAsync(string topic, EasyCachingMessage message, CancellationToken cancellationToken = default(CancellationToken));
        public abstract void BaseSubscribe(string topic, Action<EasyCachingMessage> action);
        public abstract Task BaseSubscribeAsync(string topic, Action<EasyCachingMessage> action, CancellationToken cancellationToken = default(CancellationToken));

        protected Action<EasyCachingMessage> _handler;

        protected Action _reconnectHandler;
        
        protected string BusName { get; set; }

        public string Name => this.BusName;

        public void Publish(string topic, EasyCachingMessage message)
        {
            var operationId = s_diagnosticListener.WritePublishMessageBefore(new BeforePublishMessageRequestEventData(topic, message));
            try
            {
                BasePublish(topic, message);
            }
            catch (Exception ex)
            {
                s_diagnosticListener.WritePublishMessageError(operationId, ex);
                throw;
            }
            s_diagnosticListener.WritePublishMessageAfter(operationId);
        }

        public async Task PublishAsync(string topic, EasyCachingMessage message, CancellationToken cancellationToken = default(CancellationToken))
        {
            var operationId = s_diagnosticListener.WritePublishMessageBefore(new BeforePublishMessageRequestEventData(topic, message));
            try
            {
                await BasePublishAsync(topic, message, cancellationToken);
            }
            catch (Exception ex)
            {
                s_diagnosticListener.WritePublishMessageError(operationId, ex);
                throw;
            }
            s_diagnosticListener.WritePublishMessageAfter(operationId);
        }

        public void Subscribe(string topic, Action<EasyCachingMessage> action, Action reconnectAction)
        {
            _handler = action;
            _reconnectHandler = reconnectAction;
            BaseSubscribe(topic, action);
        }

        public async Task SubscribeAsync(string topic, Action<EasyCachingMessage> action, Action reconnectAction, CancellationToken cancellationToken = default(CancellationToken))
        {
            _handler = action;
            _reconnectHandler = reconnectAction;
            await BaseSubscribeAsync(topic, action);
        }

        public virtual void BaseOnMessage(EasyCachingMessage message)
        {
            var operationId = s_diagnosticListener.WriteSubscribeMessageBefore(new BeforeSubscribeMessageRequestEventData(message));
            try
            {
                _handler?.Invoke(message);
            }
            catch (Exception ex)
            {
                s_diagnosticListener.WritePublishMessageError(operationId, ex);
                throw;
            }
            s_diagnosticListener.WritePublishMessageAfter(operationId);
        }

        public virtual void BaseOnReconnect()
        {
            _reconnectHandler?.Invoke();
        }
    }
}