musikr: introduce streaming file tree

Probably the best way I can go about this.

Author: OxygenCobalt

musikr/src/main/java/org/oxycblt/musikr/covers/fs/FSCovers.kt

@@ -25,6 +25,9 @@ import android.os.ParcelFileDescriptor
 import androidx.core.net.toUri
 import java.io.InputStream
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.filterIsInstance
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.withContext
 import org.oxycblt.musikr.covers.Cover
 import org.oxycblt.musikr.covers.CoverResult
@@ -96,11 +99,12 @@ class MutableFSCovers(private val context: Context) : MutableCovers<FDCover> {
     override suspend fun create(file: DeviceFile, metadata: Metadata): CoverResult<FDCover> {
         // Since DeviceFiles is a streaming API, we have to wait for the current recursive
         // query to finally finish to be able to have a complete list of siblings to search for.
-        val parent = file.parent.await()
+        val parent = file.parent
         val bestCover =
-            parent.children
+            parent.children.flow()
                 .filterIsInstance<DeviceFile>()
                 .map { it to coverArtScore(it) }
+                .toList()
                 .maxBy { it.second }
         if (bestCover.second > 0) {
             return CoverResult.Hit(FolderCoverImpl(context, bestCover.first.uri))
@@ -128,7 +132,7 @@ class MutableFSCovers(private val context: Context) : MutableCovers<FDCover> {
         // See if the name contains any of the preferred cover names. This helps weed out
         // images that are not actually cover art and are just there.,
         var score =
-            (preferredCoverNames + requireNotNull(file.parent.await().path.name))
+            (preferredCoverNames + requireNotNull(file.parent.path.name))
                 .withIndex()
                 .filter { name.contains(it.value, ignoreCase = true) }
                 .sumOf { it.index + 1 }

musikr/src/main/java/org/oxycblt/musikr/fs/device/DeviceFS.kt

@@ -15,27 +15,31 @@
  * You should have received a copy of the GNU General Public License
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
- 
+
 package org.oxycblt.musikr.fs.device
 
 import android.content.ContentResolver
 import android.content.Context
 import android.net.Uri
 import android.provider.DocumentsContract
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.flatMapMerge
 import kotlinx.coroutines.flow.flattenMerge
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.map
 import org.oxycblt.musikr.fs.MusicLocation
 import org.oxycblt.musikr.fs.Path
 
 internal interface DeviceFS {
-    fun explore(locations: Flow<MusicLocation>): Flow<DeviceFile>
+    fun explore(locations: Flow<MusicLocation>): Flow<DeviceDirectory>
 
     companion object {
         fun from(context: Context, withHidden: Boolean): DeviceFS =
@@ -48,27 +52,39 @@ private class DeviceFSImpl(
     private val contentResolver: ContentResolver,
     private val withHidden: Boolean
 ) : DeviceFS {
-    override fun explore(locations: Flow<MusicLocation>): Flow<DeviceFile> =
-        locations.flatMapMerge { location ->
-            exploreDirectoryImpl(
-                location.uri,
-                DocumentsContract.getTreeDocumentId(location.uri),
-                location.path,
-                null)
+    override fun explore(locations: Flow<MusicLocation>): Flow<DeviceDirectory> =
+        locations.map { location ->
+            coroutineScope {
+                DeviceDirectoryImpl(
+                    location.uri,
+                    DocumentsContract.getTreeDocumentId(location.uri),
+                    location.path,
+                    null,
+                    coroutineScope = CoroutineScope(Dispatchers.IO),
+                    contentResolver = contentResolver,
+                    withHidden = withHidden
+                )
+            }
         }
+}
 
-    private fun exploreDirectoryImpl(
-        rootUri: Uri,
-        treeDocumentId: String,
-        relativePath: Path,
-        parent: Deferred<DeviceDirectory>?
-    ): Flow<DeviceFile> = flow {
-        // Make a kotlin future
-        val uri = DocumentsContract.buildChildDocumentsUriUsingTree(rootUri, treeDocumentId)
-        val directoryDeferred = CompletableDeferred<DeviceDirectory>()
-        val recursive = mutableListOf<Flow<DeviceFile>>()
-        val children = mutableListOf<DeviceFSEntry>()
-        contentResolver.useQuery(uri, PROJECTION) { cursor ->
+private class DeviceDirectoryImpl(
+    rootUri: Uri,
+    treeDocumentId: String,
+    override val path: Path,
+    override val parent: DeviceDirectoryImpl?,
+    coroutineScope: CoroutineScope,
+    contentResolver: ContentResolver,
+    withHidden: Boolean
+) : DeviceDirectory {
+    override val uri: Uri = DocumentsContract.buildDocumentUriUsingTree(rootUri, treeDocumentId)
+    override val children = FiniteHotFlow<DeviceFSEntry>(coroutineScope) { emit ->
+        contentResolver.useQuery(
+            DocumentsContract.buildChildDocumentsUriUsingTree(
+                rootUri,
+                treeDocumentId
+            ), PROJECTION
+        ) { cursor ->
             val childUriIndex =
                 cursor.getColumnIndexOrThrow(DocumentsContract.Document.COLUMN_DOCUMENT_ID)
             val displayNameIndex =
@@ -88,13 +104,22 @@ private class DeviceFSImpl(
                     continue
                 }
 
-                val newPath = relativePath.file(displayName)
+                val newPath = path.file(displayName)
                 val mimeType = cursor.getString(mimeTypeIndex)
                 val lastModified = cursor.getLong(lastModifiedIndex)
 
                 if (mimeType == DocumentsContract.Document.MIME_TYPE_DIR) {
-                    recursive.add(
-                        exploreDirectoryImpl(rootUri, childId, newPath, directoryDeferred))
+                    emit(
+                        DeviceDirectoryImpl(
+                            rootUri,
+                            childId,
+                            path = newPath,
+                            parent = this,
+                            coroutineScope,
+                            contentResolver,
+                            withHidden
+                        )
+                    )
                 } else {
                     val size = cursor.getLong(sizeIndex)
                     val childUri = DocumentsContract.buildDocumentUriUsingTree(rootUri, childId)
@@ -105,23 +130,24 @@ private class DeviceFSImpl(
                             path = newPath,
                             size = size,
                             modifiedMs = lastModified,
-                            parent = directoryDeferred)
-                    children.add(file)
+                            parent = this
+                        )
                     emit(file)
                 }
             }
-            directoryDeferred.complete(DeviceDirectory(uri, relativePath, parent, children))
-            emitAll(recursive.asFlow().flattenMerge())
         }
     }
 
     private companion object {
-        val PROJECTION =
+        private val PROJECTION =
             arrayOf(
                 DocumentsContract.Document.COLUMN_DOCUMENT_ID,
                 DocumentsContract.Document.COLUMN_DISPLAY_NAME,
                 DocumentsContract.Document.COLUMN_MIME_TYPE,
                 DocumentsContract.Document.COLUMN_SIZE,
-                DocumentsContract.Document.COLUMN_LAST_MODIFIED)
+                DocumentsContract.Document.COLUMN_LAST_MODIFIED
+            )
     }
 }
+
+

musikr/src/main/java/org/oxycblt/musikr/fs/device/DeviceFSEntries.kt

@@ -20,25 +20,35 @@ package org.oxycblt.musikr.fs.device
 
 import android.net.Uri
 import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flatMapMerge
+import kotlinx.coroutines.flow.flowOf
 import org.oxycblt.musikr.fs.Path
 
 sealed interface DeviceFSEntry {
     val uri: Uri
     val path: Path
 }
 
-data class DeviceDirectory(
-    override val uri: Uri,
-    override val path: Path,
-    val parent: Deferred<DeviceDirectory>?,
-    var children: List<DeviceFSEntry>
-) : DeviceFSEntry
+interface DeviceDirectory : DeviceFSEntry {
+    val parent: DeviceDirectory?
+    val children: FiniteHotFlow<DeviceFSEntry>
+}
 
 data class DeviceFile(
     override val uri: Uri,
     override val path: Path,
     val modifiedMs: Long,
     val mimeType: String,
     val size: Long,
-    val parent: Deferred<DeviceDirectory>
+    val parent: DeviceDirectory
 ) : DeviceFSEntry
+
+@OptIn(ExperimentalCoroutinesApi::class)
+fun DeviceDirectory.flatten(): Flow<DeviceFile> = children.flow().flatMapMerge {
+    when (it) {
+        is DeviceDirectory -> it.flatten()
+        is DeviceFile -> flowOf(it)
+    }
+}

musikr/src/main/java/org/oxycblt/musikr/fs/device/FiniteHotFlow.kt

@@ -0,0 +1,31 @@
+package org.oxycblt.musikr.fs.device
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.shareIn
+import kotlinx.coroutines.flow.takeWhile
+
+class FiniteHotFlow<T>(scope: CoroutineScope, emitter: suspend (emit: suspend (T) -> Unit) -> Unit) {
+    private sealed interface HotObject<T> {
+        data class More<T>(val value: T) : HotObject<T>
+        data object Done : HotObject<Nothing>
+    }
+
+    private val sharedFlow = flow {
+        emitter {
+            emit(HotObject.More(it))
+        }
+        emit(HotObject.Done)
+    }.shareIn(scope, SharingStarted.Lazily, replay = Int.MAX_VALUE)
+
+    @Suppress("UNCHECKED_CAST")
+    fun flow() = flow {
+        sharedFlow.takeWhile { it is HotObject.More<*> }.collect {
+            emit((it as HotObject.More<*>).value as T)
+        }
+    }
+}

musikr/src/main/java/org/oxycblt/musikr/pipeline/ExploreStep.kt

@@ -20,12 +20,14 @@ package org.oxycblt.musikr.pipeline
 
 import android.content.Context
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.flatMapMerge
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.map
@@ -40,6 +42,7 @@ import org.oxycblt.musikr.covers.CoverResult
 import org.oxycblt.musikr.covers.Covers
 import org.oxycblt.musikr.fs.MusicLocation
 import org.oxycblt.musikr.fs.device.DeviceFS
+import org.oxycblt.musikr.fs.device.flatten
 import org.oxycblt.musikr.playlist.db.StoredPlaylists
 import org.oxycblt.musikr.playlist.m3u.M3U
 
@@ -62,13 +65,13 @@ private class ExploreStepImpl(
     private val covers: Covers<out Cover>,
     private val storedPlaylists: StoredPlaylists
 ) : ExploreStep {
+    @OptIn(ExperimentalCoroutinesApi::class)
     override fun explore(locations: List<MusicLocation>): Flow<Explored> {
         val addingMs = System.currentTimeMillis()
         return merge(
             deviceFS
-                .explore(
-                    locations.asFlow(),
-                )
+                .explore(locations.asFlow())
+                .flatMapMerge { it.flatten() }
                 .filter { it.mimeType.startsWith("audio/") || it.mimeType == M3U.MIME_TYPE }
                 .distributedMap(n = 8, on = Dispatchers.IO, buffer = Channel.UNLIMITED) { file ->
                     when (val cacheResult = cache.read(file)) {