Использование RxJava для запросов Quickblox API

Question

Я пытаюсь использовать QuickbloxAndroid SDK в сочетании с RxJava, чтобы сделать некоторые прикованные и зависящие от времени вызовы API Quickblox. В API уже есть async методов. Вот два из них:

Создать сессию:

QBAuth.createSession(new QBEntityCallbackImpl<QBSession>() { 

      @Override 
      public void onSuccess(QBSession session, Bundle params) { 


      } 

      @Override 
      public void onError(List<String> errors) { 


      } 
     }); 

создать диалог (чат номер)

QBChatService.getInstance().getGroupChatManager().createDialog(qbDialog, 
      new QBEntityCallbackImpl<QBDialog>() { 

       @Override 
       public void onSuccess(QBDialog dialog, Bundle args) { 


       } 

       @Override 
       public void onError(List<String> errors) { 


       } 
      }); 

Как вы видели, каждый вызов API должен иметь Реализация обратного вызова как параметр метода. Проблема в том, что я должен совершать некоторые связанные вызовы API. Пример:

1. Создать пустой сеанс, затем
2. Войти, затем
3. Получить данные пользователя, затем
4. Вход в чат службу, а затем
5. ....

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

• Если API вызова с обратными вызовами быть внутри метода Observablecall()?

• Если onError() и onSuccess() методы API Quickblox называют onError() и методы Subscriber в onNext()/onCompleted()?

Может кто-нибудь сделать несколько примеров использования RxJava для создания сеанса, а затем создать диалог, используя при условии создать сеанс и создавать диалоговые методы API?

Answer 1

Вот пример того, как создать наблюдаемые с помощью Observable.create() в вашем случае:

Observable<QBSession> connectionObservable = Observable.create(new OnSubscribe<QBSession>() { 
    @Override 
    public void call(Subscriber<? super QBSession> subscriber) { 
     if (!subscriber.isUnsubscribed()) { 
      QBAuth.createSession(new QBEntityCallbackImpl<QBSession>() { 

       @Override 
       public void onSuccess(QBSession session, Bundle params) { 
        subscriber.onNext(session); 
        subscriber.onCompleted(); 
       } 

       @Override 
       public void onError(List<String> errors) { 
        subscriber.onError(/*some throwable here*/); 
       } 
      }); 
     } 
    } 
}); 

Observable<QBDialog> dialogCreationObservable = Observable.create(new OnSubscribe<QBDialog>() { 
    @Override 
    public void call(Subscriber<? super QBDialog> subscriber) { 
     if (!subscriber.isUnsubscribed()) { 
      QBChatService.getInstance().getGroupChatManager().createDialog(qbDialog, 
        new QBEntityCallbackImpl<QBDialog>() { 

         @Override 
         public void onSuccess(QBDialog dialog, Bundle args) { 
          subscriber.onNext(dialog); 
          subscriber.onCompleted(); 
         } 

         @Override 
         public void onError(List<String> errors) { 
          subscriber.onError(/*some throwable here*/); 
         } 
        }); 
     } 
    } 
});