Создание клиентского API REST с использованием реактивных расширений (Rx)

Question

Я пытаюсь настроить правильные варианты использования для реактивных расширений (Rx). Примерами, которые продолжают расти, являются события пользовательского интерфейса (перетаскивание, рисование) и предложения о том, что Rx подходит для асинхронных приложений/операций, таких как вызовы веб-сервисов.

Я работаю над приложением, где мне нужно написать крошечный клиентский API для службы REST. Мне нужно назвать четыре конечных пункта 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 Dictionary), чтобы я мог легко установить ссылку при разборе каждого рейса.

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

Третий шаг woul быть составляете выше abservable с GetFlights():

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

EDIT: Я до сих пор не понимаю, почему ваши услуги вернуться

IObservable<Airport> 

вместо

IObservable<IEnumerable<Airport>> 

AFAIK, от 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 будет содержать одно событие на временной шкале, которая будет содержать все справочные данные.

EDIT: Вот еще один, используя недавно отчеканенных 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 хорошо. Если вы хотите сказать, сообщите, где находится новый рейс, а затем оберните вызов REST на основе тяги в Observable.Generate может иметь какое-то значение.