diff --git a/src/RxTelegram.Bot/Api/ITrackerSetup.cs b/src/RxTelegram.Bot/Api/ITrackerSetup.cs
new file mode 100644
index 0000000..0943df7
--- /dev/null
+++ b/src/RxTelegram.Bot/Api/ITrackerSetup.cs
@@ -0,0 +1,9 @@
+using RxTelegram.Bot.Interface.BaseTypes.Enums;
+using System.Collections.Generic;
+
+namespace RxTelegram.Bot.Api;
+
+public interface ITrackerSetup
+{
+  void Set(IEnumerable<UpdateType> types);
+}
diff --git a/src/RxTelegram.Bot/Api/IUpdateManager.cs b/src/RxTelegram.Bot/Api/IUpdateManager.cs
index 56a611d..7020685 100644
--- a/src/RxTelegram.Bot/Api/IUpdateManager.cs
+++ b/src/RxTelegram.Bot/Api/IUpdateManager.cs
@@ -12,6 +12,12 @@ namespace RxTelegram.Bot.Api;
 
 public interface IUpdateManager
 {
+    /// <summary>
+    /// Allows to set custom updates tracker
+    /// </summary>
+    /// <param name="tracker"></param>
+    void Set(IObservable<Update> tracker);
+    
     /// <summary>
     /// Updates of all Types.
     /// </summary>
diff --git a/src/RxTelegram.Bot/Api/LongpollingUpdateTracker.cs b/src/RxTelegram.Bot/Api/LongpollingUpdateTracker.cs
new file mode 100644
index 0000000..844df0a
--- /dev/null
+++ b/src/RxTelegram.Bot/Api/LongpollingUpdateTracker.cs
@@ -0,0 +1,189 @@
+using RxTelegram.Bot.Interface.BaseTypes.Enums;
+using RxTelegram.Bot.Interface.Setup;
+using RxTelegram.Bot.Utils.Rx;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace RxTelegram.Bot.Api;
+public class LongPollingUpdateTracker(ITelegramBot telegramBot)
+  : IObservable<Update>, ITrackerSetup
+{
+    private readonly ITelegramBot _telegramBot = telegramBot;
+    private const int NotRunning = 0;
+    private const int Running = 1;
+    private int _isRunning = NotRunning;
+    private CancellationTokenSource _cancellationTokenSource;
+    private UpdateType[] _trackedUpdateTypes = [];
+    readonly List<IObserver<Update>> _observers = new List<IObserver<Update>>();
+
+    public void Set(IEnumerable<UpdateType> types)
+    {
+        if (types == null)
+        {
+            if (_trackedUpdateTypes != null)
+            {
+                _trackedUpdateTypes = null;
+                _cancellationTokenSource?.Cancel();
+            }
+            return;
+        }
+
+        var updateTypes = types as UpdateType[] ?? types.ToArray();
+        if (_trackedUpdateTypes == null || !updateTypes.SequenceEqual(_trackedUpdateTypes))
+        {
+            _trackedUpdateTypes = updateTypes;
+            _cancellationTokenSource?.Cancel();
+        }
+    }
+
+    internal async Task RunUpdateSafe()
+    {
+        try
+        {
+            _cancellationTokenSource = new CancellationTokenSource();
+            await RunUpdate();
+        }
+        catch (Exception ex)
+        {
+            // ignored
+            ExceptionHelpers.ThrowIfFatal(ex);
+        }
+        finally
+        {
+            _cancellationTokenSource.Dispose();
+            _cancellationTokenSource = null;
+
+            if (!_observers.Any())
+            {
+                Volatile.Write(ref _isRunning, NotRunning);
+            }
+            else
+            {
+                RunUpdateTask();
+            }
+        }
+        void RunUpdateTask() => Task.Run(RunUpdateSafe);
+    }
+
+    private int? _offset; // Offset must be preserved for all errors except TaskCanceledException.
+                        // Using a local variable may cause duplicates if an exception occurs.
+    internal async Task RunUpdate()
+    {
+
+        while (_observers.Count != 0)
+        {
+            try
+            {
+                // if the token already canceled before the first request reset token
+                if (_cancellationTokenSource.IsCancellationRequested)
+                {
+                    _cancellationTokenSource = new CancellationTokenSource();
+                }
+
+                var getUpdate = new GetUpdate
+                                {
+                                    Offset = _offset,
+                                    Timeout = 60,
+
+                                    // if there is a null value in the list, it means that all updates are allowed
+                                    AllowedUpdates = _trackedUpdateTypes
+                                };
+
+                var result = await _telegramBot.GetUpdate(getUpdate, _cancellationTokenSource.Token);
+                if (!result.Any())
+                {
+                    await Task.Delay(1000);
+                    continue;
+                }
+
+                NotifyObservers(result);
+
+                var lastId = result.Length - 1;
+                _offset = result[lastId].UpdateId + 1;
+            }
+            catch (TaskCanceledException)
+            {
+                _offset = null;
+                _cancellationTokenSource = new CancellationTokenSource();
+            }
+            catch (Exception exception)
+            {
+                OnException(exception);
+                throw;
+            }
+        }
+    }
+    internal void NotifyObservers(Update[] updates)
+    {
+        for (var uid = 0; uid != updates.Length; ++uid)
+        {
+            for (var oid = 0; oid != _observers.Count; ++oid)
+            {
+                _observers[oid].OnNext(updates[uid]);
+            }
+        }
+    }
+    internal void OnException(Exception exception)
+    {
+        IObserver<Update>[] current;
+        lock (_observers)
+        {
+            current = _observers.ToArray(); // Caching current observers to prevent
+                                            // notifying those who subscribed after an error occurred.
+            _observers.Clear();
+        }
+
+        for (var oid = 0; oid != current.Length; ++oid)
+        {
+            try
+            {
+                current[oid].OnError(exception);
+            }
+            catch (Exception ex)
+            {
+                // Ignore exceptions from observers without an error handler,
+                // as it would break the process and propagate the exception to the outer scope.
+                ExceptionHelpers.ThrowIfFatal(ex);
+            }
+        }
+    }
+    internal void Remove(IObserver<Update> observer)
+    {
+        if (!_observers.Contains(observer))
+        {
+            return;
+        }
+
+        lock (_observers)
+        {
+            _observers.Remove(observer);
+        }
+
+        if (!_observers.Any() && Volatile.Read(ref _isRunning) == Running)
+        {
+            _cancellationTokenSource?.Cancel();
+        }
+    }
+    public IDisposable Subscribe(IObserver<Update> observer)
+    {
+        if (observer == null)
+        {
+            throw new ArgumentNullException(nameof(observer));
+        }
+
+        lock (_observers)
+        {
+            _observers.Add(observer);
+        }
+
+        if (Interlocked.Exchange(ref _isRunning, Running) == NotRunning)
+        {
+            Task.Run(RunUpdateSafe);
+        }
+
+        return new DisposableAction(() => Remove(observer));
+    }
+}
diff --git a/src/RxTelegram.Bot/Api/UpdateDistributor.cs b/src/RxTelegram.Bot/Api/UpdateDistributor.cs
new file mode 100644
index 0000000..a38874e
--- /dev/null
+++ b/src/RxTelegram.Bot/Api/UpdateDistributor.cs
@@ -0,0 +1,170 @@
+using RxTelegram.Bot.Interface.BaseTypes;
+using RxTelegram.Bot.Interface.BaseTypes.Enums;
+using RxTelegram.Bot.Interface.InlineMode;
+using RxTelegram.Bot.Interface.Payments;
+using RxTelegram.Bot.Interface.Setup;
+using RxTelegram.Bot.Utils.Rx;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+#if NETSTANDARD2_1
+
+using RxTelegram.Bot.Utils;
+
+#endif
+
+namespace RxTelegram.Bot.Api;
+
+public sealed class UpdateDistributor : IUpdateManager, IDisposable
+{
+
+    #region Observable Update properties
+    private readonly Dictionary<UpdateType, UpdateTypeInfo> _updateInfos;
+    private readonly UpdateTypeInfo _updateInfo;
+    private readonly IEnumerable<UpdateType> _trackedTypes;
+    private readonly ReactiveProperty<IObservable<Update>> _tracker;
+    private bool _isDisposed;
+    private readonly object _lock;
+    public IObservable<CallbackQuery> CallbackQuery => Selector(UpdateType.CallbackQuery, update => update.CallbackQuery);
+    public IObservable<Message> ChannelPost => Selector(UpdateType.ChannelPost, update => update.ChannelPost);
+    public IObservable<ChatBoostUpdated> ChatBoost => Selector(UpdateType.ChatBoost, update => update.ChatBoost);
+    public IObservable<ChatJoinRequest> ChatJoinRequest => Selector(UpdateType.ChatJoinRequest, update => update.ChatJoinRequest);
+    public IObservable<ChatMemberUpdated> ChatMember => Selector(UpdateType.ChatMember, update => update.ChatMember);
+    public IObservable<ChosenInlineResult> ChosenInlineResult => Selector(UpdateType.ChosenInlineResult, update => update.ChosenInlineResult);
+    public IObservable<Message> EditedChannelPost => Selector(UpdateType.EditedChannelPost, update => update.EditedChannelPost);
+    public IObservable<Message> EditedMessage => Selector(UpdateType.EditedMessage, update => update.EditedMessage);
+    public IObservable<InlineQuery> InlineQuery => Selector(UpdateType.InlineQuery, update => update.InlineQuery);
+    public IObservable<Message> Message => Selector(UpdateType.Message, update => update.Message);
+    public IObservable<ChatMemberUpdated> MyChatMember => Selector(UpdateType.MyChatMember, update => update.MyChatMember);
+    public IObservable<Poll> Poll => Selector(UpdateType.Poll, update => update.Poll);
+    public IObservable<PollAnswer> PollAnswer => Selector(UpdateType.PollAnswer, update => update.PollAnswer);
+    public IObservable<PreCheckoutQuery> PreCheckoutQuery => Selector(UpdateType.PreCheckoutQuery, update => update.PreCheckoutQuery);
+    public IObservable<ChatBoostRemoved> RemovedChatBoost => Selector(UpdateType.RemovedChatBoost, update => update.RemovedChatBoost);
+    public IObservable<ShippingQuery> ShippingQuery => Selector(UpdateType.ShippingQuery, update => update.ShippingQuery);
+    public IObservable<Update> Update => Selector(null, update => update);
+    #endregion
+
+#if NETSTANDARD2_1
+  public IAsyncEnumerable<CallbackQuery> CallbackQueryEnumerable() => CallbackQuery.ToAsyncEnumerable();
+  public IAsyncEnumerable<Message> ChannelPostEnumerable() => ChannelPost.ToAsyncEnumerable();
+  public IAsyncEnumerable<ChatBoostUpdated> ChatBoostEnumerable() => ChatBoost.ToAsyncEnumerable();
+  public IAsyncEnumerable<ChatJoinRequest> ChatJoinRequestEnumerable() => ChatJoinRequest.ToAsyncEnumerable();
+  public IAsyncEnumerable<ChatMemberUpdated> ChatMemberEnumerable() => ChatMember.ToAsyncEnumerable();
+  public IAsyncEnumerable<ChosenInlineResult> ChosenInlineResultEnumerable() => ChosenInlineResult.ToAsyncEnumerable();
+  public IAsyncEnumerable<Message> EditedChannelPostEnumerable() => EditedChannelPost.ToAsyncEnumerable();
+  public IAsyncEnumerable<Message> EditedMessageEnumerable() => EditedMessage.ToAsyncEnumerable();
+  public IAsyncEnumerable<InlineQuery> InlineQueryEnumerable() => InlineQuery.ToAsyncEnumerable();
+  public IAsyncEnumerable<Message> MessageEnumerable() => Message.ToAsyncEnumerable();
+  public IAsyncEnumerable<ChatMemberUpdated> MyChatMemberEnumerable() => MyChatMember.ToAsyncEnumerable();
+  public IAsyncEnumerable<ShippingQuery> ShippingQueryEnumerable() => ShippingQuery.ToAsyncEnumerable();
+  public IAsyncEnumerable<PollAnswer> PollAnswerEnumerable() => PollAnswer.ToAsyncEnumerable();
+  public IAsyncEnumerable<Poll> PollEnumerable() => Poll.ToAsyncEnumerable();
+  public IAsyncEnumerable<PreCheckoutQuery> PreCheckoutQueryEnumerable() => PreCheckoutQuery.ToAsyncEnumerable();
+  public IAsyncEnumerable<ChatBoostRemoved> RemovedChatBoostEnumerable() => RemovedChatBoost.ToAsyncEnumerable();
+  public IAsyncEnumerable<Update> UpdateEnumerable() => Update.ToAsyncEnumerable();
+
+#endif
+    public UpdateDistributor(IObservable<Update> updateTracker)
+    {
+        _lock = new();
+        _updateInfos = Enum.GetValues(typeof(UpdateType))
+          .Cast<UpdateType>()
+          .ToDictionary(x => x, _ => new UpdateTypeInfo());
+
+        _updateInfo = new UpdateTypeInfo();
+
+        _trackedTypes = _updateInfos.Where(x => x.Value.Listeners != 0)
+           .Select(x => x.Key);
+
+        _tracker = new ReactiveProperty<IObservable<Update>>(updateTracker);
+        Set(updateTracker);
+    }
+    private void AddListener(UpdateType? type)
+    {
+        lock (_lock)
+        {
+            var info = GetInfo(type);
+            ++info.Listeners;
+
+            if (info.Listeners != 1)
+            {
+                return;
+            }
+
+            UpdateTrackerTypes();
+        }
+    }
+    private UpdateTypeInfo GetInfo(UpdateType? type)
+      => type == null ? _updateInfo : _updateInfos[(UpdateType)type];
+
+    private void RemoveListener(UpdateType? type)
+    {
+        lock (_lock)
+        {
+            var info = GetInfo(type);
+            --info.Listeners;
+            if (info.Listeners != 0)
+            {
+                return;
+            }
+
+            UpdateTrackerTypes();
+        }
+    }
+
+    private IObservable<T> Selector<T>(UpdateType? type, Func<Update, T> propertySelector)
+    where T : class
+    {
+        return _tracker.Switch().Select(propertySelector)
+          .Where(x => x != null)
+          .DoOnSubscribe(() => AddListener(type))
+          .Finally(() => RemoveListener(type));
+    }
+
+    public void Set(IObservable<Update> tracker)
+    {
+        // Configure the current tracker to listen for all types of updates
+        // before switching to a new one
+        (_tracker.Current as ITrackerSetup)?.Set(null);
+
+        _tracker.OnNext(tracker);
+        UpdateTrackerTypes();
+    }
+    private void UpdateTrackerTypes()
+    {
+        if (_tracker.Current is not ITrackerSetup setup)
+        {
+            return;
+        }
+
+        lock (_lock)
+        {
+            setup.Set(_updateInfo.Listeners != 0 || !_trackedTypes.Any() ?
+                          null : _trackedTypes);
+        }
+    }
+
+    public void Dispose() => Dispose(true);
+    private void Dispose(bool explicitDisposing)
+    {
+        if (_isDisposed)
+        {
+            return;
+        }
+
+        if (explicitDisposing)
+        {
+            _tracker.Dispose();
+        }
+
+        _isDisposed = true;
+    }
+
+    ~UpdateDistributor() => Dispose(false);
+
+    private sealed class UpdateTypeInfo
+    {
+        public int Listeners { get; set; }
+    }
+}
diff --git a/src/RxTelegram.Bot/Api/UpdateManager.cs b/src/RxTelegram.Bot/Api/UpdateManager.cs
index cdc386d..51a6740 100644
--- a/src/RxTelegram.Bot/Api/UpdateManager.cs
+++ b/src/RxTelegram.Bot/Api/UpdateManager.cs
@@ -16,6 +16,8 @@ namespace RxTelegram.Bot.Api;
 
 public class UpdateManager : IUpdateManager
 {
+    public void Set(IObservable<Update> tracker) => throw new NotImplementedException();
+
     public IObservable<Update> Update => _update;
 
     public IObservable<Message> Message => _message;
diff --git a/src/RxTelegram.Bot/TelegramBot.Builder.cs b/src/RxTelegram.Bot/TelegramBot.Builder.cs
new file mode 100644
index 0000000..132ef76
--- /dev/null
+++ b/src/RxTelegram.Bot/TelegramBot.Builder.cs
@@ -0,0 +1,42 @@
+using System;
+using RxTelegram.Bot.Api;
+using RxTelegram.Bot.Interface.Setup;
+
+namespace RxTelegram.Bot;
+
+public partial class TelegramBot
+{
+    public class Builder
+    {
+        private string _token;
+        private IObservable<Update> _tracker;
+        private IUpdateManager _updateManager;
+        public Builder() { }
+        public Builder(string token) : this() { _token = token; }
+
+        public Builder SetToken(string token)
+        {
+            _token = token;
+            return this;
+        }
+        public Builder SetTracker(IObservable<Update> tracker)
+        {
+            _tracker = tracker;
+            return this;
+        }
+        public Builder SetManager(IUpdateManager updateManager)
+        {
+            _updateManager = updateManager;
+            return this;
+        }
+        public TelegramBot Build()
+        {
+            var bot = new TelegramBot(_token);
+            _tracker ??= new LongPollingUpdateTracker(bot);
+            bot.Updates = _updateManager ?? new UpdateDistributor(_tracker);
+            bot.Updates.Set(_tracker);
+
+            return bot;
+        }
+    }
+}
diff --git a/src/RxTelegram.Bot/TelegramBot.cs b/src/RxTelegram.Bot/TelegramBot.cs
index 7a8c1e5..e472e7f 100644
--- a/src/RxTelegram.Bot/TelegramBot.cs
+++ b/src/RxTelegram.Bot/TelegramBot.cs
@@ -30,15 +30,19 @@
 
 namespace RxTelegram.Bot;
 
-public class TelegramBot : BaseTelegramBot, ITelegramBot
+public partial class TelegramBot : BaseTelegramBot, ITelegramBot
 {
     public TelegramBot(string token) : this(new BotInfo(token))
     {
     }
 
-    public TelegramBot(BotInfo botInfo) : base(botInfo) => Updates = new UpdateManager(this);
+    public TelegramBot(BotInfo botInfo) : base(botInfo)
+    {
+        var tracker = new LongPollingUpdateTracker(this);
+        Updates = new UpdateDistributor(tracker);
+    }
 
-    public IUpdateManager Updates { get; }
+    public IUpdateManager Updates { get; private set; }
 
     /// <summary>
     ///     Use this method to get basic info about a file and prepare it for downloading.
diff --git a/src/RxTelegram.Bot/Utils/Rx/DisposableAction.cs b/src/RxTelegram.Bot/Utils/Rx/DisposableAction.cs
new file mode 100644
index 0000000..b73b912
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/DisposableAction.cs
@@ -0,0 +1,10 @@
+using System;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+public sealed class DisposableAction(Action action) : IDisposable
+{
+    public static DisposableAction Empty { get; } = new DisposableAction(() => { });
+    private readonly Action _action = action ?? throw new ArgumentNullException(nameof(action));
+    public void Dispose() => _action();
+}
diff --git a/src/RxTelegram.Bot/Utils/Rx/DoOnDisposeObservable.cs b/src/RxTelegram.Bot/Utils/Rx/DoOnDisposeObservable.cs
new file mode 100644
index 0000000..5a39b85
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/DoOnDisposeObservable.cs
@@ -0,0 +1,20 @@
+using System;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+internal class DoOnDisposeObservable<T>(IObservable<T> source, Action onDispose) : IObservable<T>
+{
+    private readonly IObservable<T> _source = source ?? throw new ArgumentNullException(nameof(source));
+    private readonly Action _onDispose = onDispose ?? throw new ArgumentNullException(nameof(onDispose));
+
+    public IDisposable Subscribe(IObserver<T> observer)
+    {
+
+        var subscription = _source.Subscribe(observer);
+        return new DisposableAction(() =>
+          {
+              subscription.Dispose();
+              _onDispose();
+          });
+    }
+}
diff --git a/src/RxTelegram.Bot/Utils/Rx/DoOnSubscribeObservable.cs b/src/RxTelegram.Bot/Utils/Rx/DoOnSubscribeObservable.cs
new file mode 100644
index 0000000..3570a56
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/DoOnSubscribeObservable.cs
@@ -0,0 +1,17 @@
+using System;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+internal class DoOnSubscribeObservable<T>(IObservable<T> source, Action onSubscribe) : IObservable<T>
+{
+    private readonly IObservable<T> _source = source ?? throw new ArgumentNullException(nameof(source));
+    private readonly Action _onSubscribe = onSubscribe ?? throw new ArgumentNullException(nameof(onSubscribe));
+
+    public IDisposable Subscribe(IObserver<T> observer)
+    {
+
+        var subscription = _source.Subscribe(observer);
+        _onSubscribe();
+        return subscription;
+    }
+}
diff --git a/src/RxTelegram.Bot/Utils/Rx/ExceptionHelpers.cs b/src/RxTelegram.Bot/Utils/Rx/ExceptionHelpers.cs
new file mode 100644
index 0000000..ff3a6a9
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/ExceptionHelpers.cs
@@ -0,0 +1,25 @@
+using System;
+using System.Threading;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+internal static class ExceptionHelpers
+{
+    internal static void ThrowIfFatal(Exception ex)
+    {
+        if (ex is OutOfMemoryException || ex is StackOverflowException
+          || ex is ThreadAbortException)
+        {
+            throw ex;
+        }
+    }
+
+    internal static void ThrowIfDisposed(this IDisposable disposable, Func<bool> isDisposed)
+    {
+        if (isDisposed())
+        {
+            throw new ObjectDisposedException(disposable.GetType().FullName);
+        }
+
+    }
+}
diff --git a/src/RxTelegram.Bot/Utils/Rx/FinallyObservable.cs b/src/RxTelegram.Bot/Utils/Rx/FinallyObservable.cs
new file mode 100644
index 0000000..b59ae2a
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/FinallyObservable.cs
@@ -0,0 +1,66 @@
+using System;
+using System.Threading;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+internal class FinallyObservable<T>(IObservable<T> source, Action onTerminate) : IObservable<T>
+{
+    private readonly IObservable<T> _source = source ?? throw new ArgumentNullException(nameof(source));
+    private readonly Action _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.DisposeSubscription(subscription));
+    }
+
+    private sealed class FinallyObserver(IObserver<T> observer, Action onTerminate) : IObserver<T>
+    {
+        private int _terminated;
+
+        public void DisposeSubscription(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();
+            }
+        }
+    }
+}
diff --git a/src/RxTelegram.Bot/Utils/Rx/Interfaces/ISubject.cs b/src/RxTelegram.Bot/Utils/Rx/Interfaces/ISubject.cs
new file mode 100644
index 0000000..24db1d7
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/Interfaces/ISubject.cs
@@ -0,0 +1,6 @@
+using System;
+
+namespace RxTelegram.Bot.Utils.Rx.Interfaces;
+
+public interface ISubject<in Tin, out Tout> : IObserver<Tin>, IObservable<Tout> { }
+public interface ISubject<T> : ISubject<T, T> { }
\ No newline at end of file
diff --git a/src/RxTelegram.Bot/Utils/Rx/ObservableOperators.cs b/src/RxTelegram.Bot/Utils/Rx/ObservableOperators.cs
new file mode 100644
index 0000000..adc491a
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/ObservableOperators.cs
@@ -0,0 +1,19 @@
+using System;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+public static class ObservableOperators
+{
+    public static IObservable<T> DoOnDispose<T>(this IObservable<T> source, Action onDispose)
+      => new DoOnDisposeObservable<T>(source, onDispose);
+    public static IObservable<T> DoOnSubscribe<T>(this IObservable<T> source, Action onSubscribe)
+      => new DoOnSubscribeObservable<T>(source, onSubscribe);
+    public static IObservable<T> Finally<T>(this IObservable<T> source, Action onTerminate)
+      => new FinallyObservable<T>(source, onTerminate);
+    public static IObservable<TOut> Select<TIn, TOut>(this IObservable<TIn> source, Func<TIn, TOut> selector)
+      => new SelectObservable<TIn, TOut>(source, selector);
+    public static IObservable<T> Switch<T>(this IObservable<IObservable<T>> source)
+      => new SwitchObservable<T>(source);
+    public static IObservable<T> Where<T>(this IObservable<T> source, Func<T, bool> predicate)
+      => new WhereObservable<T>(source, predicate);
+}
diff --git a/src/RxTelegram.Bot/Utils/Rx/ReactiveProperty.cs b/src/RxTelegram.Bot/Utils/Rx/ReactiveProperty.cs
new file mode 100644
index 0000000..155cad2
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/ReactiveProperty.cs
@@ -0,0 +1,139 @@
+using RxTelegram.Bot.Utils.Rx.Interfaces;
+using System;
+using System.Collections.Generic;
+using System.Threading;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+public class ReactiveProperty<T>() : ISubject<T>, IDisposable
+{
+    private static readonly List<IObserver<T>> Terminated = [];
+    private readonly object _lock = new object();
+    private Exception _error;
+    private T _current;
+    public T Current
+    {
+        get => _current;
+        protected set { _current = value; HasValue = true; }
+    }
+    public bool HasValue { get; private set; }
+    private List<IObserver<T>> _observers = new();
+    public bool IsDisposed { get; private set; }
+    public bool IsError => _error != null;
+    public Exception Error => _error;
+
+    public ReactiveProperty(T initValue) : this()
+    {
+        Current = initValue;
+    }
+
+    public void OnCompleted()
+    {
+        lock (_lock)
+        {
+            ThrowIfDisposed();
+            foreach (var observer in _observers)
+            {
+                observer.OnCompleted();
+            }
+
+            _observers = Terminated;
+        }
+    }
+    public void OnError(Exception error)
+    {
+        lock (_lock)
+        {
+            ThrowIfDisposed();
+            _error = error;
+            Current = default;
+            foreach (var observer in _observers)
+            {
+                try
+                {
+                    observer.OnError(error);
+                }
+                catch (Exception ex)
+                {
+                    //ignore
+                    // if observer doesn't have exception handler block
+                    // it will lead to breaking process and throwing exception to outer scope
+                    ExceptionHelpers.ThrowIfFatal(ex);
+                }
+            }
+
+            _observers = Terminated;
+        }
+    }
+    public void OnNext(T value)
+    {
+        lock (_lock)
+        {
+            ThrowIfDisposed();
+            Current = value;
+
+            foreach (var observer in _observers)
+            {
+                observer.OnNext(value);
+            }
+        }
+    }
+    public IDisposable Subscribe(IObserver<T> observer)
+    {
+        lock (_lock)
+        {
+            ThrowIfDisposed();
+            if (_observers == Terminated)
+            {
+                observer.OnCompleted();
+                return DisposableAction.Empty;
+            }
+
+            _observers.Add(observer);
+            if (HasValue)
+            {
+                observer.OnNext(Current);
+            }
+
+            return new DisposableAction(() =>
+                {
+                    lock (_lock)
+                    {
+                        _observers.Remove(observer);
+                    }
+                });
+        }
+    }
+
+    protected void ThrowIfDisposed()
+    {
+        if (IsDisposed)
+        {
+            throw new ObjectDisposedException(nameof(ReactiveProperty<T>));
+        }
+    }
+
+    ~ReactiveProperty() => Dispose(false);
+
+    public void Dispose()
+    {
+        Dispose(true);
+        GC.SuppressFinalize(this);
+    }
+
+    protected virtual void Dispose(bool explicitDisposing)
+    {
+        if (IsDisposed)
+        {
+            return;
+        }
+
+        if (explicitDisposing)
+        {
+            OnCompleted();
+            Interlocked.Exchange(ref _observers, Terminated);
+        }
+
+        IsDisposed = true;
+    }
+}
diff --git a/src/RxTelegram.Bot/Utils/Rx/SelectObservable.cs b/src/RxTelegram.Bot/Utils/Rx/SelectObservable.cs
new file mode 100644
index 0000000..c747e2e
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/SelectObservable.cs
@@ -0,0 +1,39 @@
+using System;
+
+namespace RxTelegram.Bot.Utils.Rx;
+internal class SelectObservable<TIn, TOut>(IObservable<TIn> source, Func<TIn, TOut> selector) : IObservable<TOut>
+{
+    private readonly IObservable<TIn> _source = source ?? throw new ArgumentNullException(nameof(source));
+    private readonly Func<TIn, TOut> _selector = selector ?? throw new ArgumentNullException(nameof(selector));
+
+    public IDisposable Subscribe(IObserver<TOut> observer)
+    {
+        if (observer == null)
+        {
+            throw new ArgumentNullException(nameof(observer));
+        }
+
+        var selectObserver = new SelectObserver(observer, _selector);
+        return _source.Subscribe(selectObserver);
+    }
+    private sealed class SelectObserver(IObserver<TOut> observer, Func<TIn, TOut> selector) : IObserver<TIn>
+    {
+        public void OnCompleted() => observer.OnCompleted();
+        public void OnError(Exception error) => observer.OnError(error);
+        public void OnNext(TIn value)
+        {
+            TOut result;
+            try
+            {
+                result = selector(value);
+            }
+            catch (Exception ex)
+            {
+                observer.OnError(ex);
+                return;
+            }
+
+            observer.OnNext(result);
+        }
+    }
+}
diff --git a/src/RxTelegram.Bot/Utils/Rx/SwitchObservable.cs b/src/RxTelegram.Bot/Utils/Rx/SwitchObservable.cs
new file mode 100644
index 0000000..5a7b4e7
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/SwitchObservable.cs
@@ -0,0 +1,99 @@
+using System;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+internal class SwitchObservable<T>(IObservable<IObservable<T>> source) : IObservable<T>
+{
+    private readonly IObservable<IObservable<T>> _source = source ?? throw new ArgumentNullException(nameof(source));
+
+    public IDisposable Subscribe(IObserver<T> observer)
+    {
+        var switchObserver = new SwitchObserver(observer);
+        var stream = _source.Subscribe(switchObserver);
+        return new DisposableAction(() =>
+        {
+            stream.Dispose();
+            switchObserver.Dispose();
+        });
+    }
+
+    private sealed class SwitchObserver(IObserver<T> observer) : IObserver<IObservable<T>>, IDisposable
+    {
+        private readonly object _lock = new();
+        private IObserver<T> _observer = observer ?? throw new ArgumentNullException(nameof(observer));
+        private IDisposable _subscription;
+        private bool _isDisposed;
+
+        public void OnCompleted()
+        {
+            lock (_lock)
+            {
+                if (_isDisposed)
+                {
+                    return;
+                }
+
+                _observer?.OnCompleted();
+                Dispose();
+            }
+        }
+
+        public void OnError(Exception error)
+        {
+            lock (_lock)
+            {
+                if (_isDisposed)
+                {
+                    return;
+                }
+
+                _observer?.OnError(error);
+                Dispose();
+            }
+        }
+
+        public void OnNext(IObservable<T> stream)
+        {
+            if (stream == null)
+            {
+                throw new ArgumentNullException(nameof(stream));
+            }
+
+            lock (_lock)
+            {
+                if (_isDisposed)
+                {
+                    throw new ObjectDisposedException(nameof(SwitchObserver));
+                }
+
+                _subscription?.Dispose();
+                _subscription = stream.Subscribe(_observer);
+            }
+        }
+
+        ~SwitchObserver() => Dispose(false);
+
+        public void Dispose()
+        {
+            Dispose(true);
+            GC.SuppressFinalize(this);
+        }
+
+        private void Dispose(bool explicitDisposing)
+        {
+            if (_isDisposed)
+            {
+                return;
+            }
+
+            if (explicitDisposing)
+            {
+                _subscription?.Dispose();
+                _subscription = null;
+                _observer = null;
+            }
+
+            _isDisposed = true;
+        }
+    }
+}
diff --git a/src/RxTelegram.Bot/Utils/Rx/WhereObservable.cs b/src/RxTelegram.Bot/Utils/Rx/WhereObservable.cs
new file mode 100644
index 0000000..3c56938
--- /dev/null
+++ b/src/RxTelegram.Bot/Utils/Rx/WhereObservable.cs
@@ -0,0 +1,101 @@
+using System;
+
+namespace RxTelegram.Bot.Utils.Rx;
+
+internal class WhereObservable<T>(IObservable<T> source, Func<T, bool> predicate) : IObservable<T>
+{
+    private readonly IObservable<T> _source = source ?? throw new ArgumentNullException(nameof(source));
+    private readonly Func<T, bool> _predicate = predicate ?? throw new ArgumentNullException(nameof(predicate));
+
+    public IDisposable Subscribe(IObserver<T> observer)
+    {
+        var where = new WhereObserver(observer, _predicate);
+        var subscription = _source.Subscribe(where);
+        return new DisposableAction(() =>
+        {
+            subscription.Dispose();
+            where.Dispose();
+        });
+    }
+    private sealed class WhereObserver(IObserver<T> observer, Func<T, bool> predicate) : IObserver<T>, IDisposable
+    {
+        private readonly object _lock = new();
+        private readonly Func<T, bool> _predicate = predicate ?? throw new ArgumentNullException(nameof(predicate));
+        private IObserver<T> _observer = observer ?? throw new ArgumentNullException(nameof(observer));
+        private bool _isCompleted;
+        private bool _isDisposed;
+
+        public void OnCompleted()
+        {
+            lock (_lock)
+            {
+                if (_isCompleted || _isDisposed)
+                {
+                    return;
+                }
+
+                _isCompleted = true;
+            _observer?.OnCompleted();
+            }
+        }
+
+        public void OnError(Exception error)
+        {
+            lock (_lock)
+            {
+                if (_isCompleted || _isDisposed)
+                {
+                    return;
+                }
+
+                _isCompleted = true;
+                _observer?.OnError(error);
+            }
+        }
+
+        public void OnNext(T value)
+        {
+            lock (_lock)
+            {
+                if (_isCompleted || _isDisposed)
+                {
+                    return;
+                }
+
+                var isPassed = false;
+                try
+                {
+                    isPassed = _predicate(value);
+                }
+                catch (Exception error)
+                {
+                    OnError(error);
+                }
+                if (isPassed)
+                {
+                    _observer.OnNext(value);
+                }
+            }
+        }
+
+        public void Dispose()
+        {
+            Dispose(true);
+            GC.SuppressFinalize(this);
+        }
+        private void Dispose(bool explicitDisposing)
+        {
+            if (_isDisposed) return;
+
+            lock (_lock)
+            {
+                if (explicitDisposing)
+                {
+                    _observer = null;
+                }
+                _isDisposed = true;
+            }
+        }
+        ~WhereObserver() => Dispose(false);
+    }
+}