我应该如何使用 Rx + DynamicData 定期检查许多在线服务的更新？

Question

我有一个基本的日历/议程应用程序，它将列出一系列帐户和日历中的最新事件。例如，假设我有 3 个帐户：两个不同的 Microsoft 帐户和一个 Google 帐户。我目前将它们存储为服务中名为

SourceCache<Account, string>

的

Accounts

(

AccountsService

)。

SourceCache<T1,T2>

是 DynamicData 的一部分……它基本上构成了反应式集合。我希望它具有反应性，这样当我添加或删除帐户时，应用程序中的所有内容（设置页面、日历页面等）都会自动更新。

现在，每个帐户可以有多个

Calendar

。对于每个

Calendar

，我想下载所有即将推出的

CalendarEvent

。问题是我需要定期执行此操作，以查看是否添加了新事件或事件是否已更改。

这就是我目前正在做的事情，但我担心这可能是非常糟糕的 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

s 在线 ID（已修复）绑定的密钥，所以新下载的事件只是替换旧的事件在缓存中。我在用户界面上没有看到任何闪烁。

**更正：它似乎有效是因为我在另一次试验中不小心留下了一个黑客。在原始帐户 SourceCache 上，我正在运行一个调用 Accounts.Refresh() 的计时器。如果我把它拿出来，就没有任何作用。

这是正确的方法吗？请赐教...我在 Rx 和 DynamicData 方面遇到了一些困难。操作员太多了我还不知道该怎么办。

谢谢！

Answer 1

您似乎已经回答了您自己的问题，并且代码看起来比原始代码更优化。不过我建议进行一些优化：

1. 使用 AutoRefreshOnObservable 这将允许您创建单个 observable，而不是使用 AutoRefresh 两次。

2. 我无法从代码中判断 CalendarEventsObservable 是否返回单个值或集合。如果它返回一个集合，您可以使用 EditDiff，而不是使用 AddOrUpdate。 EditDiff 期望指定比较的相等性，并防止触发不必要的通知。

Answer 2

我稍微重写了代码，看起来效果好多了。它甚至可以在没有黑客的情况下工作，我在原始的

Refresh()

Account

上手动调用

SourceCache

。

我意识到我一直在滥用 Publish 方法。此外，在转换语句中订阅可观察对象是行不通的，因为它不是 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();

现在，在 UI 的视图模型中，我订阅了

SourceCache<CalendarEvent,string>

；