https://kotlinlang.org logo
Join Slack
Powered by
Hi again :smile: Is there a way to convert from `S...
# reaktive
o
Hi again 😄 Is there a way to convert from 
Single
to a suspended function? I couldn't find anything in 
coroutines-interop
module. Currently I'm doing
Copy code
val body = suspendCoroutine<String?> { continuation ->
    get("<https://postman-echo.com/get>", headers, query)
        .subscribe(
            onSuccess = { continuation.resume(it) }, 
            onError = { continuation.resumeWithException(it) } 
        )
}
It works in JVM and JS, but in Native I get
Copy code
kotlin.native.concurrent.FreezingException: freezing of $subscribe_2$<anonymous>_67$FUNCTION_REFERENCE$1621@e3e09f38 has failed, first blocker is EventLoopImpl@1c250dc8
a
No there is none. I did not expect any use cases for this. I would do somehting like this:
Copy code
suspend fun <T> Single<T>.suspendingGet(): T =
    suspendCancellableCoroutine { continuation ->
        val disposableWrapper = DisposableWrapper()
        continuation.invokeOnCancellation { disposableWrapper.dispose() }
        subscribe(
            object : SingleObserver<T> {
                override fun onSubscribe(disposable: Disposable) {
                    disposableWrapper.set(disposable)
                }

                override fun onSuccess(value: T) {
                    continuation.resume(value)
                }

                override fun onError(error: Throwable) {
                    continuation.resumeWithException(error)
                }
            }
        )
    }
About the execption. I suppose the 
get(...)
is executed in background and so freezes the observer. AFAIK Continuation (as well as other things like Job, etc.) can not be frozen currently.
o
Awesome, this fixes the disposable issue I had in my solution
The use case I found for this is, while testing, you can return a suspended function (with a fix to a promise in JS as in https://youtrack.jetbrains.com/issue/KT-22228) and the test runner will wait for the completion before finishing the test case, avoid having to 
blockingGet
i.e
Copy code
@Test
fun test_suspendingGet() = runAsyncTest {
    val value = singleOf(1)
        .delay(300, computationScheduler) // simulate a long running op
        .suspendingGet()
    
    assertEquals(1, value)
}
a
But you can test Rx chains without coroutines interop!
Just use TestScheduler. It provides a way to advance time.
Here is a sample for your case:
Copy code
class MyLogic {

    fun doSomething(): Single<Int> =
        singleOf(1)
            .delay(300L, computationScheduler)
}

class MyLogicTest {

    private val logic = MyLogic()
    private val testComputationScheduler = TestScheduler()
    private val computationTimer = testComputationScheduler.timer

    @BeforeTest
    fun before() {
        overrideSchedulers(computation = { testComputationScheduler })
    }

    @AfterTest
    fun after() {
        overrideSchedulers()
    }

    @Test
    fun doSomething_returns_1_after_300_millis() {
        val observer = logic.doSomething().test()
        computationTimer.advanceBy(300L)

        observer.assertSuccess(1)
    }
    
    @Test
    fun doSomething_does_not_success_before_300_millis() {
        val observer = logic.doSomething().test()
        computationTimer.advanceBy(299L)

        observer.assertNotSuccess()
    }
}
Also sometimes it is useful to pass schedulers via constructor with a good parameter name:
Copy code
class MyLogic(
    private val delayScheduler: Scheduler
) {

    fun doSomething(): Single<Int> =
        singleOf(1)
            .delay(300L, delayScheduler)
}
Hope this helps
o
Hi, yes, usually I use TestScheduler, even in the same way (passed in the constructor) but this is more like an Integration test, I'm testing real HTTP request
a
Tests with concurrency are usually flaky. I prefer to abstract low level networking with an interface, fake it and do integration testing with TestSchedulers.
o
I knew the example would be confusing 🤦, anyway, I can't use TestScheduler and I couldn't find a way make the test case wait for the 
Single
to complete,
I'm not testing my Business Logic, I'm testing a network layer I'm writing to stop using Ktor
Copy code
expect fun request(method: HttpMethod, url: String, headers: Headers, query: Query, body: RequestBody?): Single<ResponseBody>
This works like a charm
Copy code
@Test
    fun get_request() = runBlockingTest {
        val body = withContext(Default) { get("<https://httpbin.org/get>", headers, query).suspendingGet() }
        val echo = json.parse(PostmanEcho.serializer(), body)

        checkHeaders(headers, echo)
        checkQuery(query, echo)
    }
a
Ok, maybe this makes sense. I would not test simple low level network requests.
Like I would create my own httpclient without tests. As there would be no logic inside.
Everything else would be using a fake
o
Yes. that's exactly what I'm doing, but even tho, I wanted to test the Http client
a
Ok then) understandable
o
😌
Also that test works in the three platforms, and it doesn't depend on coroutines, just for testing
a
I would like to see how such tests would look like) the would execute real requests
o
Yeah, sadly OkHttp 
MockWebServer
only works on JVM
a
Or similar
This is the very low level I would not test
So it does direct requests without any logic
o
Almost like that
Copy code
actual fun request(method: HttpMethod, url: String, headers: Headers, query: Query, body: RequestBody?): Single<ResponseBody> =
    single { emitter ->
        val urlComponent = NSURLComponents(string = url).apply {
            queryItems = query.entries.map { (key, value) -> NSURLQueryItem(key, value) }
        }

        val request = NSMutableURLRequest.requestWithURL(urlComponent.URL!!).apply {
            setHTTPMethod(method.name)
            setAllHTTPHeaderFields(headers as Map<Any?, String>)
            if (body != null) setHTTPBody((body as NSString).dataUsingEncoding(NSUTF8StringEncoding))

            setCachePolicy(NSURLRequestReloadIgnoringCacheData)
        }

        val session = NSURLSession.sessionWithConfiguration(
            configuration = NSURLSessionConfiguration.defaultSessionConfiguration(),
            delegate = object : NSObject(), NSURLSessionDelegateProtocol {
                override fun URLSession(
                    session: NSURLSession,
                    didReceiveChallenge: NSURLAuthenticationChallenge,
                    completionHandler: (NSURLSessionAuthChallengeDisposition, NSURLCredential?) -> Unit
                ) {
                    if (didReceiveChallenge.protectionSpace.host in trustedHosts) {
                        completionHandler(
                            NSURLSessionAuthChallengeUseCredential,
                            NSURLCredential.credentialForTrust(didReceiveChallenge.protectionSpace.serverTrust)
                        )
                    } else {
                        completionHandler(
                            NSURLSessionAuthChallengePerformDefaultHandling,
                            null
                        )
                    }
                }
            },
            delegateQueue = NSOperationQueue.currentQueue
        )

        val callback: (NSData?, NSURLResponse?, NSError?) -> Unit = { data, _, error ->
            if (data != null) emitter.onSuccess(NSString.create(data, NSUTF8StringEncoding) as String)
            else emitter.onError(RequestError(error?.localizedDescription ?: "Unknown error"))
        }

        val task = session.dataTaskWithRequest(request, callback.freeze())

        emitter.setDisposable(Disposable { task.cancel() })
        task.resume()
    }
I'm using 
<http://httpbin.org|httpbin.org>
to testing
Again, more like an Integration tests
a
Yep, looks legit)
o
I also have it for JS using 
fetch
and JSVM using 
OkHttp
a
Cool, should you have any questions, ask)
o
Aweome, I'll, thanks again
a
I still not 100% sure why do you need coroutines though.
o
I was using 
blockingGet
but it doesn't work in JS
The other solution is to return a suspended function and the test runner will wait for it to complete, instead of finishing immediately
AFAIK
a
So suspendingGet processes the event loop somehow. It should be possible to do it without coroutines.
o
You're right, even in iOS I don't need to run the looper
I'm using coroutines only to return it, as in
Copy code
@Test
fun get_request() = runBlockingTest {
    ...
}
👍 1
a
But anyway, if it works then good!
How your 
request
looks like in JS?
o
Yes 🤞
Copy code
actual fun request(method: HttpMethod, url: String, headers: Headers, query: Query, body: RequestBody?): Single<ResponseBody> =
    single { emitter ->
        val queryStr = query.entries.joinToString("&") { (name, value) -> "$name=${encodeURIComponent(value)}" }
        val urlWithQuery = "$url?$queryStr"
        val headersObj = headers.entries.fold(js("({})")) { o, (name, value) -> o[name] = value; o }

        val req = if (body != null) requestInit(method = method.name, headers = headersObj, body = body)
        else requestInit(method = method.name, headers = headersObj)

        window.fetch(urlWithQuery, req)
            .then { it.text() }
            .then { emitter.onSuccess(it) }
            .catch { emitter.onError(it) }
    }
a
Thanks! Will check)
A bit later
o
For sure, thank you
This is also needed:
Copy code
internal fun requestInit(
    method: String? = undefined,
    headers: dynamic = undefined,
    body: dynamic = undefined,
    referrer: String? = undefined,
    referrerPolicy: dynamic = undefined,
    mode: RequestMode? = undefined,
    credentials: RequestCredentials? = undefined,
    cache: RequestCache? = undefined,
    redirect: RequestRedirect? = undefined,
    integrity: String? = undefined,
    keepalive: Boolean? = undefined,
    window: Any? = undefined
): RequestInit {
    val o = js("({})")

    o["method"] = method
    o["headers"] = headers
    o["body"] = body
    o["referrer"] = referrer
    o["referrerPolicy"] = referrerPolicy
    o["mode"] = mode
    o["credentials"] = credentials
    o["cache"] = cache
    o["redirect"] = redirect
    o["integrity"] = integrity
    o["keepalive"] = keepalive
    o["window"] = window

    return o
}

external fun encodeURIComponent(decodedURI: String): String
For now I'm keeping it simple
Copy code
enum class HttpMethod { GET, HEAD, POST, PUT, DELETE, CONNECT, OPTIONS, TRACE, PATCH }

typealias Headers = Map<String, String>
typealias Query = Map<String, String>
typealias RequestBody = String
typealias ResponseBody = String
a
As far as I remember in JS runBlockingTest returns Promise. And the framework waits for completion. So this should be the trick. You could achieve the same by converting Single to Promise in JS. There is interop available in Reaktive.
o
Yes, it returns a Promise (patched)
Copy code
actual fun runBlockingTest(block: suspend () -> Unit): dynamic = GlobalScope.promise { block() }
a
So some sort of expect/actual that does blockingGet in non JS and Single.toPromise() in js
E.g. 
expect fun reaktiveTest(block: () -> Single<*>):  Any
o
🤔 But then how I deal with the 
Promise
in a way that is transparent (this is exactly what coroutines are doing)
a
I thought you should just return the result of this function from your test method.
o
Ummm, I think I follow you, let me try something
a
And in actual JS impl convert Single to Promise
o
smart
a
If it works I will add it to reaktive testing utilis
o
Ok, I'll keep you posted
a
Seems working! Just checked. Will do a PR. But there is actually no Single.toPromise convertsion. Onle the opposite one. So this is to be added first.
o
😓 I could't find it in the interop module
Can you share how the use case ended?
a
You won't find, it does not exist. Only Promise.toSingle does, in the Reaktive module itself.
Will share soon
o
Awesome
a
So the API is not 100% final, if you have any concerns please let me know.
Copy code
fun <T> Single<T>.asPromise(): Promise<T> =
    Promise { resolve, reject ->
        subscribe(onSuccess = resolve, onError = reject)
    }
Copy code
expect fun Completable.testAwait()

class Tst {

    @Test
    fun foo() =
        singleOf(1)
            .observeOn(mainScheduler)
            .asCompletable()
            .testAwait()
}
JVM + Native
Copy code
actual fun Completable.testAwait(): Unit = blockingAwait()
JS
Copy code
actual fun Completable.testAwait(): dynamic = 
    asSingle(Unit).asPromise()
o
it looks great, and about who to assert the value inside the Single, with 
doOn..
should be enought?
Concretely 
doOnAfterSuccess
a
Nope, doOnAfterSuccess is too late. doOnBeforeSuccess should work. let me check
Copy code
class Tst {

    @Test
    fun foo() =
        singleOf(1)
            .observeOn(mainScheduler)
            .doOnBeforeSuccess {
                assertEquals(2, it)
            }
            .asCompletable()
            .testAwait()
}
This works
o
Awesome!!!
a
Cool. Will do a PR later 🙂
o
Thanks again
🎉 2
5 Views
Open in Slack
PreviousNext
Powered by