使用反應式擴展（Rx）創建REST客戶端API

Question

我試圖讓我的頭圍繞反應式擴展（Rx）的正確用例。不斷出現的示例是UI事件（拖放和繪圖），以及Rx適用於異步應用程序/操作（如Web服務調用）的建議。

我正在研究一個應用程序，我需要爲REST服務編寫一個小型客戶端API。我需要撥打四個REST端點，三個獲取一些參考數據（機場，航空公司和狀態），第四個是主要服務，可以爲您提供給定機場的航班時間。

我已經創建的類暴露了三個基準數據服務，並且這些方法看起來是這樣的：

public Observable<IEnumerable<Airport>> GetAirports() 
public Observable<IEnumerable<Airline>> GetAirlines() 
public Observable<IEnumerable<Status>> GetStatuses() 
public Observable<IEnumerable<Flights>> GetFlights(string airport) 

在我GetFlights方法我希望每個航班召開參考機場它的背離，以及航空公司經營航班。爲此，我需要GetAirports和GetAirlines中的數據可用。每個機場，航空公司和狀態都將被添加到Dictionar（ie.e詞典）中，以便在解析每個航班時輕鬆設置參考。

flight.Airport = _airports[flightNode.Attribute("airport").Value] 
flight.Airline = _airlines[flightNode.Attribute("airline").Value] 
flight.Status = _statuses[flightNode.Attribute("status").Value] 

我目前的執行情況，現在看起來是這樣的：

public IObservable<IEnumerable<Flight>> GetFlightsFrom(Airport fromAirport) 
{ 
    var airports = new AirportNamesService().GetAirports(); 
    var airlines = new AirlineNamesService().GetAirlines(); 
    var statuses = new StatusService().GetStautses(); 


    var referenceData = airports 
     .ForkJoin(airlines, (allAirports, allAirlines) => 
          { 
           Airports.AddRange(allAirports); 
           Airlines.AddRange(allAirlines); 
           return new Unit(); 
          }) 
     .ForkJoin(statuses, (nothing, allStatuses) => 
          { 
           Statuses.AddRange(allStatuses); 
           return new Unit(); 
          }); 

    string url = string.Format(_serviceUrl, 1, 7, fromAirport.Code); 

    var flights = from data in referenceData 
        from flight in GetFlightsFrom(url) 
        select flight; 

    return flights; 
} 

private IObservable<IEnumerable<Flight>> GetFlightsFrom(string url) 
{ 
    return WebRequestFactory.GetData(new Uri(url), ParseFlightsXml); 
} 

當前的實現是基於謝爾蓋的回答，並使用ForkJoin保證順序執行，而且我引用數據獲取航班之前加載。這個實現比開發像之前實現的「ReferenceDataLoaded」事件更優雅。

Answer 1

我認爲，如果您收到來自每個REST呼叫的實體列表，您的呼叫應該有一個不同的簽名 - 您沒有觀察返回集合中的每個值，您正在觀察呼叫完成事件。因此，對於機場，它應該有簽名：

public IObservable<Aiports> GetAirports() 

下一個步驟將是並行運行前三並等待所有的人：

var ports_lines_statuses = 
    Observable.ForkJoin(GetAirports(), GetAirlines(), GetStatuses()); 

第三步竟被是組成上述abservable與GetFlights（）：

var decoratedFlights = 
    from pls in ports_lines_statuses 
    let airport = MyAirportFunc(pls) 
    from flight in GetFlights(airport) 
    select flight; 

編輯：我還是不明白，爲什麼你的服務回報

IObservable<Airport> 

代替

IObservable<IEnumerable<Airport>> 

據我所知，從REST調用你的所有實體一次 - 但也許你做分頁？ 無論如何，如果你想要做的RX你可以使用.BufferWithCount（）的緩衝：

var allAirports = new AirportNamesService() 
     .GetAirports().BufferWithCount(int.MaxValue); 
... 

那麼你可以申請ForkJoin：

var ports_lines_statuses = 
    allAirports 
     .ForkJoin(allAirlines, PortsLinesSelector) 
     .ForkJoin(statuses, ... 

ports_lines_statuses將包含在時間軸上的單個事件，這將包含所有參考數據。

編輯：這裏是另外一個使用新出爐的ListObservable（最新版本只）：

allAiports = airports.Start(); 
allAirlines = airlines.Start(); 
allStatuses = statuses.Start(); 

... 
whenReferenceDataLoaded = 
    Observable.Join(airports.WhenCompleted() 
       .And(airlines.WhenCompleted()) 
       .And(statuses.WhenCompleted()) 
       Then((p, l, s) => new Unit())); 



    public static IObservable<Unit> WhenCompleted<T>(this IObservable<T> source) 
    { 
     return source 
      .Materialize() 
      .Where(n => n.Kind == NotificationKind.OnCompleted) 
      .Select(_ => new Unit()); 
    } 

Answer 2

用例這裏基於拉 - IEnumerable的罰款。如果你想說，通知新航班進入的地方，然後在Observable.Generate內包裝一個基於拉的REST調用可能有一定的價值。