hermes-client/Bus/ServiceBusObservable.cs

using System.Reactive;
using Serilog;

namespace TwitchChatTTS.Bus
{
    public class ServiceBusObservable : ObservableBase<ServiceBusData>
    {
        private readonly string _topic;
        private readonly ServiceBusCentral _central;
        private readonly ILogger _logger;

        public ServiceBusObservable(string topic, ServiceBusCentral central, ILogger logger)
        {
            _topic = topic;
            _central = central;
            _logger = logger;
        }

        protected override IDisposable SubscribeCore(IObserver<ServiceBusData> observer)
        {
            _central.Add(_topic, observer);
            return new ServiceBusUnsubscriber(_topic, _central, observer);
        }

        public IDisposable Subscribe(Action<ServiceBusData> action) {
            return Subscribe(new ServiceBusObserver(action, _logger));
        }

        private sealed class ServiceBusUnsubscriber : IDisposable
        {
            private readonly string _topic;
            private readonly ServiceBusCentral _central;
            private readonly IObserver<ServiceBusData> _receiver;

            public ServiceBusUnsubscriber(string topic, ServiceBusCentral central, IObserver<ServiceBusData> receiver)
            {
                _topic = topic;
                _central = central;
                _receiver = receiver;
            }

            public void Dispose()
            {
                _central.RemoveObserver(_topic, _receiver);
            }
        }
    }
}