问题检测 RX 中可观察到的 OnCompleted

Question

我在 RX 中检测 onComplited 操作时遇到问题。 我正在从 Web 服务将数据加载到 ListBox。

基本情况运行良好： - “Complited item”位于底部

App.ViewModel.LoadData();

IObservable<WebServiceClass.ItemGetValues> observable = App.ViewModel.Items
   .ToObservable(Scheduler.NewThread);

var items = new ObservableCollection<WebServiceClass.ItemGetValues>();
ListBox1.ItemsSource = items;

observable.ObserveOnDispatcher().Subscribe(
    item => { items.Add( new WebServiceClass.ItemGetValues(item.nazwa, item.skrot, item.id) ); },
    () => { items.Add( new WebServiceClass.ItemGetValues("Complited", "", "") ); }
);

但是当我使用下面的代码为每个项目调用 Web 服务时，首先添加“Complited item”

App.ViewModel.LoadData();

IObservable<WebServiceClass.ItemGetValues> observable = App.ViewModel.Items
   .ToObservable(Scheduler.NewThread);

var items = new ObservableCollection<WebServiceClass.ItemGetValues>();
ListBox1.ItemsSource = items;

observable.ObserveOnDispatcher().Subscribe
    (item =>
        {
            //items.Add(item);
            var request = Observable.FromAsyncPattern<string, string>(client.BeginGetLastValue, client.EndGetLastValue);
            request(item.skrot).ObserveOnDispatcher().Subscribe(
                    (it) =>
                    {
                        Func<string, WebServiceClass.ItemGetValues> deserializeFirst = r =>
                                        ((List<WebServiceClass.ItemGetValues>)JsonConvert
                                        .DeserializeObject(r,
                                            typeof(List<WebServiceClass.ItemGetValues>)))
                                        .First();
                        item.zm_dzienna = deserializeFirst(it).zm_dzienna;
                        items.Add(item);
                    }
            );
        },
    () => { items.Add(new WebServiceClass.ItemGetValues("Complited", "Complited", "0")); }
    );

如何正确检测 onComplited 操作？

---

解决方案

deserializeFirst 函数中出现 nullReference 异常问题。

App.ViewModel.LoadData();

IObservable<WebServiceClass.ItemGetValues> observable = App.ViewModel.Items
    .ToObservable(Scheduler.NewThread);

var items = new ObservableCollection<WebServiceClass.ItemGetValues>();
ListBox1.ItemsSource = items;

var request = Observable.FromAsyncPattern<string, string>(client.BeginGetLastValue, client.EndGetLastValue);

observable.SelectMany(
    item => request(item.skrot).Select(it => {
                                                Func<string, WebServiceClass.ItemGetValues> deserializeFirst = r =>
                                                ((List<WebServiceClass.ItemGetValues>)JsonConvert
                                                .DeserializeObject(r,
                                                    typeof(List<WebServiceClass.ItemGetValues>)))
                                                .First();
                                                item.zm_dzienna = deserializeFirst(it).zm_dzienna;
                                                return item; 
                                                })
    ).SubscribeOnDispatcher().Subscribe(
            result => { Dispatcher.BeginInvoke(delegate { items.Add(result); }); },
            () => { Dispatcher.BeginInvoke(delegate { items.Add(new WebServiceClass.ItemGetValues("c","c","c")); }); }
        );

谨致问候， 武卡斯

Answer 1

您需要将内部查询拉入 Rx 序列，而不是从订阅中运行它。这样做可以让您获得 Rx 的强大功能、可组合性和可测试性。并为您带来您想要的结果。

这是实现所需目标的一个开始。

Func<string, WebServiceClass.ItemGetValues> deserializeFirst = r =>
        ((List<WebServiceClass.ItemGetValues>)JsonConvert
        .DeserializeObject(r,typeof(List<WebServiceClass.ItemGetValues>)))
        .First();
var request = Observable.FromAsyncPattern<string, string>(client.BeginGetLastValue, 
                                                          client.EndGetLastValue);
observable
    .SelectMany((item) => request(item.skrot))
    .Select((it) => item.zm_dzienna = deserializeFirst(it).zm_dzienna)
    .ObserveOnDispatcher()
    .Subscribe(
        item => items.Add(item), 
        () => items.Add(new WebServiceClass.ItemGetValues("Complited", "Complited", "0")));