Improve UpdateManager

[HedgehogNSK]

Hi there. :) I have an idea how to improve the UpdateManager. But I don't have much time to fully accomplish this task. So I share my yet not working draft code here. I suppose you will understand the idea.
The idea is bring more reactivity to UpdateManager.

RIght now

UpdateManager uses Enum for each type of update request and it uses several classes to process and distribute responses. And the core uses strange connection between async method and creates reactive subscription from it.
UpdateManager.ctor creates instance of inner type called Observable for each type of request separate which subscribes inside its own constructor to internal IDictionary<UpdateTypeWrapper<UpdateType?>, List<object>> . Each subscription tryies to run RunUpdateSafe method which has to be launched only once. Inside try catch block there are check if it runs only once. Then it calls the Task RunUpdate which runs until there at least 1 subscriber, that setups request and calls _telegramBot.GetUpdate(getUpdate, _cancellationTokenSource.Token);. Each second inside of the task it checks result. And at last it calls DistributeUpdates methods that uses List<Action<Update>> to spread recieved data from Telegram server. Where each action set by enum, and calls OnNext method. As a resault each second there are a lot of unnecessary calls of many methods and we have almost 500 lines of code that very hard to read.

How to improve

ALL of those classes and methods could be collapsed in 2 much more smaller RX methods. Each type of request could be easily filtred from main stream when and only when it has subscribers.

public class UpdateManager : IUpdateManager
{
  const int timeout = 60;
  public ISubject<Update> OnUpdate {get;}= new Subject<Update>();

    GetUpdate Create(int? offset) => new GetUpdate
    {
      Offset = offset,
      Timeout = timeout,

      // if there is a null value in the list, it means that all updates are allowed
      AllowedUpdates = UpdateTypes.Contains(null!) ? null
                  : UpdateTypes.Where(x => x.HasValue).Select(x => x.Value)
    };
    
  ///Creates an observable for update
  public IObservable<Update> Updates(ITelegramBot telegramBot)
  {
    int? offset = null;
    var getUpdate = Create(offset);
    var observable = Observable.Start(() => Create(offset))
      .SelectMany(Observable.FromAsync(() => telegramBot.GetUpdate(getUpdate, _cancellationTokenSource.Token)))
      .Repeat()
      .Publish()
      .RefCount()
      .Where(_updates => _updates.Any())
      .Do(_result => offset = _result.Max(x => x.UpdateId) + 1)
      .SelectMany(x => x);

    return observable;
  }
    
  public  IDisposable Launcher(ITelegramBot telegramBot) 
      => Updates(telegramBot).Multicast(OnUpdate).Subscribe();
      
  /// And since now we can get necessary type of message 
  ///while listening the main stream by using Where
  public IObservable<Message> Message1 => OnUpdate
    .Where(_update => _update.Message != null) 
    .Select(_update => _update.Message); 

  public IObservable<CallbackQuery> CallbackQuery1 => OnUpdate
    .Where(_update => _update.CallbackQuery != null)
   .Select(_update => _update.CallbackQuery);
  //Same approach for all other types of update request
 }

So as you can see Message1 and CallbackQuery1 properties do the same that Message and CallbackQuery respectively do. In addition, a user can now connect to different bots at one stream using Launcher method. I'm not sure if this feature is necessary for someone right now but it's just appeared by itself cause of RX approach. And as I said before, I didn't finished the code. I had problem with restoring request after RefCount drops to 0 and then a new subcriber appeares.

A subscriptions to the current Updates produces periodic errors In my own project and the subscriptions were breaking. So I should figure out where I could handle this errors. And while I was delving into code I came to this solution. This draft code has been lying idle for 3 months now. That's why I decide to share my thoughts and code sample. I suppose that it will significantly improve memmory efficiency and code readability. Probably this change should slightly improve speed of code as well.

P.S. The subscription were breaking most time due to HTTP Request response error 502 from a host of my bot. So I make a workaround to restore connection. Perhaps it could be useful to someone.

//TelegramBot bot 
var observableMessages = bot.Updates.Message
       .Catch((Exception _ex) =>
       {
         Console.WriteLine("Updates.Messages exception: {0}", _ex);
         ExceptionHandler.OnError(_ex);
         return Observable.Empty<Message>().Delay(TimeSpan.FromSeconds(1));
       })
       .Repeat()
       .Select(_message =>/*Your processing*/);
       //And then subsribe observable

[niklasweimann]

Hey @HedgehogNSK 👋
This is an interesting approach and I do appreciate your effort. Do I understand your suggestion correctly, that it would require the library to also depend on System.Reactive? I would like to keep the dependencies of this library close to 0. In the long term I would also like to remove Newtonsoft.Json from the library.

[HedgehogNSK]

Yes. In my suggestion it should depends on System.Reavctive.
Ok. I understood, that's unwanted depedency.

[niklasweimann]

Ok, but I will check if there is a way to improve the UpdateManager without the dependency 😊 Thanks for bringing this up to our attention.