Creando una API de cliente REST usando Reactive Extensions (Rx)

Question

Estoy tratando de entender los casos de uso correctos para las extensiones reactivas (Rx). Los ejemplos que siguen surgiendo son los eventos UI (arrastrar y soltar, el dibujo) y las sugerencias de que Rx es adecuado para aplicaciones/operaciones asíncronas, como las llamadas al servicio web.

Estoy trabajando en una aplicación donde tengo que escribir una pequeña API cliente para un servicio REST. Necesito llamar cuatro puntos finales REST, tres para obtener algunos datos de referencia (Aeropuertos, Aerolíneas y Estados), y el cuarto es el servicio principal que le dará los horarios de vuelo para un aeropuerto determinado.

He creado clases que exponen los tres servicios de datos de referencia y los métodos ser algo como esto:

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

En mi método GetFlights Quiero que cada vuelo para almacenar una referencia al aeropuerto se salgan de, y la Aerolínea operando el vuelo. Para hacer eso, necesito que los datos de GetAirports y GetAirlines estén disponibles. Cada aeropuerto, línea aérea y estado se agregarán a un Dictionar (es decir, un diccionario) para que pueda establecer fácilmente la referencia al analizar cada vuelo.

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

Mi implementación actual ahora se ve así:

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

La implementación actual se basa en la respuesta de Sergey, y utiliza ForkJoin para asegurar la ejecución secuencial y que me refiero a los datos se carga antes de los vuelos. Esta implementación es mucho más elegante que tener que iniciar un evento "ReferenceDataLoaded" como mi implementación anterior.

Answer 1

Creo que si recibe una lista de entidades de cada llamada REST, su llamada debe tener una firma diferente: no está observando cada valor en la colección de devolución, está observando el evento de la finalización de la llamada. Así que para los aeropuertos, se debe tener la firma:

public IObservable<Aiports> GetAirports() 

El siguiente paso sería correr tres primeros en paralelo y esperar en todos ellos:

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

El tercer paso woul ser para componer la por encima de abservable con los GetFlights():

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

EDIT: yo todavía no entiendo por qué sus servicios vuelven

IObservable<Airport> 

en lugar de

IObservable<IEnumerable<Airport>> 

yo sepa, del resto de llamar a obtener todas las entidades a la vez - pero tal vez lo hace de paginación? De todos modos, si desea hacer el búfer RX se puede utilizar .BufferWithCount():

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

continuación, puede aplicar ForkJoin:

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

ports_lines_statuses contendrían un solo evento en la línea de tiempo que contendría todos los datos de referencia.

EDIT: Aquí hay otro, utilizando el ListObservable de nuevo cuño (última versión solamente):

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

El caso de uso que aquí se basa Pull - IEnumerable está muy bien. Si quiere decir, notifique dónde entra un nuevo vuelo, entonces envolver una llamada REST de extracción dentro de Observable.Generate puede ser de algún valor.