Làm thế nào để gọi lại chức năng async từ rx đăng ký?

Question

Tôi muốn gọi lại một chức năng không đồng bộ trong một thuê bao Rx.

Ví dụ: như thế:

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(); 
} 

Những gì cần phải được thực hiện, để nắm bắt những ngoại lệ đúng cách?

Answer 1

Bạn không muốn vượt qua một phương pháp async-Subscribe, bởi vì đó sẽ tạo ra một phương pháp async void. Làm hết sức mình để tránh async void.

Trong trường hợp của bạn, tôi nghĩ những gì bạn muốn là gọi phương thức async cho mỗi phần tử của chuỗi và sau đó lưu vào bộ nhớ cache tất cả kết quả. Trong trường hợp đó, sử dụng SelectMany để gọi phương thức async cho mỗi phần tử, và Replay để bộ nhớ cache (cộng với một Connect để có được những cán bóng):

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(); 
    } 
} 

Tôi đã thay đổi Results tài sản để được trả lại từ phương pháp Trigger thay , mà tôi nghĩ rằng ý nghĩa hơn, vì vậy kiểm tra bây giờ trông giống như:

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

Answer 2

Thay đổi này để:

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

Đăng ký không giữ hoạt động không đồng bộ bên trong Observable.

Answer 3

Paul Betts' câu trả lời làm việc trong hầu hết các tình huống, nhưng nếu bạn muốn chặn dòng trong khi chờ đợi cho các chức năng async để kết thúc bạn cần một cái gì đó như thế này:

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

Hoặc:

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