Как мне использовать Rx + DynamicData для периодической проверки обновлений многих онлайн-сервисов?

Question

У меня есть простое приложение Календарь / Повестка дня, которое будет перечислять самые последние события из ряда учетных записей и календарей.Например, скажем, у меня есть 3 учетные записи: две разные учетные записи Microsoft и одна учетная запись Google.В настоящее время я храню их как SourceCache<Account, string> с именем Accounts в службе (AccountsService).

SourceCache<T1,T2> является частью DynamicData ..., которая в основном создает Reactive Collections.Я хочу, чтобы он был реактивным, чтобы при добавлении или удалении учетной записи все содержимое приложения (страница настроек, страницы календаря и т. Д.) Обновлялось автоматически.

Теперь каждая учетная запись может иметь несколько Calendar s.,И для каждого из этих Calendar я хочу загрузить все предстоящие CalendarEvent s.Суть в том, что мне нужно делать это через регулярные промежутки времени, чтобы увидеть, были ли добавлены новые события или были ли они изменены.

Вот как я сейчас это делаю, но, боюсь, это возможнодействительно плохой Rx.

var calendarSet = this.accountsService.Accounts.Connect()
    .ObserveOn(RxApp.TaskpoolScheduler)
    .TransformMany(x =>
    {
        ReadOnlyObservableCollection<Models.Calendar> subCalendars;
        x.CalendarService.Calendars.Connect()
            .AutoRefreshOnObservable(calendar => calendar.IsEnabledChanged)
            .AutoRefreshOnObservable(calendar => calendar.IsColorChanged)
            .Filter(calendar=>calendar.IsEnabled)
            .Bind(out subCalendars)
            .Subscribe();
        return subCalendars;
     }, x => x.CacheKey)
     .ObserveOnDispatcher()
     .Publish();

calendarSet
    .Bind(out calendars)
    .Subscribe();


var eventSet = calendarSet
    .ObserveOn(RxApp.TaskpoolScheduler)
    .Transform( calendar =>
    {
        var events = new List<Models.CalendarEvent>();
        Debug.WriteLine(calendar.Name);
        calendar.CalendarService.CalendarEventsObservable(calendar).Subscribe(items =>
        {
            events.AddRange(items);
        });
        return events;
    })
    .TransformMany(x => x, x => x.Key)
    .Filter(x => x.EndDateTime > DateTimeOffset.Now)
    .Sort(new Models.CalendarEventSorter())
    .ObserveOnDispatcher()
    .Bind(out calendarEvents)
    .Subscribe();

calendarSet.Connect();

Большая часть - это то, где события также загружаются через подписку на наблюдаемое.Вот где я установил таймер, который позволяет мне контролировать частоту проверки онлайн-сервисов.Это выглядит так (20 секунд только для тестирования!):

public IObservable<List<Models.CalendarEvent>> CalendarEventsObservable(Models.Calendar calendar)
    {
        var obs = Observable.Interval(TimeSpan.FromSeconds(20)).SelectMany(async x =>
        {
            var items = await GetAllEventsForCalendarAsync(calendar);
            return items;
        });

        return obs;
    }

Кажется, это работает!Я могу включить / отключить определенные календари, и события будут появляться или исчезать из моего связанного списка.Я вижу, что события обновляются через регулярные промежутки времени… и я предполагаю, что, поскольку я использую TransformMany с ключом, привязанным к сетевому идентификатору CalenderEvent (который фиксирован), вновь загруженные события просто заменяют старыев кеше.Я не вижу мерцания в пользовательском интерфейсе.

** Исправление: кажется, работает из-за взлома, который я случайно оставил в другом испытании.На исходных учетных записях SourceCache я запускаю таймер, который вызывает Accounts.Refresh ().Если я возьму это, ничего не работает.

Это правильный способ сделать это?Пожалуйста, просветите меня ... Я немного борюсь с Rx и DynamicData.Операторов так много, что я пока не знаю, что делать.

Спасибо!

Answer 1

Вы, кажется, ответили на свой вопрос, и код выглядит более оптимальным, чем оригинал.Однако я предлагаю несколько вариантов оптимизации:

1. Использовать AutoRefreshOnObservable, который позволит вам создать одну наблюдаемую, а не дважды использовать AutoRefresh.

2. Iиз кода невозможно определить, возвращает ли CalendarEventsObservable одно значение или коллекцию.Если он возвращает коллекцию, вместо использования AddOrUpdate вы можете использовать EditDiff.EditDiff ожидает равенства по сравнению с указанным и предотвращает запуск ненужных уведомлений.

Answer 2

Я немного переписал код, и он, кажется, работает намного лучше.Это даже работает без взлома, когда я вручную вызываю Refresh() на оригинальном Account SourceCache.

Я понял, что неправильно использовал метод публикации.Кроме того, подписка на наблюдаемое в операторе преобразования не будет работать, потому что это не наблюдаемая DynamicData (с наборами изменений).Вместо этого я решил создать SubscribeMany, чтобы подписаться на все CalendarEventObservable с каждого Calendar, и в логике действий подписки я бы заполнил новый SourceCache из CalendarEvent с.Никакие события не будут удалены, но дублирующие события просто перезапишут старые из-за ключа кеша, и я мог отфильтровывать события из кедаров, которые не были выбраны.

 private SourceCache<Models.CalendarEvent, string> calendarEventCache;
 public IObservableCache<Models.CalendarEvent, string> CalendarEventCache => calendarEventCache.Connect().AutoRefreshOnObservable(x=>x.Parent.IsEnabledChanged).Filter(x=>x.Parent.IsEnabled).AsObservableCache();

//////////////////
///IN CONSTRUCTOR:

        calendarEventCache = new SourceCache<Models.CalendarEvent, string>(x => x.Key);

        calendarsCache = this.accountsService.Accounts.Connect()
            .ObserveOn(RxApp.TaskpoolScheduler)
            .TransformMany(x =>
            {
                ReadOnlyObservableCollection<Models.Calendar> subCalendars;
                x.CalendarService.Calendars.Connect()
                            .AutoRefreshOnObservable(calendar => calendar.IsEnabledChanged)
                            .AutoRefreshOnObservable(calendar => calendar.IsColorChanged)
                            .Filter(calendar => calendar.IsEnabled)
                            .Bind(out subCalendars)
                            .Subscribe();
                return subCalendars;
            }, x => x.CacheKey)
            .ObserveOnDispatcher()
            .AsObservableCache();

        calendarsCache.Connect()
            .ObserveOn(RxApp.TaskpoolScheduler)
            .SubscribeMany(calendar =>
            {
                return calendar.CalendarService.CalendarEventsObservable(calendar).Subscribe(calendarEvent =>
               {
                   calendarEventCache.AddOrUpdate(calendarEvent);
               });
            })
            .Subscribe();

Теперь в моей модели представления для пользовательского интерфейса я подписываюсь на SourceCache<CalendarEvent,string>;