using System;

using System.Threading.Tasks;

using System.Threading;

using System.Collections.Generic;

using System.Collections.ObjectModel;

using System.Collections.Concurrent;

​

public class Program {

    public static void Main(string[] args){

        var eventStream = new EventStream();

        List<IDisposable> disposables = new();

        disposables.Add(eventStream);

        Console.WriteLine("Press enter to exit");

        Console.ReadLine();

        Console.WriteLine("Cleaning up");

        foreach(var d in disposables){

            d.Dispose();

        }

    }

}

​

public record StreamEvent(int Payload);

​

public class EventStream: IObservable<StreamEvent>, IDisposable {

    private int _nextSubscriptionId = 1;

    private IDictionary<int, EventStreamSubscription> _subscriptions = new ConcurrentDictionary<int, EventStreamSubscription>();

​

    private CancellationTokenSource _cancellationTokenSource;

    private int _payload = 1;

    public EventStream(){

        _cancellationTokenSource = new CancellationTokenSource();

        Console.WriteLine("Creating the event stream");

        var token = _cancellationTokenSource.Token;

        Task.Run(async () => {

            Console.WriteLine("reading from stream");

            await Task.Delay(TimeSpan.FromMilliseconds(500), token);

            token.ThrowIfCancellationRequested();

            // this method can be considered to be OnEventArrived

            var evt = new StreamEvent(_payload++);

            PublishEvent(evt);

        });

    }

    private void PublishEvent(StreamEvent evt){

        // will be called from OnEventArrived

        Console.WriteLine("Publishing event: " + evt.Payload);

        foreach(var subscription in _subscriptions.Values){

            try{

                subscription.Observer.OnNext(evt);

            }catch(Exception ex){

                // OnNext failed...

            }

        }

    }

    public IDisposable Subscribe(IObserver<StreamEvent> observer){

        var subscription = new EventStreamSubscription(_nextSubscriptionId++, observer, RemoveSubscription);

        _subscriptions.Add(subscription.SubscriptionId, subscription);

        return subscription;

    }

​

    private void RemoveSubscription(int subscriptionId){

        _subscriptions.Remove(subscriptionId);

    }

    public void Dispose(){

        Console.WriteLine("Disposing the event stream");

        _cancellationTokenSource.Cancel();

    }

}

​

public class EventStreamSubscription : IDisposable {

    private IObserver<StreamEvent> _observer;

    private int _subscriptionId;

    private Action<int> _disposeCallback;

    public EventStreamSubscription(int subscriptionId, IObserver<StreamEvent> observer, Action<int> disposeCallback){

        _subscriptionId = subscriptionId;

        _observer = observer;

        _disposeCallback = disposeCallback;

    }

    public int SubscriptionId { get => _subscriptionId; }

    public IObserver<StreamEvent> Observer { get => _observer; }

    public void Dispose(){

        Dispose(true);

    }

    protected void Dispose(bool disposing){

        if(disposing){

            if(_disposeCallback != null){

                _disposeCallback(_subscriptionId);

            }

        }

    }

    ~EventStreamSubscription(){

        Dispose(false);

    }

}