data: stream source artifacts with kotlinx-io

Author: michalharakal

docs/modules/ROOT/pages/tutorials/data-sources-getting-started.adoc

@@ -40,9 +40,9 @@ dependencies {
 === Resolve one artifact
 
 `JvmDataSourceResolver` materializes remote artifacts into a cache and returns
-a `DataSourceArtifact` that can be read as bytes. Public Hugging Face files do
-not need credentials. Private files can use an `Authorization` header, or the
-JVM resolver will read `HF_TOKEN` / `HUGGING_FACE_HUB_TOKEN` from the
+a `DataSourceArtifact` that opens a `kotlinx.io.Source`. Public Hugging Face
+files do not need credentials. Private files can use an `Authorization` header,
+or the JVM resolver will read `HF_TOKEN` / `HUGGING_FACE_HUB_TOKEN` from the
 environment when the URI provider is Hugging Face.
 
 [source,kotlin]
@@ -60,6 +60,14 @@ val artifact = resolver.resolve(
 println(artifact.filename)
 println(artifact.localPath)
 
+val source = artifact.openSource()
+try {
+    // Pass the source to a parser/loader for model-sized artifacts.
+} finally {
+    source.close()
+}
+
+// Convenience for small sidecars and tests.
 val bytes = artifact.readBytes()
 ----
 

skainet-data/skainet-data-source/build.gradle.kts

@@ -18,6 +18,7 @@ kotlin {
     sourceSets {
         commonMain.dependencies {
             implementation(libs.kotlinx.coroutines)
+            implementation(libs.kotlinx.io.core)
         }
 
         commonTest.dependencies {

skainet-data/skainet-data-source/src/commonMain/kotlin/sk/ainet/data/source/DataSourceModels.kt

@@ -1,5 +1,9 @@
 package sk.ainet.data.source
 
+import kotlinx.io.Sink
+import kotlinx.io.Source
+import kotlinx.io.readByteArray
+
 /**
  * Cache behavior requested by a caller resolving a data artifact.
  */
@@ -80,9 +84,38 @@ public class DataSourceArtifact(
     public val localPath: String?,
     public val sizeBytes: Long?,
     public val cacheHit: Boolean,
-    private val byteReader: suspend () -> ByteArray
+    private val sourceOpener: suspend () -> Source
 ) {
-    public suspend fun readBytes(): ByteArray = byteReader()
+    /**
+     * Opens a fresh source for this artifact. Callers own and must close it.
+     */
+    public suspend fun openSource(): Source = sourceOpener()
+
+    /**
+     * Convenience for small artifacts. Prefer [openSource] or [copyTo] for
+     * model-scale data.
+     */
+    public suspend fun readBytes(): ByteArray {
+        val source = openSource()
+        return try {
+            source.readByteArray()
+        } finally {
+            source.close()
+        }
+    }
+
+    /**
+     * Streams this artifact into [sink]. The source is closed after copying;
+     * [sink] is left open for the caller.
+     */
+    public suspend fun copyTo(sink: Sink): Long {
+        val source = openSource()
+        return try {
+            source.transferTo(sink)
+        } finally {
+            source.close()
+        }
+    }
 }
 
 /**