Mixing reactive programming with non-reactive return requirements

Question

Variable context from an initial non-reactive caller

The whole application cannot be reactive i.e. this method needs to return a result here

public string GetTextOfInterest()
{           
    var searchQuery = "Single Responsibility";
    var titleShouldContain = "SRP";
    var uriShouldContain = "pattern";
    var pageShouldContain = "arguments";
    return this.NonReactive(searchQuery, titleShouldContain, uriShouldContain,pageShouldContain);
}

Non-reactive blocking implementation

This non-reactive implementation works fine

private string NonReactive(string searchQuery, string titleShouldContain, string uriShouldContain, string pageShouldContain)
{
    // Googler returns result directly here and does not move on before          
    var gooResult = this.googler.Search(searchQuery);
    if (gooResult.Title.Contains(titleShouldContain) && gooResult.Uri.Contains(uriShouldContain))
    {
        // Blocking here again to wait for this result
        var scrResult = this.scraper.Scrape(gooResult.Uri);
        if (scrResult.Body.Contains(pageShouldContain))
        {
            // Blocking up to this point ensures sequential execution and ability to return this final result
            return scrResult.Body;
        }
    }
    return null;
}

How to answer reactively?

This reactive implementation does not work

private string ReactiveButUseless(string searchQuery, string titleShouldContain, string uriShouldContain, string pageShouldContain)
{            
    googler.Stream.Subscribe(
        gooResult =>
            {
                if (gooResult.Title.Contains(titleShouldContain) && gooResult.Uri.Contains(uriShouldContain))
                {
                    scraper.Stream.Subscribe(
                        scrResult =>
                            {
                                if (scrResult.Body.Contains(pageShouldContain))
                                {
                                    // Get final result here, but no good as method has already returned (null)
                                    var finalResult = scrResult.Body;
                                }
                            });
                    scraper.Scrape(gooResult.Uri);                            
                }
            });      
    googler.Search(searchQuery);            
    return null;
}

Is there any way of implementing a working reactive solution?

• without having to change the Googler and Scraper implementation because they need to stay clean and reusable for other purposes (otherwise could pass in delegates)
• and without blocking? Could wait for result of stream and block, but then it is no better than the non-reactive blocking implementation? Or is it just not possible to mix reactive and non-reactive styles without blocking?

Answer 1

Short answer is no.

The two things you want are mutually exclusive, you cannot continue a code block while waiting for an answer from an asynchronous server and somehow guarantee to get it back before the code block returns (with the exception of async/await [see below]).

That being said if you want to use Reactive, there are a couple approaches to try:

First some clean up:

    private IObservable<string> GetScrapeStream(Uri resultUri, string pageShouldContain)
    {
        return Observable.Create<string>(observer =>
        {
            var subscription =
                scraper.Stream
                .Where(result => result.Body.Contains(pageShouldContain))
                .Select(result => result.Body)
                .Subscribe(observer);

            scraper.Scrape(resultUri);
            return subscription;
        });
    }

    private IObservable<string> GetGoogleStream(string searchQuery, 
                                                string titleShouldContain, 
                                                string uriShouldContain, 
                                                string pageShouldContain)
    {
        return Observable.Create<string>(observer =>
        {
            var subscription =
                googler.Stream
                .Where(result => result.Title.Contains(titleShouldContain) &&
                                 result.Uri.Contains(uriShouldContain))
                .SelectMany(result =>
                {
                    return GetScrapeStream(result.Uri, pageShouldContain);
                })
                .Subscribe(observer);

            googler.Search(searchQuery);
            return subscription;
        });
    }

Now since I already said you can't have your cake and eat it too, you have two options really:

//This will block, and Last() is actually deprecated, so it will probably be going away
private string ReactiveButBlocking(string searchQuery, 
                                   string titleShouldContain, 
                                   string uriShouldContain, 
                                   string pageShouldContain)
{
  return GetGoogleStream(searchQuery, titleShouldContain, uriShouldContain, pageShouldContain)
         .Last();
}

If you have at least .Net 4.5 you also have async/await available to you, which are supported very nicely in Rx, however, your method signature will change slightly, and in the caller method you will need to do string result = await ReactiveAndAsync(...); Realize though that this is not your standard blocking, because it actually frees up the the thread until the method returns a value.

see reference

private async Task<string> ReactiveAndAsync(string searchQuery, 
                                  string titleShouldContain, 
                                  string uriShouldContain, 
                                  string pageShouldContain)
{
  return await GetGoogleStream(searchQuery, titleShouldContain, uriShouldContain, pageShouldContain);
}

---

tl;dr

Rx doesn't do anything magical, it won't defeat causality and it isn't perfect for every use case, at least not until 3.0. However, it will let you create logic streams that hopefully improve readability, extensibility and a separation of concerns.