Complex flow with Observables

Question

I have a sequence of Images (IObservable<ImageSource>) that goes through this "pipeline".

1. Each image is recognized using OCR
2. • If the results have valid values, the are uploaded to a service that can register a set of results at a given time (not concurrently).
   • If the results have any invalid value, they are presented to the user in order to fix them. After they are fixed, the process continues.
3. During the process, the UI should stay responsive.

---

The problem is that I don't know how to handle the case when the user has to interact. I just cannot do this

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

...because when ocrResults have to be fixed by the user, the flow should be kept on hold until the valid values are accepted (ie. the user could execute a Command clicking a Button)

How do I say: if the results are NOT valid, wait until the user fixes them?

Answer 1

I'm assuming your UploadAsync method returns a Task to allow you to wait for it to finished? If so, there are overloads of SelectMany that handle tasks.

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

Answer 2

Assuming you've got an async method which implements the "user fix process":

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

Then your observable becomes:

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));

Answer 3

This seems to be a mix of UX, WPF and Rx all wrapped up in one problem. Trying to solve it with only Rx is probably going to send you in to a tail spin. I am sure you could solve it with just Rx, and no more thought about it, but would you want to? Would it be testable, loosely coupled and easy to maintain?

In my understanding of the problem you have to following steps

1. User Uploads/Selects some images
2. The system performs OCR on each image
3. If the OCR tool deems the image source to be valid, the result of the processing is uploaded
4. If the OCR tool deems the image source to be invalid, the user "fixes" the result and the result is uploaded

But this may be better described as

1. User Uploads/Selects some images
2. The system performs OCR on each image
3. The result of the OCR is placed in a validation queue
4. While the result is invalid, a user is required to manually update it to a valid state.
5. The valid result is uploaded

So this to me seem that you need a task/queue based UI so that a User can see invalid OCR results that they need to work on. This also then tells me that if a person is involved, that it should probably be outside of the Rx query.

Step 1 - Perform ORC

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

Step 2 - Validate Result

//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.

Step 3 - Upload result

Upload(ocrResults)

So here Rx is just a tool in our arsenal, not the one hammer that needs to solve all problems. I have found that with most "Rx" problems that grow in size, that Rx just acts as the entry and exit points for various Queue structures. This allows us to make the queuing in our system explicit instead of implicit (i.e. hidden inside of Rx operators).