system.reactive - 在 C# 中使用 Reactive Extensions 时如何显示进度

Question

我在 C# 中使用响应式(Reactive)扩展来执行一些计算。这是我的代码到目前为止的样子。我试图将代码包装起来，以便在我的计算方法中执行一系列任务时显示进度

这是可观察的

IObservable<ResultWithProgress<SampleResult>> Calculate(){
  return Observable.Create<ResultWithProgress<SampleResult>>(obs => {
     var someTask = DoSomeTask1();
     obs.OnNext(new ResultWithProgress(){Progress = 25, ProgressText     ="Completed Task1"});
     var someOtherTask = DoSomeMoreTask();
     obs.OnNext(new ResultWithProgress(){Progress = 50, ProgressText ="Completed Task2"});
     var calcResult = DoSomeMoreTask2();
     obs.OnNext(new ResultWithProgress(){Progress = 75, ProgressText = "Completed Task3"});
     var calcResult = FinalCalc();
     obs.OnNext(new ResultWithProgress(){Progress = 100, ProgressText ="Completed Task4", Result = calcResult});
     obs.OnCompleted();
  }

}

结果类包装进度和结果

class ResultWithProgress<T>{
 public int Progress {get; set;}
 public Result T {get; set;}
 public string ProgressText {get; set;}
}

包含最终结果的结果对象
类 SampleResult{}

用法:

Calculate().Subscribe(resultWithProgress => {
  if(resultWithProgress.Result == null) //Show progress using   resultWithProgress.Progress
  else // get the result
})

我不知何故觉得这可能不是最好的方法。感觉在没有 Result 的情况下多次创建 ResultWithProgress 对象似乎是一种代码异味，尤其是当我有 10 个以上的任务要在我的 Calculate() 中执行时

如果您能给我任何有关如何使用它的指示，或者我是否错误地处理了这个问题，我将不胜感激？

Answer 1

此答案使用 Enigmativity 的答案所讨论的相同原则。

此版本使用 Create 的异步重载.

它还使用了 .NET 4.5 IProgress而不是原始 Action<T>报告进展。

struct CalculationProgress
{
    public int Progress { get; private set; }
    public string ProgressText { get; private set; }

    public CalculationProgress(int progress, string progressText)
        : this()
    {
        Progress = progress;
        ProgressText = progressText;
    }
}

public IObservable<Result> Calculate(IProgress<CalculationProgress> progress)
{
    return Observable.Create<Result>((observer, cancellationToken) =>
    {
        // run the work on a background thread
        // so we do not block the subscriber
        // and thus the subscriber has a chance
        // to unsubscribe (and cancel the work if desired)
        return Task.Run(() =>
        {
            DoSomeTask1();
            cancellationToken.ThrowIfCancellationRequested();
            progress.Report(new CalculationProgress(25, "First task"));

            DoSomeTask2();
            cancellationToken.ThrowIfCancellationRequested();
            progress.Report(new CalculationProgress(50, "Second task"));

            DoSomeTask3();
            cancellationToken.ThrowIfCancellationRequested();
            progress.Report(new CalculationProgress(75, "third task"));

            var result = DoFinalCalculation();            
            cancellationToken.ThrowIfCancellationRequested();
            progress.Report(new CalculationProgress(100, "final task"));

            observer.OnNext(result);
        }, cancellationToken);
    });
}