flow based callback for thumbnail generation

lib/src/main/java/com/otaliastudios/transcoder/internal/thumbnails/DefaultThumbnailsEngine.kt

@@ -26,8 +26,11 @@ import com.otaliastudios.transcoder.thumbnail.SingleThumbnailRequest
 import com.otaliastudios.transcoder.thumbnail.Thumbnail
 import com.otaliastudios.transcoder.thumbnail.ThumbnailRequest
 import com.otaliastudios.transcoder.time.DefaultTimeInterpolator
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.receiveAsFlow
 import kotlinx.coroutines.isActive
 import java.util.ArrayList
 
@@ -81,12 +84,10 @@ class DefaultThumbnailsEngine(
     private inner class IgnoringEosDataSource(
         private val source: DataSource,
     ) : DataSource by source {
-        override fun requestKeyFrameTimestamps(): Long {
-            return source.requestKeyFrameTimestamps()
-        }
-        override fun getKeyFrameTimestamps(): ArrayList<Long> {
-            return source.keyFrameTimestamps
-        }
+
+        override fun requestKeyFrameTimestamps() = source.requestKeyFrameTimestamps()
+
+        override fun getKeyFrameTimestamps() = source.keyFrameTimestamps
 
         override fun isDrained(): Boolean {
             if (source.isDrained) {
@@ -157,21 +158,22 @@ class DefaultThumbnailsEngine(
                                 "deltaUs=${stub.localizedUs - stub.actualLocalizedUs}"
                         )
                         val thumbnail = Thumbnail(stub.request, stub.positionUs, bitmap)
-                        progress(thumbnail)
+                        val callbackStatus = progress.trySend(thumbnail)
+                        log.i("Callback Send Status ${callbackStatus.isSuccess}")
                     }
                 }
         }
     }
 
-    private lateinit var progress: (Thumbnail) -> Unit
+    private val progress = Channel<Thumbnail>(Channel.BUFFERED)
 
-    private fun DataSource.lastKeyFrame(): Long {
-        return keyFrameAt(keyFrameTimestamps.size - 1)
-    }
 
-    private inline fun DataSource.keyFrameAt(index: Int, defaultValue: ((Int)-> Long) = {_ -> -1}): Long {
-        return keyFrameTimestamps.getOrElse(index, defaultValue)
-    }
+    private fun DataSource.lastKeyFrame() = keyFrameAt(keyFrameTimestamps.size - 1)
+
+    override val progressFlow: Flow<Thumbnail> = progress.receiveAsFlow()
+
+    private inline fun DataSource.keyFrameAt(index: Int, defaultValue: ((Int)-> Long) = {_ -> -1}) =
+        keyFrameTimestamps.getOrElse(index, defaultValue)
 
     private fun DataSource.search(timestampUs: Long): Int {
         if (keyFrameTimestamps.isEmpty())
@@ -206,12 +208,11 @@ class DefaultThumbnailsEngine(
         return nextKeyFrameIndex
     }
 
-    override suspend fun queueThumbnails(list: List<ThumbnailRequest>, progress: (Thumbnail) -> Unit) {
+    override suspend fun queueThumbnails(list: List<ThumbnailRequest>) {
         val segment = segments.next(TrackType.VIDEO)
         segment?.let {
             this.updatePositions(list, it.index)
         }
-        this.progress = progress
         while (currentCoroutineContext().isActive) {
             val advanced = segments.next(TrackType.VIDEO)?.advance() ?: false
             val completed = !advanced && !segments.hasNext() // avoid calling hasNext if we advanced.

lib/src/main/java/com/otaliastudios/transcoder/internal/thumbnails/ThumbnailsEngine.kt

@@ -7,10 +7,13 @@ import com.otaliastudios.transcoder.internal.DataSources
 import com.otaliastudios.transcoder.internal.utils.Logger
 import com.otaliastudios.transcoder.thumbnail.Thumbnail
 import com.otaliastudios.transcoder.thumbnail.ThumbnailRequest
+import kotlinx.coroutines.flow.Flow
 
 abstract class ThumbnailsEngine {
 
-    abstract suspend fun queueThumbnails(list: List<ThumbnailRequest>, progress: (Thumbnail) -> Unit)
+    abstract val progressFlow: Flow<Thumbnail>
+
+    abstract suspend fun queueThumbnails(list: List<ThumbnailRequest>)
 
     abstract suspend fun removePosition(positionUs: Long)
 
@@ -43,24 +46,5 @@ abstract class ThumbnailsEngine {
             return engine
         }
     }
-    suspend fun queue(list: List<ThumbnailRequest>) {
-        engine?.queueThumbnails(list) {
-            dispatcher.dispatchThumbnail(it)
-        }
 
-        try {
-            dispatcher.dispatchCompletion()
-        } catch (e: Exception) {
-            if (e.isInterrupted()) {
-                log.i("Transcode canceled.", e)
-                dispatcher.dispatchCancel()
-            } else {
-                log.e("Unexpected error while transcoding.", e)
-                dispatcher.dispatchFailure(e)
-                throw e
-            }
-        } finally {
-            engine?.cleanup()
-        }
-    }
 }