Merge pull request #177 from adamint/bug/43-flow-block-thread

Change flow complete to suspendComplete

Author: adamint

src/commonMain/kotlin/com.adamratzman.spotify/SpotifyRestAction.kt

@@ -137,19 +137,21 @@ class SpotifyRestActionPaging<Z : Any, T : AbstractPagingObject<Z>>(api: Spotify
     /**
      * Synchronously retrieve all [AbstractPagingObject] associated with this rest action
      */
-    fun getAll() = api.tracks.toAction { complete().getAllImpl() }
+    fun getAll(context: CoroutineContext = Dispatchers.Default) = api.tracks.toAction { suspendComplete(context).getAllImpl() }
 
     /**
      * Synchronously retrieve all [Z] associated with this rest action
      */
-    fun getAllItems() = api.tracks.toAction { complete().getAllImpl().toList().map { it.items }.flatten() }
+    fun getAllItems(context: CoroutineContext = Dispatchers.Default) =
+            api.tracks.toAction { suspendComplete(context)
+            .getAllImpl().toList().map { it.items }.flatten() }
 
     /**
      * Consume each [Z] by [consumer] as it is retrieved
      */
-    fun streamAllItems(consumer: (Z) -> Unit): SpotifyRestAction<Unit> {
+    fun streamAllItems(context: CoroutineContext = Dispatchers.Default, consumer: (Z) -> Unit): SpotifyRestAction<Unit> {
         return api.tracks.toAction {
-            complete().getAllImpl().toList().forEach { it.items.forEach { item -> consumer(item) } }
+            suspendComplete(context).getAllImpl().toList().forEach { it.items.forEach { item -> consumer(item) } }
         }
     }
 
@@ -159,7 +161,7 @@ class SpotifyRestActionPaging<Z : Any, T : AbstractPagingObject<Z>>(api: Spotify
     @FlowPreview
     @JvmOverloads
     @ExperimentalCoroutinesApi
-    fun flowOrdered(context: CoroutineContext = Dispatchers.Default): Flow<Z> = flow<Z> {
+    fun flowOrdered(context: CoroutineContext = Dispatchers.Default): Flow<Z> = flow {
         emitAll(flowPagingObjectsOrdered().flatMapConcat { it.asFlow() })
     }.flowOn(context)
 
@@ -169,8 +171,8 @@ class SpotifyRestActionPaging<Z : Any, T : AbstractPagingObject<Z>>(api: Spotify
     @JvmOverloads
     @ExperimentalCoroutinesApi
     fun flowPagingObjectsOrdered(context: CoroutineContext = Dispatchers.Default): Flow<AbstractPagingObject<Z>> =
-        flow<AbstractPagingObject<Z>> {
-            complete().also { master ->
+        flow {
+            suspendComplete(context).also { master ->
                 emitAll(master.flowStartOrdered())
                 emit(master)
                 emitAll(master.flowEndOrdered())
@@ -183,7 +185,7 @@ class SpotifyRestActionPaging<Z : Any, T : AbstractPagingObject<Z>>(api: Spotify
     @FlowPreview
     @JvmOverloads
     @ExperimentalCoroutinesApi
-    fun flow(context: CoroutineContext = Dispatchers.Default): Flow<Z> = flow<Z> {
+    fun flow(context: CoroutineContext = Dispatchers.Default): Flow<Z> = flow {
         emitAll(flowPagingObjects().flatMapConcat { it.asFlow() })
     }.flowOn(context)
 
@@ -193,8 +195,8 @@ class SpotifyRestActionPaging<Z : Any, T : AbstractPagingObject<Z>>(api: Spotify
     @JvmOverloads
     @ExperimentalCoroutinesApi
     fun flowPagingObjects(context: CoroutineContext = Dispatchers.Default): Flow<AbstractPagingObject<Z>> =
-        flow<AbstractPagingObject<Z>> {
-            complete().also { master ->
+        flow {
+            suspendComplete(context).also { master ->
                 emitAll(master.flowBackward())
                 emit(master)
                 emitAll(master.flowForward())

src/commonMain/kotlin/com.adamratzman.spotify/models/PagingObjects.kt

@@ -2,11 +2,11 @@
 package com.adamratzman.spotify.models
 
 import com.adamratzman.spotify.SpotifyApi
+import com.adamratzman.spotify.SpotifyRestAction
 import com.adamratzman.spotify.http.SpotifyEndpoint
 import com.adamratzman.spotify.models.serialization.toCursorBasedPagingObject
 import com.adamratzman.spotify.models.serialization.toPagingObject
 import com.adamratzman.spotify.utils.runBlocking
-import kotlin.reflect.KClass
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.Flow
@@ -18,6 +18,8 @@ import kotlinx.coroutines.flow.toList
 import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.Transient
+import kotlin.coroutines.CoroutineContext
+import kotlin.reflect.KClass
 
 /*
     Types used in PagingObjects and CursorBasedPagingObjects:
@@ -102,6 +104,7 @@ class PagingObject<T : Any>(
             pagingObjects.add(nxt)
             nxt = nxt.next?.let { nxt?.getNext() }
         }
+
         // we don't need to reverse here, as it's in order
         return pagingObjects.asSequence()
     }
@@ -115,7 +118,8 @@ class PagingObject<T : Any>(
     /**
      * Get all items of type [T] associated with the request
      */
-    suspend fun getAllItems() = endpoint!!.toAction { getAll().complete().map { it.items }.flatten() }
+    override suspend fun getAllItems(context: CoroutineContext)
+            = endpoint!!.toAction { getAll().suspendComplete(context).map { it.items }.flatten().asSequence() }
 }
 
 /**
@@ -150,8 +154,8 @@ class CursorBasedPagingObject<T : Any>(
     /**
      * Get all items of type [T] associated with the request
      */
-    fun getAllItems() = endpoint!!.toAction {
-        getAll().complete().map { it.items }.flatten().toList()
+    override suspend fun getAllItems(context: CoroutineContext) = endpoint!!.toAction {
+        getAll().suspendComplete(context).map { it.items }.flatten().asSequence()
     }
 
     @Suppress("UNCHECKED_CAST")
@@ -219,6 +223,8 @@ abstract class AbstractPagingObject<T : Any>(
     internal abstract suspend fun getImpl(type: PagingTraversalType): AbstractPagingObject<T>?
     internal abstract suspend fun getAllImpl(): Sequence<AbstractPagingObject<T>>
 
+    internal abstract suspend fun getAllItems(context: CoroutineContext = Dispatchers.Default): SpotifyRestAction<Sequence<T>>
+
     private suspend fun getNextImpl() = getImpl(PagingTraversalType.FORWARDS)
     private suspend fun getPreviousImpl() = getImpl(PagingTraversalType.BACKWARDS)