带有 Observables 的复杂流

Question

我有一系列图像 (IObservable<ImageSource>) 通过这个“管道”。

1. 使用 OCR 识别每个图像
2. • 如果结果具有有效值，则会将其上传到可以在给定时间（非同时）注册一组结果的服务。
   • 如果结果有任何无效值，则会将它们呈现给用户以进行修复。修复后，该过程将继续。
3. 在此过程中，用户界面应保持响应。

---

问题是当用户必须交互时，我不知道如何处理这种情况。我就是不能这样做

        subscription = images                
            .Do(source => source.Freeze())
            .Select(image => OcrService.Recognize(image))                
            .Subscribe(ocrResults => Upload(ocrResults));

...因为当用户必须修复 ocrResults 时，应保持流程暂停直到接受有效值（即，用户可以执行单击按钮的命令）

怎么说：如果结果无效，等到用户修复它们？

Answer 1

这似乎是一个混合了 UX、WPF 和 Rx 的问题。试图只用 Rx 解决它可能会让你陷入困境。我相信你可以只用 Rx 来解决它，而不再考虑它，但你想要吗？它是否可测试、松耦合且易于维护？

根据我对问题的理解，您必须执行以下步骤

1. 用户上传/选择一些图片
2. 系统对每张图像执行 OCR
3. 如果OCR工具认为图片来源有效，则上传处理结果
4. 如果 OCR 工具认为图像源无效，则用户“修复”结果并上传结果

但这可能更好地描述为

1. 用户上传/选择一些图片
2. 系统对每张图像执行 OCR
3. OCR 的结果被放入验证队列中
4. 虽然结果无效，但需要用户手动将其更新为有效状态。
5. 上传有效结果

所以在我看来，您需要一个基于任务/队列的 UI，以便用户可以看到他们需要处理的无效 OCR 结果。这也告诉我，如果涉及一个人，它可能应该在 Rx 查询之外。

第 1 步 - 执行 ORC

subscription = images                
        .Subscribe(image=>
        {
          //image.Freeze() --probably should be done by the source sequence
          var result = _ocrService.Recognize(image);
          _validator.Enqueue(result);
        });

第 2 步 - 验证结果

//In the Enqueue method, if queue is empty, ProcessHead();
//Else add to queue.
//When Head item is updated, ProcessHead();
//ProcessHead method checks if the head item is valid, and if it is uploads it and remove from queue. Once removed from queue, if(!IsEmpty) {ProcessHead();}


//Display Head of the Queue (and probably queue depth) to user so they can interact with it.

第 3 步 - 上传结果

Upload(ocrResults)

所以这里 Rx 只是我们武器库中的一个工具，而不是需要解决所有问题的一把锤子。我发现随着大多数“Rx”问题的规模不断扩大，Rx 只是充当各种Queue 结构的入口和出口点。这允许我们在我们的系统中使队列显式而不是隐式（即隐藏在 Rx 运算符中）。

Answer 2

我假设您的 UploadAsync 方法返回一个 Task 以允许您等待它完成？如果是这样，则有处理任务的 SelectMany 重载。

images.Select(originalImage => ImageOperations.Resize(originalImage))
    .SelectMany(resizedImg => imageUploader.UploadAsync(resizedImg))
    .Subscribe();

Answer 3

假设您有一个实现“用户修复过程”的异步方法：

/* show the image to the user, which fixes it, returns true if fixed, false if should be skipped */
async Task UserFixesTheOcrResults(ocrResults);

那么你的 observable 变成：

subscription = images                
        .Do(source => source.Freeze())
        .Select(image => OcrService.Recognize(image))
        .Select(ocrResults=> {
            if (ocrResults.IsValid)
                return Observable.Return(ocrResults);
            else
                return UserFixesTheOcrResults(ocrResults).ToObservable().Select(_ => ocrResults)
        })
        .Concat()             
        .Subscribe(ocrResults => Upload(ocrResults));