Added Finally observable operator.

Replaced `DoOnDispose` with `Finally` in `UpdateDistributor` to fix subscriber tracking.

Author: HedgehogNSK

src/RxTelegram.Bot/Api/UpdateDistributor.cs

@@ -111,7 +111,7 @@ public IObservable<T> Selector<T>(UpdateType? type, Func<Update, T> propertySele
     return _tracker.Switch().Select(propertySelector)
       .Where(x => x != null)
       .DoOnSubscribe(() => AddListener(type))
-      .DoOnDispose(() => RemoveListener(type));
+      .Finally(() => RemoveListener(type));
   }
 
   public void Set(IObservable<Update> tracker)

src/RxTelegram.Bot/Utils/Rx/DoOnDisposeObservable.cs

@@ -7,10 +7,10 @@ internal class DoOnDisposeObservable<T> : IObservable<T>
   private readonly IObservable<T> _source;
   private readonly Action _onDispose;
 
-  public DoOnDisposeObservable(IObservable<T> source, Action onTerminate)
+  public DoOnDisposeObservable(IObservable<T> source, Action onDispose)
   {
     this._source = source ?? throw new ArgumentNullException(nameof(source));
-    this._onDispose = onTerminate ?? throw new ArgumentNullException(nameof(onTerminate));
+    this._onDispose = onDispose ?? throw new ArgumentNullException(nameof(onDispose));
   }
 
   public IDisposable Subscribe(IObserver<T> observer)

src/RxTelegram.Bot/Utils/Rx/FinallyObservable.cs

@@ -0,0 +1,77 @@
+using System;
+using System.Threading;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+internal class FinallyObservable<T> : IObservable<T>
+{
+  private readonly IObservable<T> _source;
+  private readonly Action _onTerminate;
+
+  public FinallyObservable(IObservable<T> source, Action onTerminate)
+  {
+    _source = source ?? throw new ArgumentNullException(nameof(source));
+    _onTerminate = onTerminate ?? throw new ArgumentNullException(nameof(onTerminate));
+  }
+
+  public IDisposable Subscribe(IObserver<T> observer)
+  {
+    if (observer == null)
+      throw new ArgumentNullException(nameof(observer));
+
+    var finallyObserver = new FinallyObserver(observer, _onTerminate);
+    var subscription = _source.Subscribe(finallyObserver);
+
+    return new DisposableAction(() => finallyObserver.Dispose(subscription));
+  }
+
+  private class FinallyObserver : IObserver<T>
+  {
+    private readonly IObserver<T> _observer;
+    private readonly Action _onTerminate;
+    private int _terminated;
+
+    public FinallyObserver(IObserver<T> observer, Action onTerminate)
+    {
+      _observer = observer;
+      _onTerminate = onTerminate;
+    }
+    public void Dispose(IDisposable subscription)
+    {
+      subscription.Dispose();
+      Terminate();
+    }
+
+    public void OnCompleted()
+    {
+      _observer.OnCompleted();
+      Terminate();
+    }
+
+    public void OnError(Exception error)
+    {
+      try
+      {
+        _observer.OnError(error);
+      }
+      catch (Exception ex)
+      {
+        ExceptionHelpers.ThrowIfFatal(ex);
+      }
+      finally
+      {
+        Terminate();
+      }
+    }
+
+    public void OnNext(T value) => _observer.OnNext(value);
+
+    private void Terminate()
+    {
+      if (Interlocked.Exchange(ref _terminated, 1) == 0)
+      {
+        _onTerminate();
+      }
+    }
+  }
+}

src/RxTelegram.Bot/Utils/Rx/ObservableOperators.cs

@@ -8,6 +8,8 @@ static public IObservable<T> DoOnDispose<T>(this IObservable<T> source, Action o
     => new DoOnDisposeObservable<T>(source, onDispose);
   static public IObservable<T> DoOnSubscribe<T>(this IObservable<T> source, Action onSubscribe)
     => new DoOnSubscribeObservable<T>(source, onSubscribe);
+  static public IObservable<T> Finally<T>(this IObservable<T> source, Action onTerminate)
+    => new FinallyObservable<T>(source,onTerminate);
   static public IObservable<KOut> Select<TIn, KOut>(this IObservable<TIn> source, Func<TIn, KOut> selector)
     => new SelectObservable<TIn, KOut>(source, selector);
   static public IObservable<T> Switch<T>(this IObservable<IObservable<T>> source)