如何订阅长时间运行的消息源

Question

我正在尝试reactiveui，但我不明白如何让一个简单的场景发挥作用：我有一个在聊天室中侦听消息的方法。因此，它是长期运行的，并在发现消息时触发事件。

使用reactiveui，我想在窗口打开时启动这种长期运行的方法，并在屏幕上的列表框中填充新消息。因为我使用的是rx，所以我认为我需要一个长期运行方法的IOobservable版本，所以我做了一个这样的：

    public static IObservable<Message> ObservableStream(int roomid, CancellationToken token)
    {
        return Observable.Create<Message>(
            async (IObserver<Message> observer) =>
                {
                 ...
                }
         );
    }

但是，我不知道如何将其引入reactiveui。我需要ObservableAsPropertyHelper<List<Message>>吗？目前，我只是在Task.Factory.Startnew中启动长时间运行的方法，然后在事件上手动添加到绑定到前端列表框的消息列表中。这是有效的，但它没有使用任何reactiveui，我突然想到应该有一种reactiveui的方法来做到这一点：

public class MainWindowViewModel : ReactiveObject
{
    private ThreadSafeObservableCollection<Message> _Messages;
    public ThreadSafeObservableCollection<Message> Messages
    {
        get { return _Messages; }
        set { 
            this.RaiseAndSetIfChanged(x => x._Messages, value);
        }
    }

    public MainWindowViewModel()
    {
        Client.NewMessage += (sender, args) => Messages.Add(args.Message);
        var task = Task.Factory.StartNew(() => Client.GetStream(token), token, TaskCreationOptions.LongRunning, TaskScheduler.Current);
    }
}

// IN the code-behind
this.OneWayBind(ViewModel, x => x.Messages, x => x.MessageList.ItemsSource);

Answer 1

怎么样：

var Messages = ObservableStream(...).CreateCollection();

然后，您可以收听ItemAdded等的Messages，也可以通过OneWayBind绑定它，UI会自动更新。

我需要ObservableAsPropertyHelper>吗？

因此，通常情况下，这对大多数Web API调用来说都是个好主意，但由于您每次都在流式传输列表，而不是替换，因此您需要在启动时创建一个集合，并在项目进入时向其添加项目。另一种方法是：

var Messages = new ReactiveList<Message>();
ObservableStream(...).ObserveOn(RxApp.MainThreadScheduler).Subscribe(x => Message.Add(x));