cref - remarks c#

¿Cómo recuperar la función asíncrona desde rx subscribe?

Question

Me gustaría recuperar una función asíncrona dentro de una suscripción Rx.

Por ejemplo, así:

public class Consumer
{
private readonly Service _service = new Service();
public ReplaySubject<string> Results = new ReplaySubject<string>();
public void Trigger()
{
Observable.Timer(TimeSpan.FromMilliseconds(100)).Subscribe(async _ => await RunAsync());
}
public Task RunAsync()
{
return _service.DoAsync();
}
}
public class Service
{
public async Task<string> DoAsync()
{
return await Task.Run(() => Do());
}
private static string Do()
{
Thread.Sleep(TimeSpan.FromMilliseconds(200));
throw new ArgumentException("invalid!");
return "foobar";
}
}
[Test]
public async Task Test()
{
var sut = new Consumer();
sut.Trigger();
var result = await sut.Results.FirstAsync();
}

¿Qué se necesita hacer para detectar la excepción correctamente?

Answer 1

Cambiar esto a:

Observable.Timer(TimeSpan.FromMilliseconds(100))
.SelectMany(async _ => await RunAsync())
.Subscribe();

Suscribirse no mantiene la operación asincrónica dentro de Observable.

Answer 2

No desea pasar un método async a Subscribe , porque eso creará un método de async void . Haz tu mejor async void para evitar el async void .

En su caso, creo que lo que quiere es llamar al método async para cada elemento de la secuencia y luego almacenar en caché todos los resultados. En ese caso, use SelectMany para llamar al método async para cada elemento y Replay en caché (más un Connect para poner en marcha la bola):

public class Consumer
{
private readonly Service _service = new Service();
public IObservable<string> Trigger()
{
var connectible = Observable.Timer(TimeSpan.FromMilliseconds(100))
.SelectMany(_ => RunAsync())
.Replay();
connectible.Connect();
return connectible;
}
public Task<string> RunAsync()
{
return _service.DoAsync();
}
}

Cambié la propiedad Results que se devolverá del método Trigger lugar, que creo que tiene más sentido, por lo que ahora la prueba se ve así:

[Test]
public async Task Test()
{
var sut = new Consumer();
var results = sut.Trigger();
var result = await results.FirstAsync();
}

Answer 3

La respuesta de Paul Betts funciona en la mayoría de los escenarios, pero si quieres bloquear la transmisión mientras esperas que termine la función asíncrona, necesitas algo como esto:

Observable.Interval(TimeSpan.FromSeconds(1))
.Select(l => Observable.FromAsync(asyncMethod))
.Concat()
.Subscribe();

O:

Observable.Interval(TimeSpan.FromSeconds(1))
.Select(_ => Observable.Defer(() => asyncMethod().ToObservable()))
.Concat()
.Subscribe();