执行多个 RxJava 调用并将结果用于下一次调用

Question

我正在尝试使用 RxJava 来解决这个问题。我有 3 个调用需要依次执行，使用最后一次调用的结果进行下一次调用。

为了简单起见，我在我的服务器上托管了 3 个文件，它们是 3 个调用：

http://jimclermonts.nl/booky/step1user

http://jimclermonts.nl/booky/step2cookie

http://jimclermonts.nl/booky/step3token

这样做的正确和最干净的方法是什么？

build.gradle：

// Retrofit
implementation 'com.squareup.retrofit2:retrofit:2.3.0'
implementation 'com.squareup.retrofit2:converter-moshi:2.3.0'
implementation 'com.squareup.okhttp3:logging-interceptor:3.9.0'
implementation 'com.squareup.okhttp3:okhttp:3.9.0'

// reactive extensions
implementation "com.squareup.retrofit2:adapter-rxjava2:2.3.0"
implementation 'io.reactivex.rxjava2:rxjava:2.1.7'
implementation 'io.reactivex.rxjava2:rxandroid:2.0.2'

MainActivity：

override fun onCreate(savedInstanceState: Bundle?) {
    super.onCreate(savedInstanceState)
    setContentView(R.layout.activity_main)

    val result = SomeWebService().getToken()
    if (result != null) {
        result.doOnComplete { 
            //how to get the token value?
        }
    }
}

SomeWebService

class SomeWebService {

    lateinit var apiService: ApiService
    var tokenResult: Observable<String>? = null

    fun getToken() : Observable<String>? {
        if (tokenResult == null) {
            apiService = ApiService.retrofit.create(ApiService::class.java)
            val body = step1ApiUserResponse()
            val cookie = step2CookieResponse(body.blockingSingle())
            val tokenResult = step3TokenResponse(cookie.blockingSingle())
            this.tokenResult = tokenResult
            tokenResult.doOnComplete {  }
    } else {
        tokenResult!!.doOnComplete {  }
    }
    return tokenResult
    }

    fun step1ApiUserResponse(): Observable<String> {
        return Observable.create {
            apiService.getStep1User()
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnSuccess { user ->
                        val body = user.getResponse()
                        if (body != null) {
                            it.onNext(body)
                        }
                    }
                    .doOnError {
                        it.printStackTrace()
                    }
        }
    }

    fun step2CookieResponse(body: String): Observable<String> {
        return Observable.create {
            apiService.getStep2Cookie(body)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnSuccess { result ->
                        val bodyResult = result.body().toString()
                        it.onNext(bodyResult)
                    }
        }
    }

    fun step3TokenResponse(cookie: String): Observable<String> {
        return Observable.create {
            apiService.getStep3Token(cookie)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnSuccess { result ->
                        val body = result.getToken()
                        if (body != null) {
                            it.onNext(body)
                        }
                    }
        }
    }
}

ApiService：

interface ApiService {

    @GET("/booky/step1user")
    fun getStep1User(): Single<UserResponse>

    @GET("/booky/step2cookie")
    fun getStep2Cookie(body: String): Single<Response>

    @GET("/booky/step3token")
    fun getStep3Token(@Header("Cookie") sessionId: String): Single<TokenResponse>

    companion object {

        val interceptor = HttpLoggingInterceptor().setLevel(HttpLoggingInterceptor.Level.BODY)
        val client = OkHttpClient.Builder()
                .addInterceptor(interceptor)
                .build()

        val retrofit = Retrofit.Builder()
                .baseUrl("http://jimclermonts.nl")
                .addConverterFactory(MoshiConverterFactory.create().asLenient())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .client(client)
                .build()
    }
}

TokenResponse：

class TokenResponse {

    @Json(name = "Token")
    private var token: String? = null

    fun getToken(): String? {
        return token
    }

    fun setToken(token: String) {
        this.token = token
    }
}

用户响应：

class UserResponse {

    @Json(name = "Response")
    private var response: String? = null

    fun getResponse(): String? {
        return response
    }

    fun setResponse(response: String) {
        this.response = response
    }
}

Answer 1

主活动：

  val service = SomeWebService()

    service.getToken()
            .subscribe(
            { token ->
                Log.d("TOKEN", token)
            },
            {e ->
                Log.e("Token error", e.localizedMessage)
            }
    )

一些网络服务：

class SomeWebService {

    lateinit var apiService: ApiService
    var tokenResult: Observable<String>? = null

    fun getToken(): Observable<String> {
        apiService = ApiService.retrofit.create(ApiService::class.java)
        return step1ApiUserResponse()
                .flatMap { body ->
                    step2CookieResponse(body)
                            .flatMap { cookie ->
                                step3TokenResponse(cookie)
                            }

                }
    }

    fun step1ApiUserResponse(): Observable<String?> {
        return apiService.getStep1User()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .map { user ->
                    Log.d("Step1", user.toString())
                    user.getResponse()
                }
                .toObservable()
    }

    fun step2CookieResponse(body: String): Observable<String?> {
        return apiService.getStep2Cookie(body)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .map { result ->
                    result.getCookie()
                }
                .toObservable()
    }

    fun step3TokenResponse(cookie: String): Observable<String?> {
        return apiService.getStep3Token(cookie)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .map { result ->
                    result.getToken()
                }
                .toObservable()
    }
}