Thăm dò ý kiến ​​một webservice bằng cách sử dụng Reactive Extensions và kết quả x cuối cùng

Question

Tôi đang cố gắng sử dụng Reactive Extensions (Rx) cho một nhiệm vụ có vẻ phù hợp, bỏ phiếu tại một khoảng thời gian cụ thể một dịch vụ web và hiển thị x kết quả.

Tôi có một dịch vụ web gửi cho tôi trạng thái của một công cụ mà tôi muốn theo dõi. Tôi muốn thăm dò ý kiến ​​công cụ này với tốc độ cụ thể và hiển thị trong danh sách 20 trạng thái cuối cùng đã được thăm dò ý kiến.

Vì vậy danh sách của tôi sẽ giống như "cửa sổ chuyển động" của kết quả dịch vụ.

Tôi đang phát triển một ứng dụng WPF với Caliburn.Micro, nhưng tôi không nghĩ rằng điều này rất phù hợp.

Những gì tôi quản lý để có được đến nay là như sau (chỉ là một ứng dụng mẫu mà tôi bị hack một cách nhanh chóng, tôi sẽ không làm điều này trong các ShellViewModel trong ứng dụng thực tế):

public class ShellViewModel : Caliburn.Micro.PropertyChangedBase, IShell 
{ 
    private ObservableCollection<string> times; 
    private string currentTime; 

    public ShellViewModel() 
    { 
     times = new ObservableCollection<string>(); 

     Observable 
      .Interval(TimeSpan.FromSeconds(1)) 
      .SelectMany(x => this.GetCurrentDate().ToObservable()) 
      .ObserveOnDispatcher() 
      .Subscribe(x => 
      { 
       this.CurrentTime = x; 
       this.times.Add(x); 
      }); 
    } 

    public IEnumerable<string> Times 
    { 
     get 
     { 
      return this.times; 
     } 
    } 

    public string CurrentTime 
    { 
     get 
     { 
      return this.currentTime; 
     } 
     set 
     { 
      this.currentTime = value; 
      this.NotifyOfPropertyChange(() => this.CurrentTime); 
     } 
    } 

    private async Task<string> GetCurrentDate() 
    { 
     var client = new RestClient("http://www.timeapi.org"); 
     var request = new RestRequest("/utc/now.json"); 

     var response = await client.ExecuteGetTaskAsync(request); 

     return response.Content; 
    } 
} 

Trong xem tôi chỉ có một nhãn được gắn với thuộc tính CurrentTime và một danh sách được liên kết với thuộc tính Times.

Vấn đề tôi có là:

• Nó không chỉ giới hạn ở 20 hạng mục trong danh sách như tôi luôn thêm các mục vào ObservableCollection nhưng tôi không thể tìm thấy một cách tốt hơn để DataBind
• Khoảng thời gian không hoạt động như tôi muốn. Nếu truy vấn mất hơn 1 giây để chạy, hai truy vấn sẽ chạy song song, mà tôi không muốn xảy ra. Mục tiêu của tôi là truy vấn lặp lại vô thời hạn nhưng với tốc độ không quá 1 truy vấn mỗi giây. Nếu một truy vấn làm cho hơn 1 giây kết thúc, nó sẽ đợi nó kết thúc và kích hoạt trực tiếp truy vấn mới.

Second chỉnh sửa:

chỉnh sửa trước dưới đây được tôi là ngu ngốc và rất bối rối, nó gây nên sự kiện liên tục vì Interval là một cái gì đó liên tục mà không bao giờ kết thúc. Giải pháp của Brandon là chính xác và hoạt động như mong đợi.

Edit:

Dựa trên ví dụ Brandon, tôi đã cố gắng để làm đoạn mã sau vào LinqPad:

Observable 
    .Merge(Observable.Interval(TimeSpan.FromSeconds(2)), Observable.Interval(TimeSpan.FromSeconds(10))) 
    .Repeat() 
    .Scan(new List<double>(), (list, item) => { list.Add(item); return list; }) 
    .Subscribe(x => Console.Out.WriteLine(x)) 

Và tôi có thể thấy rằng ghi vào giao diện điều khiển xảy ra mỗi 2 giây, và không phải tất cả 10. Vì vậy, lặp lại không chờ đợi cho cả hai Observable được hoàn thành trước khi lặp đi lặp lại.

Answer 1

Hãy thử điều này:

// timer that completes after 1 second 
var intervalTimer = Observable 
    .Empty<string>() 
    .Delay(TimeSpan.FromSeconds(1)); 

// queries one time whenever subscribed 
var query = Observable.FromAsync(GetCurrentDate); 

// query + interval timer which completes 
// only after both the query and the timer 
// have expired 

var intervalQuery = Observable.Merge(query, intervalTimer); 

// Re-issue the query whenever intervalQuery completes 
var queryLoop = intervalQuery.Repeat(); 

// Keep the 20 most recent results 
// Note. Use an immutable list for this 
// https://www.nuget.org/packages/microsoft.bcl.immutable 
// otherwise you will have problems with 
// the list changing while an observer 
// is still observing it. 
var recentResults = queryLoop.Scan(
    ImmutableList.Create<string>(), // starts off empty 
    (acc, item) => 
    { 
     acc = acc.Add(item); 
     if (acc.Count > 20) 
     { 
      acc = acc.RemoveAt(0); 
     } 

     return acc; 
    }); 

// store the results 
recentResults 
    .ObserveOnDispatcher() 
    .Subscribe(items => 
    { 
     this.CurrentTime = items[0]; 
     this.RecentItems = items; 
    }); 

Answer 2

Điều này sẽ bỏ qua các tin nhắn khoảng thời gian trong khi GetCurrentDate đang được tiến hành.

Observable 
    .Interval(TimeSpan.FromSeconds(1)) 
    .GroupByUntil(p => 1,p => GetCurrentDate().ToObservable().Do(x => { 
          this.CurrentTime = x; 
          this.times.Add(x); 
         })) 
    .SelectMany(p => p.LastAsync()) 
    .Subscribe();