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

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

12 голосов
/ 14 мая 2010

Я пытаюсь обдумать правильные варианты использования 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. Каждый аэропорт, авиакомпания и статус будут добавлены в словарь (т. Е. Словарь), чтобы я мог легко установить эталон при разборе каждого рейса. 

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», как в моей предыдущей реализации.

Ответы [ 2 ]

2 голосов
/ 14 мая 2010

Я думаю, что если вы получаете список объектов от каждого вызова 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;

РЕДАКТИРОВАТЬ: Я до сих пор не понимаю, почему ваши услуги возвращаются 

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

РЕДАКТИРОВАТЬ: Вот еще один, используя недавно отчеканенный 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());
    }
0 голосов
/ 15 мая 2010

Вариант использования в данном случае основан на pull - IEnumerable в порядке. Если вы хотите сказать, уведомите, где поступает новый рейс, тогда завершение вызова REST на основе извлечения в Observable.Generate может иметь некоторое значение.

...