ReactiveUI 7.0如何处理抛出异常时释放的可观察对象

Question

我真的开始研究这个 rx 的东西了...基本上，我跟随 this video 只是为了在我开始真正使用它之前教自己更多关于 ReactiveUI 的知识！

当我们使用 WhenAnyValue 执行节流搜索时，我试图创造一种情况。而且，如果搜索函数抛出异常，我想在视图模型上设置一个名为IsError 的属性（这样我就可以显示一个 X 或其他东西）。这是我工作的 ViewModel 的重要部分：

public ReactiveCommand<string, IEnumerable<DictItem>> SearchCmmand;

...  in vm constructor:

//create our command async from task. executes on worker thread
SearchCmmand = ReactiveCommand.CreateFromTask<string, IEnumerable<DicItem>>(async x => {
    this.IsError = false;
    //this may throw an exception:
    return await GetFilteredAsync(this.SearchText); 
  });

//SearchCommand is subscribable.  
//set the Filtered Items property. executes on main thread
SearchCmmand.Subscribe(filteredItems => {
  this.FilteredItems = filteredItems;
});

//any unhandled exceptions that are thown in SearchCommand will bubble up through the ThrownExceptions observable    
SearchCmmand.ThrownExceptions.Subscribe(ex=> {
  this.IsError = true;
  //but after this, then the WhenAnyValue no longer continues to work.
  //how to get it back?
});

//invoke the command when SearchText changes
this.WhenAnyValue(v => v.SearchText)
  .Throttle(TimeSpan.FromMilliseconds(500))
  .InvokeCommand(SearchCmmand);

这很有效。当我的GetFilteredAsync 抛出异常时，SearchCmmand.ThrownExceptions 被调用，我可以设置我的IsError 属性。

但是，当SearchCmmand.ThrownExceptions 第一次发生时，this.WhenAnyValue(v => v.SearchText) 停止工作。我可以看到它被处理掉了。对 SearchText 的后续更改不会调用该命令。 （尽管如果我绑定了一个按钮，该命令仍然有效）

这似乎是预期的行为，但我们怎样才能让 observable 再次工作呢？我意识到我可以将它全部包装在 try/catch 中并返回不是异常的东西，但是，我在video（大约 39:03）中看到，在他的情况下，searchtext 在异常之后继续工作被抛出？ （该视频的源代码是here）。

我还看到 here 一些关于 UserError 的内容，但现在标记为旧版。

Answer 1

好的，所以我有一些工作，但我会发布它。我必须处理几个问题。一个是我在我的命令异步任务代码中设置我的IsError=false 属性（它在后台线程上触发并因此引发异常），另一个是处理如何在 ThrownExceptions 冒泡后重新订阅 observable向上。我发现有两种方法/解决方法有效：

1. 处理命令代码中的异常，这样 ThrownExceptions 就不会真正被触发。
2. 如果 ThrownExceptions 确实被触发，则处置并重新订阅 WhenAnyValue observable 以使其继续运行。 （这需要为 WhenAnyValue 对象保留一个变量。）

这里是似乎工作的整个视图模型代码。警告：我自己是 rx/rxui 的新手，我不知道这是否是做这一切的最佳方式！我在想可能有更好的方法！

public class SearchViewModel1 : ReactiveObject {

IEnumerable<DictItem> itemList; //holds the master items. used like a repo (just for demo, i'd use a separate repo or service class for real)

ObservableAsPropertyHelper<bool> _isBusy;
public bool IsBusy {
  get { return _isBusy.Value; }
}

bool _isError;
public bool IsError {
  get { return _isError; }
  set { this.RaiseAndSetIfChanged(ref _isError, value); }
}

//the filtered items property that we want to bind our list to
IEnumerable<DictItem> _filteredItems;
public IEnumerable<DictItem> FilteredItems {
  get { return _filteredItems; }
  set { this.RaiseAndSetIfChanged(ref _filteredItems, value); }
}

//the search text, this will be bound
//and this viewmodel will respond to changes to the property. 
string _searchText;
public string SearchText {
  get { return _searchText; }
  set { this.RaiseAndSetIfChanged(ref _searchText, value); }
}

//this is the reacive command that takes a string as a parameter, 
public ReactiveCommand<string, IEnumerable<DictItem>> SearchCmmand { get; set; }

//a reference to our observable in case we lose it and need to resubscribe
IDisposable whenAnySearchText;


//convenience method to set the IsError property. can be called by a worker thread
void SetIsErrorFromWorkerThread(bool isError) {
  Observable.Return(isError)
    .SubscribeOn(RxApp.MainThreadScheduler)
    .Subscribe(b => this.IsError = b);
}


//constructor is where we wire it all up
public SearchViewModel1(IEnumerable<DictItem> itemList) {

  this.itemList = itemList;

  FilteredItems = itemList;

  //this observable keeps track of when SearchText is blank.
  var searchTextHasValue = this.WhenAnyValue(x => x.SearchText)
    .Select(x => !string.IsNullOrWhiteSpace(x));

  //create our command async from task.
  //it will only actually fire if searchTextHasValue is true.
  SearchCmmand = ReactiveCommand.CreateFromTask<string, IEnumerable<DictItem>>(async x => {
        SetIsErrorFromWorkerThread(false);
        //first we'll try to capture any exceptions here, so we don't lose the observable.
        try {
          return await GetFilteredAsync(SearchText, itemList);
        } catch (Exception ex) {
          SetIsErrorFromWorkerThread(true);
          return Enumerable.Empty<DictItem>();
        }
    },
    searchTextHasValue);

  //searchCommand is subscribable.  set the Filtered Items property synchronous here on main thread
  SearchCmmand.Subscribe(filteredItems => {
    FilteredItems = filteredItems;
  });

  //any unhandled exceptions that are thown in SearchCommand will bubble up through the ThrownExceptions observable
  SearchCmmand.ThrownExceptions.Subscribe(ex => {
    //note: because we are handling exceptions in the command code,
    //this should be a very last-case and never-happen scenario.  
    //but we seem to be able to recover by re-subscribing the observable
    IsError = true;
    //we have lost the subscription.  so set it again?
    //is this even a good idea?
    whenAnySearchText.Dispose();
    whenAnySearchText = this.WhenAnyValue(v => v.SearchText)
      .Throttle(TimeSpan.FromMilliseconds(500))
      .InvokeCommand(SearchCmmand);
  });

  //the IsBusy can just be wired from the Command observable stream
  _isBusy = SearchCmmand.IsExecuting.ToProperty(this, vm => vm.IsBusy);

  //bind our whenAnySearchText 
  whenAnySearchText = this.WhenAnyValue(v => v.SearchText)
    .Throttle(TimeSpan.FromMilliseconds(500))
    .InvokeCommand(SearchCmmand);
}

 //the task to run the search/filter
 async Task<IEnumerable<DictItem>> GetFilteredAsync(string filterText, IEnumerable<DictItem> items) {
   await Task.Delay(1000);
   if (filterText.Length == 5) {
     throw new InvalidOperationException("You cannot search 5 characters!  Why? No reason, it's contrived.");
   }
   return items.Where(x => x.Name.IndexOf(filterText, StringComparison.OrdinalIgnoreCase) >= 0);
 }

}

Answer 2

你可以使用interactions

public enum ErrorRecoveryOption
{
    Retry,
    Abort
}

public static class Interactions
{
    public static readonly Interaction<Exception, ErrorRecoveryOption> Errors = new Interaction<Exception, ErrorRecoveryOption>();
}

public class SomeViewModel : ReactiveObject
{
    public async Task SomeMethodAsync()
    {
        while (true)
        {
            Exception failure = null;

            try
            {
                DoSomethingThatMightFail();
            }
            catch (Exception ex)
            {
                failure = ex;
            }

            if (failure == null)
            {
                break;
            }

            // this will throw if nothing handles the interaction
            var recovery = await Interactions.Errors.Handle(failure);

            if (recovery == ErrorRecoveryOption.Abort)
            {
                break;
            }
        }
    }
}

public class RootView
{
    public RootView()
    {
        Interactions.Errors.RegisterHandler(
            async interaction =>
            {
                var action = await this.DisplayAlert(
                    "Error",
                    "Something bad has happened. What do you want to do?",
                    "RETRY",
                    "ABORT");

                interaction.SetOutput(action ? ErrorRecoveryOption.Retry : ErrorRecoveryOption.Abort);
            });
    }
}

看看这个： https://docs.reactiveui.net/en/user-guide/interactions/index.html