Stream processing, Event sourcing, Reactive, CEP… and making sense of it all

Imagine a simple iPhone application which has the following components:

        // a text box in which the user types what they are searching for.
	// a submit button on which the user can optionally press to initiate the search.
	// a List<T> of search results
	// a service that takes the search query and executes the search.

This is how you can implement the above in a FRP manner:

        public class SearchViewModel : ISearchViewModel
	{
	    public ReactiveList<SearchResults> SearchResults { get; set; }

	    private string searchQuery;
	    public string SearchQuery {
	        get { return searchQuery; }
	        set { this.RaiseAndSetIfChanged(ref searchQuery, value); }
	 
	    }
	 
	    public ReactiveCommand<List<SearchResults>> Search { get; set; }
	 
	    public ISearchService SearchService { get; set; }
	}


	public SearchViewModel(ISearchService searchService = null) : ReactiveObject, IRoutableViewHost
	{
	    SearchService = searchService ?? Locator.Current.GetService<ISearchService>();
	 
	    // Here we're describing here, in a *declarative way*, the conditions in
	    // which the Search command is enabled.  Now our Command IsEnabled is
	    // perfectly efficient, because we're only updating the UI in the scenario
	    // when it should change.
	    var canSearch = this.WhenAny(x => x.SearchQuery, x => !String.IsNullOrWhiteSpace(x.Value));
	 
	    // ReactiveCommand has built-in support for background operations and
	    // guarantees that this block will only run exactly once at a time, and
	    // that the CanExecute will auto-disable and that property IsExecuting will
	    // be set according whilst it is running.
	    Search = ReactiveCommand.CreateAsyncTask(canSearch, async _ => {
	        return await searchService.Search(this.SearchQuery);
	    });
	 
	    // ReactiveCommands are themselves IObservables, whose value are the results
	    // from the async method, guaranteed to arrive on the UI thread. We're going
	    // to take the list of search results that the background operation loaded, 
	    // and them into our SearchResults.
	    Search.Subscribe(results => {
	        SearchResults.Clear();
	        SearchResults.AddRange(results);
	    });
	 
	    // ThrownExceptions is any exception thrown from the CreateAsyncTask piped
	    // to this Observable. Subscribing to this allows you to handle errors on
	    // the UI thread. 
	    Search.ThrownExceptions
	        .Subscribe(ex => {
	            UserError.Throw("Potential Network Connectivity Error", ex);
	        });
	 
	    // Whenever the Search query changes, we're going to wait for one second
	    // of "dead airtime", then automatically invoke the subscribe command.
	    this.WhenAnyValue(x => x.SearchQuery)
	        .Throttle(TimeSpan.FromSeconds(1), RxApp.MainThreadScheduler)
	        .InvokeCommand(this, x => x.Search);
	}

If the above sounds compelling then check out http://reactiveui.net/ which uses the Reactive Extensions for .NET to create elegant, testable User Interfaces that run on any mobile or desktop platform. Supports Xamarin.iOS, Xamarin.Android, Xamarin.Mac, WPF, Windows Forms, Windows Phone 8, Windows Store apps.

If you want to learn more about FRP then I suggest the academic paper "Out of the Tar Pit" by Ben Moseley & Peter Marks @ https://raw.githubusercontent.com/papers-we-love/papers-we-l... then make your way back to the origin of FRP - http://elm-lang.org/learn/What-is-FRP.elm

As for a good example of a backend designed around Event Streams/Reactive Extensions then check out https://github.com/AdaptiveConsulting/ReactiveTrader