Merge pull request #905 from sproctor/io-dispatcher

Use Dispatchers.IO by default on multi-threaded platforms

Author: jan-tennert

Auth/src/commonMain/kotlin/io/github/jan/supabase/auth/AuthConfig.kt

@@ -5,7 +5,6 @@ import io.github.jan.supabase.SupabaseSerializer
 import io.github.jan.supabase.plugins.CustomSerializationConfig
 import io.github.jan.supabase.plugins.MainConfig
 import kotlinx.coroutines.CoroutineDispatcher
-import kotlinx.coroutines.Dispatchers
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.seconds
 
@@ -52,7 +51,8 @@ open class AuthConfigDefaults : MainConfig() {
     /**
      * The dispatcher used for all auth related network requests
      */
-    var coroutineDispatcher: CoroutineDispatcher = Dispatchers.Default
+    @Deprecated("SupabaseClientBuilder.coroutineDispatcher should be used instead")
+    var coroutineDispatcher: CoroutineDispatcher? = null
 
     /**
      * The type of login flow to use. Defaults to [FlowType.IMPLICIT]

Auth/src/commonMain/kotlin/io/github/jan/supabase/auth/AuthImpl.kt

@@ -40,6 +40,7 @@ import io.ktor.http.HttpMethod
 import io.ktor.http.HttpStatusCode
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.ensureActive
@@ -72,7 +73,7 @@ internal class AuthImpl(
 
     private val _sessionStatus = MutableStateFlow<SessionStatus>(SessionStatus.Initializing)
     override val sessionStatus: StateFlow<SessionStatus> = _sessionStatus.asStateFlow()
-    internal val authScope = CoroutineScope(config.coroutineDispatcher)
+    internal val authScope = CoroutineScope((config.coroutineDispatcher ?: supabaseClient.coroutineDispatcher) + SupervisorJob())
     override val sessionManager = config.sessionManager ?: createDefaultSessionManager()
     override val codeVerifierCache = config.codeVerifierCache ?: createDefaultCodeVerifierCache()
 

Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeImpl.kt

@@ -19,7 +19,6 @@ import io.ktor.http.path
 import io.ktor.util.decodeBase64String
 import kotlinx.atomicfu.atomic
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.delay
@@ -49,7 +48,7 @@ import kotlin.io.encoding.ExperimentalEncodingApi
     override val status: StateFlow<Realtime.Status> = _status.asStateFlow()
     private val _subscriptions = AtomicMutableMap<String, RealtimeChannel>()
     override val subscriptions: Map<String, RealtimeChannel> = _subscriptions
-    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+    private val scope = CoroutineScope(supabaseClient.coroutineDispatcher + SupervisorJob())
     private val mutex = Mutex()
     internal var accessToken by atomic<String?>(null)
     var heartbeatJob: Job? = null

Storage/src/commonMain/kotlin/io/github/jan/supabase/storage/resumable/ResumableClient.kt

@@ -125,9 +125,20 @@ internal class ResumableClientImpl(private val storageApi: BucketApi, private va
         val fingerprint = Fingerprint(source, size)
         val cacheEntry = ResumableCacheEntry(uploadUrl, path, storageApi.bucketId, Clock.System.now() + 1.days, uploadOptions.upsert, uploadOptions.contentType.toString())
         cache.set(fingerprint, cacheEntry)
-        return ResumableUploadImpl(fingerprint, path, cacheEntry, channel, 0, chunkSize, uploadUrl, httpClient, storageApi, { retrieveServerOffset(uploadUrl, path) }) {
-            cache.remove(fingerprint)
-        }
+        return ResumableUploadImpl(
+            fingerprint = fingerprint,
+            path = path,
+            cacheEntry = cacheEntry,
+            createDataStream = channel,
+            offset = 0,
+            chunkSize = chunkSize,
+            locationUrl = uploadUrl,
+            httpClient = httpClient,
+            storageApi = storageApi,
+            retrieveServerOffset = { retrieveServerOffset(uploadUrl, path) },
+            removeFromCache = { cache.remove(fingerprint) },
+            coroutineDispatcher = storageApi.supabaseClient.coroutineDispatcher
+        )
     }
 
     private suspend fun resumeUpload(channel: suspend (Long) -> ByteReadChannel, entry: ResumableCacheEntry, source: String, path: String, size: Long): ResumableUploadImpl {
@@ -142,9 +153,20 @@ internal class ResumableClientImpl(private val storageApi: BucketApi, private va
         }
         val offset = retrieveServerOffset(entry.url, path)
         if(offset < size) {
-            return ResumableUploadImpl(fingerprint, path, entry, channel, offset, chunkSize, entry.url, httpClient, storageApi, { retrieveServerOffset(entry.url, path)}) {
-                cache.remove(fingerprint)
-            }
+            return ResumableUploadImpl(
+                fingerprint = fingerprint,
+                path = path,
+                cacheEntry = entry,
+                createDataStream = channel,
+                offset = offset,
+                chunkSize = chunkSize,
+                locationUrl = entry.url,
+                httpClient = httpClient,
+                storageApi = storageApi,
+                retrieveServerOffset = { retrieveServerOffset(entry.url, path)},
+                removeFromCache = { cache.remove(fingerprint) },
+                coroutineDispatcher = storageApi.supabaseClient.coroutineDispatcher
+            )
         } else error("File already uploaded")
     }
 

Storage/src/commonMain/kotlin/io/github/jan/supabase/storage/resumable/ResumableUpload.kt

@@ -24,8 +24,8 @@ import io.ktor.utils.io.cancel
 import io.ktor.utils.io.readFully
 import io.ktor.utils.io.writeFully
 import kotlinx.atomicfu.atomic
+import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.ensureActive
@@ -84,7 +84,8 @@ internal class ResumableUploadImpl(
     private val httpClient: HttpClient,
     private val storageApi: BucketApi,
     private val retrieveServerOffset: suspend () -> Long,
-    private val removeFromCache: suspend () -> Unit
+    private val removeFromCache: suspend () -> Unit,
+    coroutineDispatcher: CoroutineDispatcher
 ): ResumableUpload {
 
     private val size = fingerprint.size
@@ -93,7 +94,7 @@ internal class ResumableUploadImpl(
     private var serverOffset = 0L
     private val _stateFlow = MutableStateFlow<ResumableUploadState>(ResumableUploadState(fingerprint, cacheEntry, UploadStatus.Progress(offset, size), paused))
     override val stateFlow: StateFlow<ResumableUploadState> = _stateFlow.asStateFlow()
-    private val scope = CoroutineScope(Dispatchers.Default)
+    private val scope = CoroutineScope(coroutineDispatcher)
     private val config = storageApi.supabaseClient.storage.config.resumable
     private lateinit var dataStream: ByteReadChannel
 

Supabase/src/androidMain/kotlin/io/github/jan/supabase/DefaultDispatcher.kt

@@ -0,0 +1,5 @@
+package io.github.jan.supabase
+
+import kotlinx.coroutines.Dispatchers
+
+actual val defaultDispatcher = Dispatchers.IO

Supabase/src/commonMain/kotlin/io/github/jan/supabase/DefaultDispatcher.kt

@@ -0,0 +1,5 @@
+package io.github.jan.supabase
+
+import kotlinx.coroutines.CoroutineDispatcher
+
+internal expect val defaultDispatcher: CoroutineDispatcher

Supabase/src/commonMain/kotlin/io/github/jan/supabase/SupabaseClient.kt

@@ -9,6 +9,7 @@ import io.github.jan.supabase.logging.i
 import io.github.jan.supabase.network.KtorSupabaseHttpClient
 import io.github.jan.supabase.plugins.PluginManager
 import io.github.jan.supabase.plugins.SupabasePlugin
+import kotlinx.coroutines.CoroutineDispatcher
 
 /**
  * The main class to interact with Supabase.
@@ -58,6 +59,12 @@ interface SupabaseClient {
     @SupabaseInternal
     val accessToken: AccessTokenProvider?
 
+    /**
+     * The default dispatcher used for launching long running jobs
+     */
+    @SupabaseInternal
+    val coroutineDispatcher: CoroutineDispatcher
+
     /**
      * Releases all resources held by the [httpClient] and all plugins the [pluginManager]
      */
@@ -93,6 +100,7 @@ internal class SupabaseClientImpl(
     override val supabaseUrl: String = config.supabaseUrl
     override val supabaseKey: String = config.supabaseKey
     override val useHTTPS: Boolean = config.networkConfig.useHTTPS
+    override val coroutineDispatcher: CoroutineDispatcher = config.coroutineDispatcher
 
     init {
         SupabaseClient.LOGGER.i {

Supabase/src/commonMain/kotlin/io/github/jan/supabase/SupabaseClientBuilder.kt

@@ -10,6 +10,7 @@ import io.github.jan.supabase.serializer.KotlinXSerializer
 import io.ktor.client.HttpClientConfig
 import io.ktor.client.engine.HttpClientEngine
 import io.ktor.client.plugins.HttpRequestTimeoutException
+import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.serialization.json.Json
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.seconds
@@ -69,6 +70,13 @@ class SupabaseClientBuilder @PublishedApi internal constructor(private val supab
      */
     var defaultSerializer: SupabaseSerializer = KotlinXSerializer(Json { ignoreUnknownKeys = true })
 
+    /**
+     * The CoroutineDispatcher used for launching long running jobs.
+     *
+     * Default: [defaultDispatcher], the IO Dispatcher on supported targets.
+     */
+    var coroutineDispatcher: CoroutineDispatcher = defaultDispatcher
+
     /**
      * Optional function for using a third-party authentication system with
      * Supabase. The function should return an access token or ID token (JWT) by
@@ -114,6 +122,7 @@ class SupabaseClientBuilder @PublishedApi internal constructor(private val supab
                 requestTimeout = requestTimeout
             ),
             defaultSerializer = defaultSerializer,
+            coroutineDispatcher = coroutineDispatcher,
             accessToken = accessToken,
             plugins = plugins
         )

Supabase/src/commonMain/kotlin/io/github/jan/supabase/SupabaseClientConfig.kt

@@ -2,6 +2,7 @@ package io.github.jan.supabase
 
 import io.github.jan.supabase.logging.LogLevel
 import io.ktor.client.engine.HttpClientEngine
+import kotlinx.coroutines.CoroutineDispatcher
 import kotlin.time.Duration
 
 internal data class SupabaseClientConfig(
@@ -10,6 +11,7 @@ internal data class SupabaseClientConfig(
     val defaultLogLevel: LogLevel,
     val networkConfig: SupabaseNetworkConfig,
     val defaultSerializer: SupabaseSerializer,
+    val coroutineDispatcher: CoroutineDispatcher,
     val accessToken: AccessTokenProvider?,
     val plugins: Map<String, PluginProvider>
 )