RxJS - Loading indicator

Question

I'm struggling to get my head around the "Rx" way of displaying a loading indicator for an AJAX stream.

$scope.$createObservableFunction("load")
        .take(1)
        .do(function(){
            $scope.loading = true;
        })
        .flatMap(contentService.AJAX_THINGY_AS_OBSERVABLE)
        .delay(300)
        .subscribe(function(content){
            console.log("content",content);
        },function(error){
            $scope.error = error
        },function() {
            $scope.loading = false;
        });

As far as I understand it I should use .do() for side effects, which I suppose setting loading is, but it doesn't feel like the right way of doing things.

Can anyone provide a cleaner/better/proper example of how to do this?

Thanks!

UPDATE 1

I decided to split this into 2 streams; requestSource and responseSource.

var loadRequestSource = $scope.$createObservableFunction("load")
    .share();

var loadResponseSource = loadRequestSource
    .flatMap(contentService.AJAX_THINGY_AS_OBSERVABLE)
    .throttle(1000)
    .share();

Then have 2 separate subscribers:

loadRequestSource.subscribe(function () {
    $scope.loading = true;
});

loadResponseSource.subscribe(function (response) {
    /* enter logic */
    $scope.loading = false;
    $scope.$digest();
}, function (err) {
    $scope.error = err;
    $scope.loading = false;
    $scope.$digest();
});

I'm liking this approach as it keeps the role of the subscribes accurate. The response subscriber doesn't need to care about setting loading to true. It only cares about setting it to false.

Answer 1

I like transforming the request/response streams into a single stream that represents the current state of your loading property:

const startLoading$ = loadRequestSource.map(() => true);
const stopLoading$ = loadResponseSource.map(() => false);
const loadingState$ = Rx.Observable.merge(startLoading$, stopLoading$);

// Finally, subscribe to the loadingState$ observable
loadingState$.subscribe(state => {
    $scope.$applyAsync(() => $scope.loading = state);
});