How can I get the first asynchronous response faster (and not execute the remainder)?
Here's the setting: there is a federal remote service that returns whether a particular value is correct or incorrect. We can send requests as we like, up to 50 remote service requests.
Since we only need to use the correct value, and the set of possible values is small (~ 700), we can simply send 15 or so 50 batch requests and the correct value will be part of the result set. Thus, I used the following code:
Observable
.Range(0, requests.Count)
.Select(i => Observable.FromAsync(async () =>
{
responses.Add(await client.FederalService.VerifyAsync(requests[i]));
Console.Write(".");
}))
.Merge(8)
.Wait();
But - what I don't like about this is that if one of the early requests has the correct value, I still use all the possibilities through the service time. I am trying to make this launch as fast as possible. I know the exit condition (response code 1 to 99, any response code between 50-59 indicates the value is "correct").
Is there a way to make this code a little smarter, so we minimize the number of requests? Unfortunately, the value we are checking is evenly distributed, so sorting the queries does nothing (which I know).
source to share
You should consider using the method FirstAsync
:
The secret in our example is the method
FirstAsync
. We really expect the first result to be returned by our observables and don't care about any additional results .
So your code could be like this:
await Observable
.Range(0, requests.Count)
.Select(i => Observable.FromAsync(async () =>
{
responses.Add(await client.FederalService.VerifyAsync(requests[i]));
Console.Write(".");
}))
.FirstAsync()
.Subscribe(Console.WriteLine);
> System.Reactive.Linq.ObservableImpl.Defer`1[System.Reactive.Unit]
Rx and Await: Some notes article contains some tricks with similar methods. For example, you have an overload for FirstAsync
that can be filtered like a method LINQ
' First
:
await Observable
.Range(0, requests.Count)
.Select(i => Observable.FromAsync(async () =>
{
responses.Add(await client.FederalService.VerifyAsync(requests[i]));
Console.Write(".");
}))
.FirstAsync(r => /* do the check here */)
.Subscribe(Console.WriteLine);
source to share
You're pretty close. Change your observable to this:
Observable
.Range(0, requests.Count)
.Select(i => Observable.FromAsync(async () =>
{
var response = await Task.FromResult(i); //replace with client.FederalService.VerifyAsync(requests[i])
responses.Add(response);
Console.Write($"{i}.");
var responseCode = response; //replace with however you get the response code.
return responseCode >= 50 && responseCode <= 59;
}))
.Merge(8)
.Where(b => b)
.Take(1)
.Wait();
This way your observable keeps emitting values, so you can keep acting on it.
source to share