Add ByteChannel

app/build.gradle.kts

@@ -89,6 +89,8 @@ dependencies {
 
     implementation("com.github.Kyant0:taglib:1.0.0-alpha17")
 
+    implementation("io.ktor:ktor-utils:2.3.0")
+
     //TODO : use fetch instead of download manager
 //    implementation("com.github.tonyofrancis.Fetch:xfetch2:3.1.6")
 

app/src/main/java/dev/brahmkshatriya/echo/playback/AudioDataSource.kt

@@ -14,6 +14,7 @@ import dev.brahmkshatriya.echo.playback.AudioResolver.Companion.copy
 class AudioDataSource(
     private val defaultDataSourceFactory: DefaultDataSource.Factory,
     private val byteStreamDataSourceFactory: ByteStreamDataSource.Factory,
+    private val byteChannelDataSourceFactory: ByteChannelDataSource.Factory,
 ) : BaseDataSource(true) {
 
     class Factory(
@@ -22,8 +23,9 @@ class AudioDataSource(
 
         private val defaultDataSourceFactory = DefaultDataSource.Factory(context)
         private val byteStreamDataSourceFactory = ByteStreamDataSource.Factory()
+        private val byteChannelDataSourceFactory = ByteChannelDataSource.Factory()
         override fun createDataSource() =
-            AudioDataSource(defaultDataSourceFactory, byteStreamDataSourceFactory)
+            AudioDataSource(defaultDataSourceFactory, byteStreamDataSourceFactory, byteChannelDataSourceFactory)
     }
 
     private var source: DataSource? = null
@@ -46,6 +48,11 @@ class AudioDataSource(
                 byteStreamDataSourceFactory.createDataSource() to spec
             }
 
+            is Streamable.Audio.Channel -> {
+                val spec = dataSpec.copy(customData = audio)
+                byteChannelDataSourceFactory.createDataSource() to spec
+            }
+
             is Streamable.Audio.Http -> {
                 val spec = audio.request.run {
                     dataSpec.copy(uri = url.toUri(), httpRequestHeaders = headers)

app/src/main/java/dev/brahmkshatriya/echo/playback/ByteChannelDataSource.kt

@@ -0,0 +1,73 @@
+package dev.brahmkshatriya.echo.playback
+
+import androidx.core.net.toUri
+import androidx.media3.common.C
+import androidx.media3.common.util.UnstableApi
+import androidx.media3.datasource.BaseDataSource
+import androidx.media3.datasource.DataSource
+import androidx.media3.datasource.DataSpec
+import dev.brahmkshatriya.echo.common.models.Streamable
+import io.ktor.utils.io.ByteReadChannel
+import io.ktor.utils.io.cancel
+import kotlinx.coroutines.runBlocking
+import java.io.IOException
+import kotlin.math.min
+
+@UnstableApi
+class ByteChannelDataSource : BaseDataSource(true) {
+
+    class Factory : DataSource.Factory {
+        override fun createDataSource() = ByteStreamDataSource()
+    }
+
+    private var audio: Streamable.Audio.Channel? = null
+    private var channel: ByteReadChannel? = null
+
+    override fun read(buffer: ByteArray, offset: Int, length: Int): Int {
+        val channel = channel ?: throw IOException("Channel is not open")
+        return runBlocking {
+            val bytesRead = channel.readAvailable(buffer, offset, length)
+            if (bytesRead == -1) C.RESULT_END_OF_INPUT else bytesRead
+        }
+    }
+
+    override fun open(dataSpec: DataSpec): Long {
+        val audio = dataSpec.customData as Streamable.Audio.Channel
+        val requestedPosition = dataSpec.position
+
+        // Attempt to seek to the requested position
+        channel = audio.channel
+        this.audio = audio
+
+        runBlocking {
+            seekChannelToPosition(channel!!, requestedPosition)
+        }
+
+        return audio.totalBytes - requestedPosition
+    }
+
+    override fun getUri() = audio?.hashCode().toString().toUri()
+
+    override fun close() {
+        runBlocking {
+            channel?.cancel()
+        }
+        channel = null
+        audio = null
+    }
+
+    private suspend fun seekChannelToPosition(channel: ByteReadChannel, requestedPosition: Long) {
+        if (requestedPosition > 0) {
+            var remaining = requestedPosition
+            val discardBuffer = ByteArray(8192)
+            while (remaining > 0) {
+                val toRead = min(remaining, discardBuffer.size.toLong()).toInt()
+                val readBytes = channel.readAvailable(discardBuffer, 0, toRead)
+                if (readBytes == -1) {
+                    throw IOException("Reached end of stream before desired position")
+                }
+                remaining -= readBytes
+            }
+        }
+    }
+}

common/build.gradle.kts

@@ -16,6 +16,7 @@ kotlin {
 
 dependencies {
     api("org.jetbrains.kotlinx:kotlinx-serialization-json:1.7.1")
+    implementation("io.ktor:ktor-utils:2.3.0")
 }
 
 publishing {

common/src/main/java/dev/brahmkshatriya/echo/common/models/Streamable.kt

@@ -1,6 +1,8 @@
 package dev.brahmkshatriya.echo.common.models
 
 import dev.brahmkshatriya.echo.common.models.Request.Companion.toRequest
+import io.ktor.utils.io.ByteChannel
+import io.ktor.utils.io.ByteReadChannel
 import kotlinx.serialization.Serializable
 import java.io.InputStream
 
@@ -68,6 +70,10 @@ data class Streamable(
             val stream: InputStream, val totalBytes: Long, override val skipSilence: Boolean? = null
         ) : Audio()
 
+        data class Channel(
+            val channel: ByteReadChannel, val totalBytes: Long, override val skipSilence: Boolean? = null
+        ) : Audio()
+
         companion object {
             fun String.toAudio(headers: Map<String, String> = mapOf()) =
                 Http(this.toRequest(headers))