Can I repeat it once based on the condition?

Question

I have a method that works, looks something like this:

public IObservable<List<Stuff>> GetGoodStuff()
{
    return Observable.FromAsync(GetAccessTokenAsync)
        .SelectMany(accessToken =>
        {
            return httpClient.SendAsync(request);
        })
        .SelectMany(response => 
        { 
            response.EnsureSuccessStatusCode(); 
            return response.Content.ReadAsStringAsync(); 
        })
        .Select(json => 
        {
            return JsonConvert.DeserializeObject<List<Stuff>>(json);
        });
}

"GetAccessTokenAsync" returns the cached access token for the api, or, for the first time, you will receive a token. The rest is pretty standard in terms of httpclient and Rx.

Here's what: I would like to catch error 401, update the access token, and then repeat everything. But only once - after that he can throw an exception to the caller.

In this middle block, I can do this:

            if (response.StatusCode == HttpStatusCode.Unauthorized)
            {
                InvalidateAccessToken();
                // what now???
            }

But then what? Not seeing how the recursive call will work. How to wrap it all up? Not seeing this yet ...

EDIT 1 - 7Nov2015

. , , "", .

, @Timothy Shields, , (, , , : -)

/// <summary>
/// Makes an httpclient request using the access token. If Unauthorized is received the access
/// token will be reacquired and the request will be retried once.
/// </summary>
/// <returns>The json result from a successful request.</returns>
async Task<string> MakeRequestWithAccessToken(string requestUri, CancellationToken cancellationToken)
{
    const int RetryCount = 1;

    HttpResponseMessage response = null;
    for (int i = 0; i <= RetryCount; i++)
    {
        var accessToken = await GetAccessTokenAsync();

        var request = new HttpRequestMessage(HttpMethod.Get, requestUri);
        request.Headers.Add("Authorization", "Bearer " + accessToken);

        var client = new RemoteService(ApiUrl).NewClient();

        response = await client.SendAsync(request, cancellationToken);
        if (i < RetryCount && response.StatusCode == HttpStatusCode.Unauthorized)
        {
            InvalidateAccessToken();
            continue;
        }

        response.EnsureSuccessStatusCode();
    }

    return await response.Content.ReadAsStringAsync();
}

public IObservable<List<Stuff>> GetGoodStuff(int maxCount)
{
    return Observable.FromAsync(async cancellationToken =>
    {
        var requestUri = string.Format("mypath.json?count={0}", maxCount);
        var json = await MakeRequestWithAccessToken(requestUri, cancellationToken);
        return JsonConvert.DeserializeObject<List<Stuff>>(json);
    });
}

Answer 1

async - await:

public IObservable<List<Stuff>> GetGoodStuff()
{
    return Observable.FromAsync(async cancellationToken =>
    {
        const int RetryCount = 1;
        for (int i = 0; i <= RetryCount; i++)
        {
            var accessToken = await GetAccessTokenAsync();
            var request = MakeRequest(accessToken);
            var response = await httpClient.SendAsync(request, cancellationToken);
            if (i < RetryCount && response.StatusCode == HttpStatusCode.Unauthorized)
            {
                InvalidateAccessToken();
                continue;
            }
            response.EnsureSuccessStatusCode();
            var json = await response.Content.ReadAsStringAsync(cancellationToken);
            return JsonConvert.DeserializeObject<List<Stuff>>(json);
        }
    });
}

, IObservable<T>.

, , "". , InvalidateAccessToken(), MakeRequest. , , .

Answer 2

, InvalidateAccessToken , GetAccessTokenAsync.

, , . , , .

public IObservable<List<Stuff>> GetGoodStuff()
{
    var invalidate = Observable.FromAsync(InvalidateAccessTokenAsync)
                .Select(x => Observable.Throw<string>(new Exception()))
                .Switch()
                .Replay()
                .RefCount();

    return Observable.FromAsync(GetAccessTokenAsync)
        .SelectMany(accessToken =>
        {
            return httpClient.SendAsync(request);
        })
        .SelectMany(response => 
        { 
            if (response.StatusCode == HttpStatusCode.Unauthorized)
            {
                return invalidate;
            }

            response.EnsureSuccessStatusCode(); 
            return response.Content.ReadAsStringAsync().ToObservable(); 
        })
        .Select(json => 
        {
            return JsonConvert.DeserializeObject<List<Stuff>>(json);
        })
        .Retry(1);
}

: @supertopi

Select invalidate IOberservable<IOberservable<string>>. , Switch .

Replay IConnectableObservable<string>, . IConnectableObservable<T> Observable.FromAsync(InvalidateAccessTokenAsync) . , . introintorx .

RefCount connect IConnectableObservable. RefCount IObservable<string>. .