RxJava: Открыть сеанс, выполнить запрос и закрыть сеанс

Question

Я новичок в Rx, и я пытаюсь создать наблюдаемое что:

1. Запросов на символические
2. Создает запрос с помощью этого маркеров и выполнить запрос на услугу сторонней
3. Закрывает сессию, используя маркер

у меня есть это:

public Observable<List<City>> getCities() { 

    authService.requestToken() //return Observable<String> 
     .concatMap(new Func1<String, Observable<? extends List<City>>>() { 
      @Override 
      public Observable<? extends List<City>> call(String token) { 
       CitiesRequest request = new CitiesRequest(token); 
       return soapClient.consume(request).map(new CityResponseParseFunc());     } 
     }) 
     //Here, I need close the session using the token. Maybe with doOnCompleted? 
} 

Я не знаю, правильно ли это и не знаю, как продолжить. Обратите внимание, что мне нужно, чтобы метод возвращал Observable<List<City>>. Это возможно?

Спасибо. Любая помощь будет оценена.

Answer 1

Если requestToken() излучает только одно значение, вы можете переключиться concatMap для flatMap и использовать эту перегруженную версию этого:

authService.requestToken() //return Observable<String> 
     .flatMap(new Func1<String, Observable<? extends List<City>>>() { 
      @Override 
      public Observable<? extends List<Integer>> call(String token) { 
       CitiesRequest request = new CitiesRequest(token); 
       return soapClient.consume(request).map(new CityResponseParseFunc()); 
      } 
     }, new Func2<String, List<City>, List<City>>() { 
      @Override 
      public List<City> call(String token, List<City> cities) { 
       authService.closeSession(token) 
       return cities; /*close session with token*/ 
      } 
     }) 

Answer 2

Вы можете возможно использовать оператор using. В качестве параметра требуется 3 функции:

1. A Func0<Resource>, который создает/получает ресурс для работы (здесь жетон).
2. A Func1<Resource, Observable<T>>, который создает Observable<T>, который будет работать с ресурсом (здесь наш Наблюдаемый список городов).
3. Функция удаления, Action1<Resource>, которая будет утилизировать ресурс в конце жизненного цикла Observable<T>.

Так как ваш authService, кажется, возвращает Observable<String>, он может сделать использование в using оператора немного более сложным, особенно для функции Dispose (необходимо кэшировать ресурс Observable<String> таким образом, чтобы избавиться функция подписавшись на него не будет регенерировать токен). Другой способ - получить синхронный токен.

Что-то вроде этого:

Observable<List<City>> cities = Observable.using(
    //resource creation function 
    () -> authService.requestToken().toBlocking().single(), 
    //observable creation function 
    token -> soapClient.consume(new CitiesRequest(token)).map(new CitiesResponseParseFunc())), 
    //resource disposal function 
    token -> authService.closeSession(token) 
);