musikr: include context in pipeline errors

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

@@ -66,20 +66,20 @@ private class EvaluateStepImpl(
         val rawSongs = filterFlow.right
         val preSongs =
             rawSongs
-                .map { tagInterpreter.interpret(it) }
+                .map { wrap(it, tagInterpreter::interpret) }
                 .flowOn(Dispatchers.Default)
                 .buffer(Channel.UNLIMITED)
         val prePlaylists =
             filterFlow.left
-                .map { playlistInterpreter.interpret(it) }
+                .map { wrap(it, playlistInterpreter::interpret) }
                 .flowOn(Dispatchers.Default)
                 .buffer(Channel.UNLIMITED)
         val graphBuilder = MusicGraph.builder()
         val graphBuild =
             merge(
                 filterFlow.manager,
-                preSongs.onEach { graphBuilder.add(it) },
-                prePlaylists.onEach { graphBuilder.add(it) })
+                preSongs.onEach { wrap(it, graphBuilder::add) },
+                prePlaylists.onEach { wrap(it, graphBuilder::add) })
         graphBuild.collect()
         val graph = graphBuilder.build()
         return libraryFactory.create(graph, storedPlaylists, playlistInterpreter)

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

@@ -15,7 +15,7 @@
  * 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.pipeline
 
 import android.content.Context
@@ -48,7 +48,8 @@ internal interface ExtractStep {
                 MetadataExtractor.from(context),
                 TagParser.new(),
                 storage.cache,
-                storage.storedCovers)
+                storage.storedCovers
+            )
     }
 }
 
@@ -70,7 +71,8 @@ private class ExtractStepImpl(
         val playlistNodes = filterFlow.left.map { ExtractedMusic.Playlist(it) }
 
         val cacheResults =
-            audioNodes.map { cache.read(it) }.flowOn(Dispatchers.IO).buffer(Channel.UNLIMITED)
+            audioNodes.map { wrap(it, cache::read) }.flowOn(Dispatchers.IO)
+                .buffer(Channel.UNLIMITED)
         val cacheFlow =
             cacheResults.divert {
                 when (it) {
@@ -84,11 +86,13 @@ private class ExtractStepImpl(
         val extractedSongs =
             Array(distributedFlow.flows.size) { i ->
                 distributedFlow.flows[i]
-                    .mapNotNull { file ->
-                        val metadata = metadataExtractor.extract(file) ?: return@mapNotNull null
-                        val tags = tagParser.parse(file, metadata)
-                        val cover = metadata.cover?.let { storedCovers.write(it) }
-                        RawSong(file, metadata.properties, tags, cover)
+                    .mapNotNull { it ->
+                        wrap(it) { file ->
+                            val metadata = metadataExtractor.extract(file) ?: return@wrap null
+                            val tags = tagParser.parse(file, metadata)
+                            val cover = metadata.cover?.let { storedCovers.write(it) }
+                            RawSong(file, metadata.properties, tags, cover)
+                        }
                     }
                     .flowOn(Dispatchers.IO)
                     .buffer(Channel.UNLIMITED)
@@ -96,7 +100,7 @@ private class ExtractStepImpl(
         val writtenSongs =
             merge(*extractedSongs)
                 .map {
-                    cache.write(it)
+                    wrap(it, cache::write)
                     ExtractedMusic.Song(it)
                 }
                 .flowOn(Dispatchers.IO)
@@ -107,7 +111,8 @@ private class ExtractStepImpl(
             cachedSongs,
             distributedFlow.manager,
             writtenSongs,
-            playlistNodes)
+            playlistNodes
+        )
     }
 }
 

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

@@ -0,0 +1,72 @@
+package org.oxycblt.musikr.pipeline
+
+import org.oxycblt.musikr.fs.DeviceFile
+import org.oxycblt.musikr.playlist.PlaylistFile
+import org.oxycblt.musikr.playlist.interpret.PrePlaylist
+import org.oxycblt.musikr.tag.interpret.PreSong
+
+class PipelineException(
+    val processing: WhileProcessing,
+    val error: Exception
+) : Exception() {
+    override val cause = error
+
+    override val message = "Error while processing ${processing}: $error"
+}
+
+sealed interface WhileProcessing {
+    class AFile internal constructor(private val file: DeviceFile) : WhileProcessing {
+        override fun toString() = "File @ ${file.path}"
+    }
+
+    class ARawSong internal constructor(private val rawSong: RawSong) : WhileProcessing {
+        override fun toString() = "Raw Song @ ${rawSong.file.path}"
+    }
+
+    class APlaylistFile internal constructor(private val playlist: PlaylistFile) : WhileProcessing {
+        override fun toString() = "Playlist File @ ${playlist.name}"
+    }
+
+    class APreSong internal constructor(private val preSong: PreSong) : WhileProcessing {
+        override fun toString() = "Pre Song @ ${preSong.path}"
+    }
+
+    class APrePlaylist internal constructor(private val prePlaylist: PrePlaylist) : WhileProcessing {
+        override fun toString() = "Pre Playlist @ ${prePlaylist.name}"
+    }
+}
+
+internal suspend fun <R> wrap(file: DeviceFile, block: suspend (DeviceFile) -> R): R =
+    try {
+        block(file)
+    } catch (e: Exception) {
+        throw PipelineException(WhileProcessing.AFile(file), e)
+    }
+
+internal suspend fun <R> wrap(song: RawSong, block: suspend (RawSong) -> R): R =
+    try {
+        block(song)
+    } catch (e: Exception) {
+        throw PipelineException(WhileProcessing.ARawSong(song), e)
+    }
+
+internal suspend fun <R> wrap(file: PlaylistFile, block: suspend (PlaylistFile) -> R): R =
+    try {
+        block(file)
+    } catch (e: Exception) {
+        throw PipelineException(WhileProcessing.APlaylistFile(file), e)
+    }
+
+internal suspend fun <R> wrap(song: PreSong, block: suspend (PreSong) -> R): R =
+    try {
+        block(song)
+    } catch (e: Exception) {
+        throw PipelineException(WhileProcessing.APreSong(song), e)
+    }
+
+internal suspend fun <R> wrap(playlist: PrePlaylist, block: suspend (PrePlaylist) -> R): R =
+    try {
+        block(playlist)
+    } catch (e: Exception) {
+        throw PipelineException(WhileProcessing.APrePlaylist(playlist), e)
+    }