Heya all, I’ve got one beginner Rx question (sorry...

# rx

[antoniomarin]

Heya all, I’ve got one beginner Rx question (sorry about it) which really confuses me… I’ve got reactive Completable flow for login, and at some point in the middle I need to call AppAuth library to performTokenRequest. That library uses callbacks for results, so I’m curious how to adapt it to my architecture? Idea is that when callback is successful I resume with doOnSuccess or doNext to the next Completable function.

private fun login(
authorisationResponse: AuthorisationResponse
): Completable {
  return Completable.defer {
	// do some validations..
	if (authorisationResponse == null) {
	    return@defer Completable.error(...)
	}

	// app auth stuff
	val tokenRequest = authorizationResponse.createTokenExchangeRequest()
	val clientAuthentication ....

	// performing token request, AppAuth library works with callbacks, so I'm
	// curious how to "pause" my Rx flow until I get result from that callback
	// and then I need to continue completable

	return@defer ??? 
	  authorisationService.performTokenRequest(tokenRequest,clientAuthentication)
	{ response, ex -> 
		// callbacks
		if (response != null) {
			//... I'm going to save token here
            authStateManager.saveToken(response)
		}

	.doOnSuccess or .doNext (continueLoginFlow)
  }
}

private fun continueLoginFlow(): Completable {
    return authStateManager.getToken()
           .map { ... }
           .flatMap { ... }
           .doOnSuccess { ... } 
           ...

	...
}

[gildor]

Completable.defer will not help in this case, defer is useful just for lazy creation of Completables from existing result.
You need Completable.create, see doc, it will give you emitter which you should notify from your callback, it allows wrap non-rx async code to rx primitives (Completable in your case)

[gildor]

But looking into your code it looks that single is probably a better option, you should use it to wrap callback and return response from it

[gildor]

It would be more clear solution which doesn't mix wrapping callback and what you do with result of this callback

[antoniomarin]

Hi @gildor thanks for looking into it, could you please explain how to wrap callback result into Single? Sorry for my lack of knowledge, further Rx research is next on my learning Todo…

I know this doesn’t make sense but it compiles 😄

return@defer Single.just(
     authorizationService.performTokenRequest(tokenRequest, clientAuthentication) { response, ex ->
         if (response != null) {
              authStateManager.updateAfterTokenResponse(response, null)
          }
       }).doOnSuccess {
           continueLoginFlow()
       }.ignoreElement()

[antoniomarin]

Oke I got it, thanks for your help 🙂

val single: Single<TokenResponse> = Single.create { emitter ->
	authorizationService.performTokenRequest(
		tokenRequest,
		clientAuthentication
	) { response, ex ->
		if (response != null) {
			emitter.onSuccess(response)
		} else {
			emitter.onError("Errror")
		}
	}
}
		
return@defer single.doOnSuccess...

[gildor]

Yes, it true for all Rx primitives, use create method

[gildor]

emitter.onError("Errror")

Probably it should be

emitter.onError(ex)

[gildor]

I would also say that it usually more convinient (especially in Kotlin) just create Rx version of callback:

fun AuthorizationService.tokenRequestSingle(
    tokenRequest: ???,
    clientAuthentication: ???
): Single<TokenResponse> = Single.create {
   ...
}

So you getting actual Rx version of this callback, it just becomes API of this service

[gildor]

also create supports cancellation, you should implement it if this callback API support it too

[antoniomarin]

I’m not sure, I’m implementing AppAuth library, so I don’t wanna do too much coupling
https://github.com/openid/AppAuth-Android/blob/master/library/java/net/openid/appauth/AuthorizationService.java

[gildor]

I don’t see any coupling with my example, it just an extension function, it can be private in class where you use it

[gildor]

it just looks and works better and more natural than local variable, like in your case above

[antoniomarin]

Oke, thanks for your suggestions 🙂
I’m doing some refactoring of legacy code, that’s why I’m a bit confused here, and doing it step by step

[gildor]

ehh, their implementation of service is really not optimal, no cancellation, everything based on AsyncTasks, oh gosh

[gildor]

I actually looked into it recently, but just decided to use own simple ad hoc implementation, fortunately it’s for our own API and for limited subset of openid

[antoniomarin]

We also had slightly modified implementation of it, but since the guy who wrote it isn’t anymore around I’ve decided to go stock, it’s simpler for other team members and maintenance wise…

[antoniomarin]

We do newer stuff in Coroutines and now looking at legacy Rx feels so strange :D