Создание клиентского API REST с использованием Reactive Extensions (Rx)

Я пытаюсь использовать правильные варианты использования для Reactive Extensions (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", как моя предыдущая реализация.

Ответ 1

Я думаю, если вы получаете список сущностей от каждого вызова REST, ваш вызов должен иметь немного другую подпись - вы не наблюдаете каждое значение в коллекции возврата, вы наблюдаете событие завершения вызова. Поэтому для аэропортов он должен иметь подпись:

public IObservable<Aiports> GetAirports()

Следующим шагом будет запуск первых трех параллельно и ожидание на всех из них:

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

Третий шаг заключается в том, чтобы составить выше абсурдное с помощью 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());
    }

Ответ 1

Ответ 2