Fix concatenation issues (#130)

lib/src/main/java/com/otaliastudios/transcoder/internal/Codecs.kt

@@ -0,0 +1,69 @@
+package com.otaliastudios.transcoder.internal
+
+import android.media.MediaCodec
+import android.media.MediaFormat
+import android.view.Surface
+import com.otaliastudios.transcoder.common.TrackStatus
+import com.otaliastudios.transcoder.common.TrackType
+import com.otaliastudios.transcoder.internal.media.MediaFormatProvider
+import com.otaliastudios.transcoder.internal.utils.Logger
+import com.otaliastudios.transcoder.internal.utils.TrackMap
+import com.otaliastudios.transcoder.internal.utils.trackMapOf
+import com.otaliastudios.transcoder.source.DataSource
+import com.otaliastudios.transcoder.strategy.TrackStrategy
+
+/**
+ * Encoders are shared between segments. This is not strictly needed but it is more efficient
+ * and solves timestamp issues that arise due to the fact that MediaCodec can alter the timestamps
+ * internally, so if we use different MediaCodec instances we don't have guarantees on monotonic
+ * output timestamps, even if input timestamps are. This would later create crashes when passing
+ * data to MediaMuxer / MPEG4Writer.
+ */
+internal class Codecs(
+        private val sources: DataSources,
+        private val tracks: Tracks,
+        private val current: TrackMap<Int>
+) {
+
+    private val log = Logger("Codecs")
+
+    val encoders = object : TrackMap<Pair<MediaCodec, Surface?>> {
+
+        override fun has(type: TrackType) = tracks.all[type] == TrackStatus.COMPRESSING
+
+        private val lazyAudio by lazy {
+            val format = tracks.outputFormats.audio
+            val codec = MediaCodec.createEncoderByType(format.getString(MediaFormat.KEY_MIME)!!)
+            codec.configure(format, null, null, MediaCodec.CONFIGURE_FLAG_ENCODE)
+            codec to null
+        }
+
+        private val lazyVideo by lazy {
+            val format = tracks.outputFormats.video
+            val codec = MediaCodec.createEncoderByType(format.getString(MediaFormat.KEY_MIME)!!)
+            codec.configure(format, null, null, MediaCodec.CONFIGURE_FLAG_ENCODE)
+            codec to codec.createInputSurface()
+        }
+
+        override fun get(type: TrackType) = when (type) {
+            TrackType.AUDIO -> lazyAudio
+            TrackType.VIDEO -> lazyVideo
+        }
+    }
+
+    val ownsEncoderStart = object : TrackMap<Boolean> {
+        override fun has(type: TrackType) = true
+        override fun get(type: TrackType) = current[type] == 0
+    }
+
+    val ownsEncoderStop = object : TrackMap<Boolean> {
+        override fun has(type: TrackType) = true
+        override fun get(type: TrackType) = current[type] == sources[type].lastIndex
+    }
+
+    fun release() {
+        encoders.forEach {
+            it.first.release()
+        }
+    }
+}

lib/src/main/java/com/otaliastudios/transcoder/internal/Segments.kt

@@ -16,13 +16,7 @@ internal class Segments(
 
     private val log = Logger("Segments")
     private val current = mutableTrackMapOf<Segment>(null, null)
-    val currentIndex = object : TrackMap<Int> {
-        override fun has(type: TrackType) = true
-        override fun get(type: TrackType): Int {
-            return current.getOrNull(type)?.index ?: -1
-        }
-    }
-
+    val currentIndex = mutableTrackMapOf(-1, -1)
     private val requestedIndex = mutableTrackMapOf(0, 0)
 
     fun hasNext(type: TrackType): Boolean {
@@ -70,12 +64,14 @@ internal class Segments(
     private fun tryCreateSegment(type: TrackType, index: Int): Segment? {
         // Return null if out of bounds, either because segments are over or because the
         // source set does not have sources for this track type.
-        log.i("tryCreateSegment($type, $index)...")
         val source = sources[type].getOrNull(index) ?: return null
         log.i("tryCreateSegment($type, $index): created!")
         if (tracks.active.has(type)) {
             source.selectTrack(type)
         }
+        // Update current index before pipeline creation, for other components
+        // who check it during pipeline init.
+        currentIndex[type] = index
         val pipeline = factory(
                 type,
                 index,

lib/src/main/java/com/otaliastudios/transcoder/internal/Timer.kt

@@ -82,7 +82,6 @@ internal class Timer(
                 }
             }
 
-            // TODO consider using localize instead of this lastOut trick.
             override fun interpolate(type: TrackType, time: Long) = when (time) {
                 Long.MAX_VALUE -> lastOut
                 else -> {

lib/src/main/java/com/otaliastudios/transcoder/internal/codec/Encoder.kt

@@ -1,20 +1,20 @@
 package com.otaliastudios.transcoder.internal.codec
 
+import android.media.MediaCodec
 import android.media.MediaCodec.*
-import android.media.MediaFormat
 import android.view.Surface
 import com.otaliastudios.transcoder.common.TrackType
-import com.otaliastudios.transcoder.common.trackType
+import com.otaliastudios.transcoder.internal.Codecs
 import com.otaliastudios.transcoder.internal.data.WriterChannel
 import com.otaliastudios.transcoder.internal.data.WriterData
 import com.otaliastudios.transcoder.internal.media.MediaCodecBuffers
-import com.otaliastudios.transcoder.internal.pipeline.BaseStep
 import com.otaliastudios.transcoder.internal.pipeline.Channel
 import com.otaliastudios.transcoder.internal.pipeline.QueuedStep
 import com.otaliastudios.transcoder.internal.pipeline.State
 import com.otaliastudios.transcoder.internal.utils.Logger
-import com.otaliastudios.transcoder.source.DataSource
 import java.nio.ByteBuffer
+import kotlin.properties.Delegates
+import kotlin.properties.Delegates.observable
 
 internal data class EncoderData(
         val buffer: ByteBuffer?, // If present, it must have correct position/remaining!
@@ -30,40 +30,61 @@ internal interface EncoderChannel : Channel {
 }
 
 internal class Encoder(
-        private val format: MediaFormat, // desired output format
+        private val codec: MediaCodec,
+        override val surface: Surface?,
+        ownsCodecStart: Boolean,
+        private val ownsCodecStop: Boolean,
 ) : QueuedStep<EncoderData, EncoderChannel, WriterData, WriterChannel>(), EncoderChannel {
 
-    private val log = Logger("Encoder")
-    override val channel = this
+    constructor(codecs: Codecs, type: TrackType) : this(
+            codecs.encoders[type].first,
+            codecs.encoders[type].second,
+            codecs.ownsEncoderStart[type],
+            codecs.ownsEncoderStop[type]
+    )
 
-    private val codec = createEncoderByType(format.getString(MediaFormat.KEY_MIME)!!).also {
-        it.configure(format, null, null, CONFIGURE_FLAG_ENCODE)
+    companion object {
+        // Debugging
+        private val log = Logger("Encoder")
+        private var dequeuedInputs by observable(0) { _, _, _ -> printDequeued() }
+        private var dequeuedOutputs by observable(0) { _, _, _ -> printDequeued() }
+        private fun printDequeued() {
+            log.v("dequeuedInputs=$dequeuedInputs dequeuedOutputs=$dequeuedOutputs")
+        }
     }
 
-    override val surface = when (format.trackType) {
-        TrackType.VIDEO -> codec.createInputSurface()
-        else -> null
-    }
+    override val channel = this
 
     private val buffers by lazy { MediaCodecBuffers(codec) }
 
     private var info = BufferInfo()
 
+
     init {
-        codec.start()
+        log.i("Encoder: ownsStart=$ownsCodecStart ownsStop=$ownsCodecStop")
+        if (ownsCodecStart) {
+            codec.start()
+        }
     }
 
     override fun buffer(): Pair<ByteBuffer, Int>? {
         val id = codec.dequeueInputBuffer(0)
         log.v("buffer(): id=$id")
+        if (id >= 0) dequeuedInputs++
         return if (id >= 0) buffers.getInputBuffer(id) to id else null
     }
 
+    private var eosReceivedButNotEnqueued = false
+
     override fun enqueueEos(data: EncoderData) {
-        if (surface != null) codec.signalEndOfInputStream()
-        else {
+        if (!ownsCodecStop) {
+            eosReceivedButNotEnqueued = true
+        } else if (surface != null) {
+            codec.signalEndOfInputStream()
+        } else {
             val flag = BUFFER_FLAG_END_OF_STREAM
             codec.queueInputBuffer(data.id, 0, 0, 0, flag)
+            dequeuedInputs--
         }
     }
 
@@ -72,14 +93,23 @@ internal class Encoder(
         else {
             val buffer = requireNotNull(data.buffer) { "Audio should always pass a buffer to Encoder." }
             codec.queueInputBuffer(data.id, buffer.position(), buffer.remaining(), data.timeUs, 0)
+            dequeuedInputs--
         }
     }
 
     override fun drain(): State<WriterData> {
-        return when (val result = codec.dequeueOutputBuffer(info, 0)) {
+        val timeoutUs = if (eosReceivedButNotEnqueued) 5000L else 0L
+        return when (val result = codec.dequeueOutputBuffer(info, timeoutUs)) {
             INFO_TRY_AGAIN_LATER -> {
-                log.e("Can't dequeue output buffer: INFO_TRY_AGAIN_LATER")
-                State.Wait
+                if (eosReceivedButNotEnqueued) {
+                    // Horrible hack. When we don't own the MediaCodec, we can't enqueue EOS so we
+                    // can't dequeue them. INFO_TRY_AGAIN_LATER is returned. We assume this means EOS.
+                    val buffer = ByteBuffer.allocateDirect(0)
+                    State.Eos(WriterData(buffer, 0L, 0) {})
+                } else {
+                    log.i("Can't dequeue output buffer: INFO_TRY_AGAIN_LATER")
+                    State.Wait
+                }
             }
             INFO_OUTPUT_FORMAT_CHANGED -> {
                 log.i("INFO_OUTPUT_FORMAT_CHANGED! format=${codec.outputFormat}")
@@ -96,6 +126,7 @@ internal class Encoder(
                     codec.releaseOutputBuffer(result, false)
                     State.Retry
                 } else {
+                    dequeuedOutputs++
                     val isEos = info.flags and BUFFER_FLAG_END_OF_STREAM != 0
                     val flags = info.flags and BUFFER_FLAG_END_OF_STREAM.inv()
                     val buffer = buffers.getOutputBuffer(result)
@@ -105,6 +136,7 @@ internal class Encoder(
                     buffer.position(info.offset)
                     val data = WriterData(buffer, timeUs, flags) {
                         codec.releaseOutputBuffer(result, false)
+                        dequeuedOutputs--
                     }
                     if (isEos) State.Eos(data) else State.Ok(data)
                 }
@@ -113,7 +145,8 @@ internal class Encoder(
     }
 
     override fun release() {
-        codec.stop()
-        codec.release()
+        if (ownsCodecStop) {
+            codec.stop()
+        }
     }
 }

lib/src/main/java/com/otaliastudios/transcoder/internal/data/Reader.kt

@@ -45,7 +45,7 @@ internal class Reader(
                 State.Eos(ReaderData(chunk, id))
             }
         } else if (!source.canReadTrack(track)) {
-            log.i("Returning State.Wait because source can't read this track right now.")
+            log.i("Returning State.Wait because source can't read $track right now.")
             State.Wait
         } else {
             nextBufferOrWait { byteBuffer, id ->

lib/src/main/java/com/otaliastudios/transcoder/internal/pipeline/pipelines.kt

@@ -2,6 +2,7 @@ package com.otaliastudios.transcoder.internal.pipeline
 
 import android.media.MediaFormat
 import com.otaliastudios.transcoder.common.TrackType
+import com.otaliastudios.transcoder.internal.Codecs
 import com.otaliastudios.transcoder.internal.audio.AudioEngine
 import com.otaliastudios.transcoder.internal.data.*
 import com.otaliastudios.transcoder.internal.data.Reader
@@ -38,27 +39,29 @@ internal fun RegularPipeline(
         sink: DataSink,
         interpolator: TimeInterpolator,
         format: MediaFormat,
+        codecs: Codecs,
         videoRotation: Int,
         audioStretcher: AudioStretcher,
         audioResampler: AudioResampler
 ) = when (track) {
-    TrackType.VIDEO -> VideoPipeline(source, sink, interpolator, format, videoRotation)
-    TrackType.AUDIO -> AudioPipeline(source, sink, interpolator, format, audioStretcher, audioResampler)
+    TrackType.VIDEO -> VideoPipeline(source, sink, interpolator, format, codecs, videoRotation)
+    TrackType.AUDIO -> AudioPipeline(source, sink, interpolator, format, codecs, audioStretcher, audioResampler)
 }
 
 private fun VideoPipeline(
         source: DataSource,
         sink: DataSink,
         interpolator: TimeInterpolator,
         format: MediaFormat,
+        codecs: Codecs,
         videoRotation: Int
 ) = Pipeline.build("Video") {
     Reader(source, TrackType.VIDEO) +
             Decoder(source.getTrackFormat(TrackType.VIDEO)!!, true) +
             DecoderTimer(TrackType.VIDEO, interpolator) +
             VideoRenderer(source.orientation, videoRotation, format) +
             VideoPublisher() +
-            Encoder(format) +
+            Encoder(codecs, TrackType.VIDEO) +
             Writer(sink, TrackType.VIDEO)
 }
 
@@ -67,13 +70,14 @@ private fun AudioPipeline(
         sink: DataSink,
         interpolator: TimeInterpolator,
         format: MediaFormat,
+        codecs: Codecs,
         audioStretcher: AudioStretcher,
         audioResampler: AudioResampler
 ) = Pipeline.build("Audio") {
     Reader(source, TrackType.AUDIO) +
             Decoder(source.getTrackFormat(TrackType.AUDIO)!!, true) +
-            DecoderTimer(TrackType.VIDEO, interpolator) +
+            DecoderTimer(TrackType.AUDIO, interpolator) +
             AudioEngine(audioStretcher, audioResampler, format) +
-            Encoder(format) +
+            Encoder(codecs, TrackType.AUDIO) +
             Writer(sink, TrackType.AUDIO)
 }

lib/src/main/java/com/otaliastudios/transcoder/internal/transcode/DefaultTranscodeEngine.kt

@@ -3,6 +3,8 @@ package com.otaliastudios.transcoder.internal.transcode
 import android.media.MediaFormat
 import com.otaliastudios.transcoder.common.TrackStatus
 import com.otaliastudios.transcoder.common.TrackType
+import com.otaliastudios.transcoder.internal.*
+import com.otaliastudios.transcoder.internal.Codecs
 import com.otaliastudios.transcoder.internal.DataSources
 import com.otaliastudios.transcoder.internal.Segments
 import com.otaliastudios.transcoder.internal.Timer
@@ -41,6 +43,8 @@ internal class DefaultTranscodeEngine(
 
     private val timer = Timer(interpolator, dataSources, tracks, segments.currentIndex)
 
+    private val codecs = Codecs(dataSources, tracks, segments.currentIndex)
+
     init {
         log.i("Created Tracks, Segments, Timer...")
     }
@@ -76,7 +80,7 @@ internal class DefaultTranscodeEngine(
             TrackStatus.REMOVING -> EmptyPipeline()
             TrackStatus.PASS_THROUGH -> PassThroughPipeline(type, source, sink, interpolator)
             TrackStatus.COMPRESSING -> RegularPipeline(type,
-                    source, sink, interpolator, outputFormat,
+                    source, sink, interpolator, outputFormat, codecs,
                     videoRotation, audioStretcher, audioResampler)
         }
     }
@@ -132,6 +136,7 @@ internal class DefaultTranscodeEngine(
         runCatching { segments.release() }
         runCatching { dataSink.release() }
         runCatching { dataSources.release() }
+        runCatching { codecs.release() }
     }
 
 

lib/src/main/java/com/otaliastudios/transcoder/internal/utils/eos.kt

@@ -20,8 +20,10 @@ private class EosIgnoringDataSink(
     override fun writeTrack(type: TrackType, byteBuffer: ByteBuffer, bufferInfo: MediaCodec.BufferInfo) {
         if (ignore()) {
             val flags = bufferInfo.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM.inv()
-            info.set(bufferInfo.offset, bufferInfo.size, bufferInfo.presentationTimeUs, flags)
-            sink.writeTrack(type, byteBuffer, info)
+            if (bufferInfo.size > 0 || flags != 0) {
+                info.set(bufferInfo.offset, bufferInfo.size, bufferInfo.presentationTimeUs, flags)
+                sink.writeTrack(type, byteBuffer, info)
+            }
         } else {
             sink.writeTrack(type, byteBuffer, bufferInfo)
         }

lib/src/main/java/com/otaliastudios/transcoder/internal/video/VideoPublisher.kt

@@ -19,7 +19,7 @@ internal class VideoPublisher: Step<Long, Channel, EncoderData, EncoderChannel>
 
     override fun initialize(next: EncoderChannel) {
         super.initialize(next)
-        surface = EglWindowSurface(core, next.surface!!, true)
+        surface = EglWindowSurface(core, next.surface!!, false)
         surface.makeCurrent()
     }