media/module/mpeg2ts/AnotherPacketSource.cpp - LeafOS-Project/android_frameworks_av - Gitiles

 /*
  * Copyright (C) 2010 The Android Open Source Project
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 //#define LOG_NDEBUG 0
 #define LOG_TAG "AnotherPacketSource"

 #include "AnotherPacketSource.h"

 #include <media/stagefright/foundation/ABuffer.h>
 #include <media/stagefright/foundation/ADebug.h>
 #include <media/stagefright/foundation/AMessage.h>
 #include <media/stagefright/foundation/AString.h>
 #include <media/stagefright/foundation/hexdump.h>
 #include <media/stagefright/foundation/avc_utils.h>
 #include <media/stagefright/MediaBuffer.h>
 #include <media/stagefright/MediaDefs.h>
 #include <media/stagefright/MetaData.h>
 #include <media/stagefright/Utils.h>
 #include <utils/Vector.h>

 #include <inttypes.h>

 namespace android {

 const int64_t kNearEOSMarkUs = 2000000LL; // 2 secs

 AnotherPacketSource::AnotherPacketSource(const sp<MetaData> &meta)
     : mIsAudio(false),
       mIsVideo(false),
       mEnabled(true),
       mFormat(NULL),
       mLastQueuedTimeUs(0),
       mEstimatedBufferDurationUs(-1),
       mEOSResult(OK),
       mLatestEnqueuedMeta(NULL),
       mLatestDequeuedMeta(NULL) {
     setFormat(meta);

     mDiscontinuitySegments.push_back(DiscontinuitySegment());
 }

 void AnotherPacketSource::setFormat(const sp<MetaData> &meta) {
     if (mFormat != NULL) {
         // Only allowed to be set once. Requires explicit clear to reset.
         return;
     }

     mIsAudio = false;
     mIsVideo = false;
     const char *mime;

     // Do not use meta if no mime.
     if (meta == NULL || !meta->findCString(kKeyMIMEType, &mime)) {
         return;
     }

     mFormat = meta;

     if (!strncasecmp("audio/", mime, 6)) {
         mIsAudio = true;
     } else if (!strncasecmp("video/", mime, 6)) {
         mIsVideo = true;
     } else if (!strncasecmp("text/", mime, 5) || !strncasecmp("application/", mime, 12)) {
         return;
     } else {
         ALOGW("Unsupported mime type: %s", mime);
     }
 }

 AnotherPacketSource::~AnotherPacketSource() {
 }

 status_t AnotherPacketSource::start(MetaData * /* params */) {
     return OK;
 }

 status_t AnotherPacketSource::stop() {
     return OK;
 }

 sp<MetaData> AnotherPacketSource::getFormat() {
     Mutex::Autolock autoLock(mLock);
     if (mFormat != NULL) {
         return mFormat;
     }

     List<sp<ABuffer> >::iterator it = mBuffers.begin();
     while (it != mBuffers.end()) {
         sp<ABuffer> buffer = *it;
         int32_t discontinuity;
         if (!buffer->meta()->findInt32("discontinuity", &discontinuity)) {
             sp<RefBase> object;
             if (buffer->meta()->findObject("format", &object)) {
                 setFormat(static_cast<MetaData*>(object.get()));
                 return mFormat;
             }
         }

         ++it;
     }
     return NULL;
 }

 status_t AnotherPacketSource::dequeueAccessUnit(sp<ABuffer> *buffer) {
     buffer->clear();

     Mutex::Autolock autoLock(mLock);
     while (mEOSResult == OK && mBuffers.empty()) {
         mCondition.wait(mLock);
     }

     if (!mBuffers.empty()) {
         *buffer = *mBuffers.begin();
         mBuffers.erase(mBuffers.begin());

         int32_t discontinuity;
         if ((*buffer)->meta()->findInt32("discontinuity", &discontinuity)) {
             if (wasFormatChange(discontinuity)) {
                 mFormat.clear();
             }

             mDiscontinuitySegments.erase(mDiscontinuitySegments.begin());
             // CHECK(!mDiscontinuitySegments.empty());
             return INFO_DISCONTINUITY;
         }

         // CHECK(!mDiscontinuitySegments.empty());
         DiscontinuitySegment &seg = *mDiscontinuitySegments.begin();

         int64_t timeUs;
         mLatestDequeuedMeta = (*buffer)->meta()->dup();
         CHECK(mLatestDequeuedMeta->findInt64("timeUs", &timeUs));
         if (timeUs > seg.mMaxDequeTimeUs) {
             seg.mMaxDequeTimeUs = timeUs;
         }

         sp<RefBase> object;
         if ((*buffer)->meta()->findObject("format", &object)) {
             setFormat(static_cast<MetaData*>(object.get()));
         }

         return OK;
     }

     return mEOSResult;
 }

 void AnotherPacketSource::requeueAccessUnit(const sp<ABuffer> &buffer) {
     // TODO: update corresponding book keeping info.
     Mutex::Autolock autoLock(mLock);
     mBuffers.push_front(buffer);
 }

 status_t AnotherPacketSource::read(
         MediaBufferBase **out, const ReadOptions *) {
     *out = NULL;

     Mutex::Autolock autoLock(mLock);
     while (mEOSResult == OK && mBuffers.empty()) {
         mCondition.wait(mLock);
     }

     if (!mBuffers.empty()) {

         const sp<ABuffer> buffer = *mBuffers.begin();
         mBuffers.erase(mBuffers.begin());

         int32_t discontinuity;
         if (buffer->meta()->findInt32("discontinuity", &discontinuity)) {
             if (wasFormatChange(discontinuity)) {
                 mFormat.clear();
             }

             mDiscontinuitySegments.erase(mDiscontinuitySegments.begin());
             // CHECK(!mDiscontinuitySegments.empty());
             return INFO_DISCONTINUITY;
         }

         mLatestDequeuedMeta = buffer->meta()->dup();

         sp<RefBase> object;
         if (buffer->meta()->findObject("format", &object)) {
             setFormat(static_cast<MetaData*>(object.get()));
         }

         int64_t timeUs;
         CHECK(buffer->meta()->findInt64("timeUs", &timeUs));
         // CHECK(!mDiscontinuitySegments.empty());
         DiscontinuitySegment &seg = *mDiscontinuitySegments.begin();
         if (timeUs > seg.mMaxDequeTimeUs) {
             seg.mMaxDequeTimeUs = timeUs;
         }

         MediaBufferBase *mediaBuffer = new MediaBuffer(buffer);
         MetaDataBase &bufmeta = mediaBuffer->meta_data();

         bufmeta.setInt64(kKeyTime, timeUs);

         int32_t isSync;
         if (buffer->meta()->findInt32("isSync", &isSync)) {
             bufmeta.setInt32(kKeyIsSyncFrame, isSync);
         }

         sp<ABuffer> sei;
         if (buffer->meta()->findBuffer("sei", &sei) && sei != NULL) {
             bufmeta.setData(kKeySEI, 0, sei->data(), sei->size());
         }

         sp<ABuffer> mpegUserData;
         if (buffer->meta()->findBuffer("mpeg-user-data", &mpegUserData) && mpegUserData != NULL) {
             bufmeta.setData(
                     kKeyMpegUserData, 0, mpegUserData->data(), mpegUserData->size());
         }

         sp<ABuffer> ap;
         if (buffer->meta()->findBuffer("audio-presentation-info", &ap) && ap != NULL) {
             bufmeta.setData(
                     kKeyAudioPresentationInfo, 0, ap->data(), ap->size());
         }

         int32_t cryptoMode;
         if (buffer->meta()->findInt32("cryptoMode", &cryptoMode)) {
             int32_t cryptoKey;
             int32_t pesOffset;
             sp<ABuffer> clearBytesBuffer, encBytesBuffer;

             CHECK(buffer->meta()->findInt32("cryptoKey", &cryptoKey));
             CHECK(buffer->meta()->findBuffer("clearBytes", &clearBytesBuffer)
                     && clearBytesBuffer != NULL);
             CHECK(buffer->meta()->findBuffer("encBytes", &encBytesBuffer)
                     && encBytesBuffer != NULL);
             CHECK(buffer->meta()->findInt32("pesOffset", &pesOffset)
                     && (pesOffset >= 0) && (pesOffset < 65536));

             bufmeta.setInt32(kKeyCryptoMode, cryptoMode);

             uint8_t array[16] = {0};
             bufmeta.setData(kKeyCryptoIV, 0, array, 16);

             array[0] = (uint8_t) (cryptoKey & 0xff);
             // array[1] contains PES header flag, which we don't use.
             // array[2~3] contain the PES offset.
             array[2] = (uint8_t) (pesOffset & 0xff);
             array[3] = (uint8_t) ((pesOffset >> 8) & 0xff);

             bufmeta.setData(kKeyCryptoKey, 0, array, 16);

             bufmeta.setData(kKeyPlainSizes, 0,
                     clearBytesBuffer->data(), clearBytesBuffer->size());

             bufmeta.setData(kKeyEncryptedSizes, 0,
                     encBytesBuffer->data(), encBytesBuffer->size());
         }


         *out = mediaBuffer;
         return OK;
     }

     return mEOSResult;
 }

 bool AnotherPacketSource::wasFormatChange(
         int32_t discontinuityType) const {
     if (mIsAudio) {
         return (discontinuityType & ATSParser::DISCONTINUITY_AUDIO_FORMAT) != 0;
     }

     if (mIsVideo) {
         return (discontinuityType & ATSParser::DISCONTINUITY_VIDEO_FORMAT) != 0;
     }

     return false;
 }

 void AnotherPacketSource::queueAccessUnit(const sp<ABuffer> &buffer) {
     int32_t damaged;
     if (buffer->meta()->findInt32("damaged", &damaged) && damaged) {
         // LOG(VERBOSE) << "discarding damaged AU";
         return;
     }

     Mutex::Autolock autoLock(mLock);
     mBuffers.push_back(buffer);
     mCondition.signal();

     int32_t discontinuity;
     if (buffer->meta()->findInt32("discontinuity", &discontinuity)){
         ALOGV("queueing a discontinuity with queueAccessUnit");

         mLastQueuedTimeUs = 0LL;
         mEOSResult = OK;
         mLatestEnqueuedMeta = NULL;

         mDiscontinuitySegments.push_back(DiscontinuitySegment());
         return;
     }

     int64_t lastQueuedTimeUs;
     CHECK(buffer->meta()->findInt64("timeUs", &lastQueuedTimeUs));
     mLastQueuedTimeUs = lastQueuedTimeUs;
     ALOGV("queueAccessUnit timeUs=%" PRIi64 " us (%.2f secs)",
             mLastQueuedTimeUs, mLastQueuedTimeUs / 1E6);

     // CHECK(!mDiscontinuitySegments.empty());
     DiscontinuitySegment &tailSeg = *(--mDiscontinuitySegments.end());
     if (lastQueuedTimeUs > tailSeg.mMaxEnqueTimeUs) {
         tailSeg.mMaxEnqueTimeUs = lastQueuedTimeUs;
     }
     if (tailSeg.mMaxDequeTimeUs == -1) {
         tailSeg.mMaxDequeTimeUs = lastQueuedTimeUs;
     }

     if (mLatestEnqueuedMeta == NULL) {
         mLatestEnqueuedMeta = buffer->meta()->dup();
     } else {
         int64_t latestTimeUs = 0;
         int64_t frameDeltaUs = 0;
         CHECK(mLatestEnqueuedMeta->findInt64("timeUs", &latestTimeUs));
         if (lastQueuedTimeUs > latestTimeUs) {
             mLatestEnqueuedMeta = buffer->meta()->dup();
             frameDeltaUs = lastQueuedTimeUs - latestTimeUs;
             mLatestEnqueuedMeta->setInt64("durationUs", frameDeltaUs);
         } else if (!mLatestEnqueuedMeta->findInt64("durationUs", &frameDeltaUs)) {
             // For B frames
             frameDeltaUs = latestTimeUs - lastQueuedTimeUs;
             mLatestEnqueuedMeta->setInt64("durationUs", frameDeltaUs);
         }
     }
 }

 void AnotherPacketSource::clear() {
     Mutex::Autolock autoLock(mLock);

     mBuffers.clear();
     mEOSResult = OK;

     mDiscontinuitySegments.clear();
     mDiscontinuitySegments.push_back(DiscontinuitySegment());

     mFormat = NULL;
     mLatestEnqueuedMeta = NULL;

     mEstimatedBufferDurationUs = -1;
 }

 void AnotherPacketSource::queueDiscontinuity(
         ATSParser::DiscontinuityType type,
         const sp<AMessage> &extra,
         bool discard) {
     Mutex::Autolock autoLock(mLock);

     if (discard) {
         // Leave only discontinuities in the queue.
         List<sp<ABuffer> >::iterator it = mBuffers.begin();
         while (it != mBuffers.end()) {
             sp<ABuffer> oldBuffer = *it;

             int32_t oldDiscontinuityType;
             if (!oldBuffer->meta()->findInt32(
                         "discontinuity", &oldDiscontinuityType)) {
                 it = mBuffers.erase(it);
                 continue;
             }

             ++it;
         }

         for (List<DiscontinuitySegment>::iterator it2 = mDiscontinuitySegments.begin();
                 it2 != mDiscontinuitySegments.end();
                 ++it2) {
             DiscontinuitySegment &seg = *it2;
             seg.clear();
         }

     }

     mEOSResult = OK;
     mLastQueuedTimeUs = 0;
     mLatestEnqueuedMeta = NULL;

     if (type == ATSParser::DISCONTINUITY_NONE) {
         return;
     }

     mDiscontinuitySegments.push_back(DiscontinuitySegment());

     sp<ABuffer> buffer = new ABuffer(0);
     buffer->meta()->setInt32("discontinuity", static_cast<int32_t>(type));
     buffer->meta()->setMessage("extra", extra);

     mBuffers.push_back(buffer);
     mCondition.signal();
 }

 void AnotherPacketSource::signalEOS(status_t result) {
     CHECK(result != OK);

     Mutex::Autolock autoLock(mLock);
     mEOSResult = result;
     mCondition.signal();
 }

 bool AnotherPacketSource::hasBufferAvailable(status_t *finalResult) {
     Mutex::Autolock autoLock(mLock);
     *finalResult = OK;
     if (!mEnabled) {
         return false;
     }
     if (!mBuffers.empty()) {
         return true;
     }

     *finalResult = mEOSResult;
     return false;
 }

 bool AnotherPacketSource::hasDataBufferAvailable(status_t *finalResult) {
     Mutex::Autolock autoLock(mLock);
     *finalResult = OK;
     if (!mEnabled) {
         return false;
     }
     List<sp<ABuffer> >::iterator it;
     for (it = mBuffers.begin(); it != mBuffers.end(); it++) {
         int32_t discontinuity;
         if (!(*it)->meta()->findInt32("discontinuity", &discontinuity)) {
             return true;
         }
     }

     *finalResult = mEOSResult;
     return false;
 }

 size_t AnotherPacketSource::getAvailableBufferCount(status_t *finalResult) {
     Mutex::Autolock autoLock(mLock);

     *finalResult = OK;
     if (!mEnabled) {
         return 0;
     }
     if (!mBuffers.empty()) {
         return mBuffers.size();
     }
     *finalResult = mEOSResult;
     return 0;
 }

 int64_t AnotherPacketSource::getBufferedDurationUs(status_t *finalResult) {
     Mutex::Autolock autoLock(mLock);
     *finalResult = mEOSResult;

     int64_t durationUs = 0;
     for (List<DiscontinuitySegment>::iterator it = mDiscontinuitySegments.begin();
             it != mDiscontinuitySegments.end();
             ++it) {
         const DiscontinuitySegment &seg = *it;
         // dequeued access units should be a subset of enqueued access units
         // CHECK(seg.maxEnqueTimeUs >= seg.mMaxDequeTimeUs);
         durationUs += (seg.mMaxEnqueTimeUs - seg.mMaxDequeTimeUs);
     }

     return durationUs;
 }

 int64_t AnotherPacketSource::getEstimatedBufferDurationUs() {
     Mutex::Autolock autoLock(mLock);
     if (mEstimatedBufferDurationUs >= 0) {
         return mEstimatedBufferDurationUs;
     }

     SortedVector<int64_t> maxTimesUs;
     List<sp<ABuffer> >::iterator it;
     int64_t t1 = 0, t2 = 0;
     for (it = mBuffers.begin(); it != mBuffers.end(); ++it) {
         int64_t timeUs = 0;
         const sp<ABuffer> &buffer = *it;
         if (!buffer->meta()->findInt64("timeUs", &timeUs)) {
             continue;
         }
         maxTimesUs.add(timeUs);
         while (maxTimesUs.size() > 2) {
             maxTimesUs.removeAt(0);
             t1 = maxTimesUs.itemAt(0);
             t2 = maxTimesUs.itemAt(1);
         }
     }
     return mEstimatedBufferDurationUs = t2 - t1;
 }

 status_t AnotherPacketSource::nextBufferTime(int64_t *timeUs) {
     *timeUs = 0;

     Mutex::Autolock autoLock(mLock);

     if (mBuffers.empty()) {
         return mEOSResult != OK ? mEOSResult : -EWOULDBLOCK;
     }

     sp<ABuffer> buffer = *mBuffers.begin();
     CHECK(buffer->meta()->findInt64("timeUs", timeUs));

     return OK;
 }

 bool AnotherPacketSource::isFinished(int64_t duration) const {
     if (duration > 0) {
         int64_t diff = duration - mLastQueuedTimeUs;
         if (diff < kNearEOSMarkUs && diff > -kNearEOSMarkUs) {
             ALOGV("Detecting EOS due to near end");
             return true;
         }
     }
     return (mEOSResult != OK);
 }

 sp<AMessage> AnotherPacketSource::getLatestEnqueuedMeta() {
     Mutex::Autolock autoLock(mLock);
     return mLatestEnqueuedMeta;
 }

 sp<AMessage> AnotherPacketSource::getLatestDequeuedMeta() {
     Mutex::Autolock autoLock(mLock);
     return mLatestDequeuedMeta;
 }

 void AnotherPacketSource::enable(bool enable) {
     Mutex::Autolock autoLock(mLock);
     mEnabled = enable;
 }

 /*
  * returns the sample meta that's delayUs after queue head
  * (NULL if such sample is unavailable)
  */
 sp<AMessage> AnotherPacketSource::getMetaAfterLastDequeued(int64_t delayUs) {
     Mutex::Autolock autoLock(mLock);
     int64_t firstUs = -1;
     int64_t lastUs = -1;
     int64_t durationUs = 0;

     List<sp<ABuffer> >::iterator it;
     for (it = mBuffers.begin(); it != mBuffers.end(); ++it) {
         const sp<ABuffer> &buffer = *it;
         int32_t discontinuity;
         if (buffer->meta()->findInt32("discontinuity", &discontinuity)) {
             durationUs += lastUs - firstUs;
             firstUs = -1;
             lastUs = -1;
             continue;
         }
         int64_t timeUs;
         if (buffer->meta()->findInt64("timeUs", &timeUs)) {
             if (firstUs < 0) {
                 firstUs = timeUs;
             }
             if (lastUs < 0 || timeUs > lastUs) {
                 lastUs = timeUs;
             }
             if (durationUs + (lastUs - firstUs) >= delayUs) {
                 return buffer->meta();
             }
         }
     }
     return NULL;
 }

 /*
  * removes samples with time equal or after meta
  */
 void AnotherPacketSource::trimBuffersAfterMeta(
         const sp<AMessage> &meta) {
     if (meta == NULL) {
         ALOGW("trimming with NULL meta, ignoring");
         return;
     }

     Mutex::Autolock autoLock(mLock);
     if (mBuffers.empty()) {
         return;
     }

     HLSTime stopTime(meta);
     ALOGV("trimBuffersAfterMeta: discontinuitySeq %d, timeUs %lld",
             stopTime.mSeq, (long long)stopTime.mTimeUs);

     List<sp<ABuffer> >::iterator it;
     List<DiscontinuitySegment >::iterator it2;
     sp<AMessage> newLatestEnqueuedMeta = NULL;
     int64_t newLastQueuedTimeUs = 0;
     for (it = mBuffers.begin(), it2 = mDiscontinuitySegments.begin(); it != mBuffers.end(); ++it) {
         const sp<ABuffer> &buffer = *it;
         int32_t discontinuity;
         if (buffer->meta()->findInt32("discontinuity", &discontinuity)) {
             // CHECK(it2 != mDiscontinuitySegments.end());
             ++it2;
             continue;
         }

         HLSTime curTime(buffer->meta());
         if (!(curTime < stopTime)) {
             ALOGV("trimming from %lld (inclusive) to end",
                     (long long)curTime.mTimeUs);
             break;
         }
         newLatestEnqueuedMeta = buffer->meta();
         newLastQueuedTimeUs = curTime.mTimeUs;
     }

     mBuffers.erase(it, mBuffers.end());
     mLatestEnqueuedMeta = newLatestEnqueuedMeta;
     mLastQueuedTimeUs = newLastQueuedTimeUs;

     DiscontinuitySegment &seg = *it2;
     if (newLatestEnqueuedMeta != NULL) {
         seg.mMaxEnqueTimeUs = newLastQueuedTimeUs;
     } else {
         seg.clear();
     }
     mDiscontinuitySegments.erase(++it2, mDiscontinuitySegments.end());
 }

 /*
  * removes samples with time equal or before meta;
  * returns first sample left in the queue.
  *
  * (for AVC, if trim happens, the samples left will always start
  * at next IDR.)
  */
 sp<AMessage> AnotherPacketSource::trimBuffersBeforeMeta(
         const sp<AMessage> &meta) {
     HLSTime startTime(meta);
     ALOGV("trimBuffersBeforeMeta: discontinuitySeq %d, timeUs %lld",
             startTime.mSeq, (long long)startTime.mTimeUs);

     sp<AMessage> firstMeta;
     int64_t firstTimeUs = -1;
     Mutex::Autolock autoLock(mLock);
     if (mBuffers.empty()) {
         return NULL;
     }

     sp<MetaData> format;
     bool isAvc = false;

     List<sp<ABuffer> >::iterator it;
     for (it = mBuffers.begin(); it != mBuffers.end(); ++it) {
         const sp<ABuffer> &buffer = *it;
         int32_t discontinuity;
         if (buffer->meta()->findInt32("discontinuity", &discontinuity)) {
             mDiscontinuitySegments.erase(mDiscontinuitySegments.begin());
             // CHECK(!mDiscontinuitySegments.empty());
             format = NULL;
             isAvc = false;
             continue;
         }
         if (format == NULL) {
             sp<RefBase> object;
             if (buffer->meta()->findObject("format", &object)) {
                 const char* mime;
                 format = static_cast<MetaData*>(object.get());
                 isAvc = format != NULL
                         && format->findCString(kKeyMIMEType, &mime)
                         && !strcasecmp(mime, MEDIA_MIMETYPE_VIDEO_AVC);
             }
         }
         if (isAvc && !IsIDR(buffer->data(), buffer->size())) {
             continue;
         }

         HLSTime curTime(buffer->meta());
         if (startTime < curTime) {
             ALOGV("trimming from beginning to %lld (not inclusive)",
                     (long long)curTime.mTimeUs);
             firstMeta = buffer->meta();
             firstTimeUs = curTime.mTimeUs;
             break;
         }
     }
     mBuffers.erase(mBuffers.begin(), it);
     mLatestDequeuedMeta = NULL;

     // CHECK(!mDiscontinuitySegments.empty());
     DiscontinuitySegment &seg = *mDiscontinuitySegments.begin();
     if (firstTimeUs >= 0) {
         seg.mMaxDequeTimeUs = firstTimeUs;
     } else {
         seg.clear();
     }

     return firstMeta;
 }

 }  // namespace android
	/*
	* Copyright (C) 2010 The Android Open Source Project
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	//#define LOG_NDEBUG 0
	#define LOG_TAG "AnotherPacketSource"

	#include "AnotherPacketSource.h"

	#include <media/stagefright/foundation/ABuffer.h>
	#include <media/stagefright/foundation/ADebug.h>
	#include <media/stagefright/foundation/AMessage.h>
	#include <media/stagefright/foundation/AString.h>
	#include <media/stagefright/foundation/hexdump.h>
	#include <media/stagefright/foundation/avc_utils.h>
	#include <media/stagefright/MediaBuffer.h>
	#include <media/stagefright/MediaDefs.h>
	#include <media/stagefright/MetaData.h>
	#include <media/stagefright/Utils.h>
	#include <utils/Vector.h>

	#include <inttypes.h>

	namespace android {

	const int64_t kNearEOSMarkUs = 2000000LL; // 2 secs

	AnotherPacketSource::AnotherPacketSource(const sp<MetaData> &meta)
	: mIsAudio(false),
	mIsVideo(false),
	mEnabled(true),
	mFormat(NULL),
	mLastQueuedTimeUs(0),
	mEstimatedBufferDurationUs(-1),
	mEOSResult(OK),
	mLatestEnqueuedMeta(NULL),
	mLatestDequeuedMeta(NULL) {
	setFormat(meta);

	mDiscontinuitySegments.push_back(DiscontinuitySegment());
	}

	void AnotherPacketSource::setFormat(const sp<MetaData> &meta) {
	if (mFormat != NULL) {
	// Only allowed to be set once. Requires explicit clear to reset.
	return;
	}

	mIsAudio = false;
	mIsVideo = false;
	const char *mime;

	// Do not use meta if no mime.
	if (meta == NULL \|\| !meta->findCString(kKeyMIMEType, &mime)) {
	return;
	}

	mFormat = meta;

	if (!strncasecmp("audio/", mime, 6)) {
	mIsAudio = true;
	} else if (!strncasecmp("video/", mime, 6)) {
	mIsVideo = true;
	} else if (!strncasecmp("text/", mime, 5) \|\| !strncasecmp("application/", mime, 12)) {
	return;
	} else {
	ALOGW("Unsupported mime type: %s", mime);
	}
	}

	AnotherPacketSource::~AnotherPacketSource() {
	}

	status_t AnotherPacketSource::start(MetaData * /* params */) {
	return OK;
	}

	status_t AnotherPacketSource::stop() {
	return OK;
	}

	sp<MetaData> AnotherPacketSource::getFormat() {
	Mutex::Autolock autoLock(mLock);
	if (mFormat != NULL) {
	return mFormat;
	}

	List<sp<ABuffer> >::iterator it = mBuffers.begin();
	while (it != mBuffers.end()) {
	sp<ABuffer> buffer = *it;
	int32_t discontinuity;
	if (!buffer->meta()->findInt32("discontinuity", &discontinuity)) {
	sp<RefBase> object;
	if (buffer->meta()->findObject("format", &object)) {
	setFormat(static_cast<MetaData*>(object.get()));
	return mFormat;
	}
	}

	++it;
	}
	return NULL;
	}

	status_t AnotherPacketSource::dequeueAccessUnit(sp<ABuffer> *buffer) {
	buffer->clear();

	Mutex::Autolock autoLock(mLock);
	while (mEOSResult == OK && mBuffers.empty()) {
	mCondition.wait(mLock);
	}

	if (!mBuffers.empty()) {
	buffer = mBuffers.begin();
	mBuffers.erase(mBuffers.begin());

	int32_t discontinuity;
	if ((*buffer)->meta()->findInt32("discontinuity", &discontinuity)) {
	if (wasFormatChange(discontinuity)) {
	mFormat.clear();
	}

	mDiscontinuitySegments.erase(mDiscontinuitySegments.begin());
	// CHECK(!mDiscontinuitySegments.empty());
	return INFO_DISCONTINUITY;
	}

	// CHECK(!mDiscontinuitySegments.empty());
	DiscontinuitySegment &seg = *mDiscontinuitySegments.begin();

	int64_t timeUs;
	mLatestDequeuedMeta = (*buffer)->meta()->dup();
	CHECK(mLatestDequeuedMeta->findInt64("timeUs", &timeUs));
	if (timeUs > seg.mMaxDequeTimeUs) {
	seg.mMaxDequeTimeUs = timeUs;
	}

	sp<RefBase> object;
	if ((*buffer)->meta()->findObject("format", &object)) {
	setFormat(static_cast<MetaData*>(object.get()));
	}

	return OK;
	}

	return mEOSResult;
	}

	void AnotherPacketSource::requeueAccessUnit(const sp<ABuffer> &buffer) {
	// TODO: update corresponding book keeping info.
	Mutex::Autolock autoLock(mLock);
	mBuffers.push_front(buffer);
	}

	status_t AnotherPacketSource::read(
	MediaBufferBase *out, const ReadOptions ) {
	*out = NULL;

	Mutex::Autolock autoLock(mLock);
	while (mEOSResult == OK && mBuffers.empty()) {
	mCondition.wait(mLock);
	}

	if (!mBuffers.empty()) {

	const sp<ABuffer> buffer = *mBuffers.begin();
	mBuffers.erase(mBuffers.begin());

	int32_t discontinuity;
	if (buffer->meta()->findInt32("discontinuity", &discontinuity)) {
	if (wasFormatChange(discontinuity)) {
	mFormat.clear();
	}

	mDiscontinuitySegments.erase(mDiscontinuitySegments.begin());
	// CHECK(!mDiscontinuitySegments.empty());
	return INFO_DISCONTINUITY;
	}

	mLatestDequeuedMeta = buffer->meta()->dup();

	sp<RefBase> object;
	if (buffer->meta()->findObject("format", &object)) {
	setFormat(static_cast<MetaData*>(object.get()));
	}

	int64_t timeUs;
	CHECK(buffer->meta()->findInt64("timeUs", &timeUs));
	// CHECK(!mDiscontinuitySegments.empty());
	DiscontinuitySegment &seg = *mDiscontinuitySegments.begin();
	if (timeUs > seg.mMaxDequeTimeUs) {
	seg.mMaxDequeTimeUs = timeUs;
	}

	MediaBufferBase *mediaBuffer = new MediaBuffer(buffer);
	MetaDataBase &bufmeta = mediaBuffer->meta_data();

	bufmeta.setInt64(kKeyTime, timeUs);

	int32_t isSync;
	if (buffer->meta()->findInt32("isSync", &isSync)) {
	bufmeta.setInt32(kKeyIsSyncFrame, isSync);
	}

	sp<ABuffer> sei;
	if (buffer->meta()->findBuffer("sei", &sei) && sei != NULL) {
	bufmeta.setData(kKeySEI, 0, sei->data(), sei->size());
	}

	sp<ABuffer> mpegUserData;
	if (buffer->meta()->findBuffer("mpeg-user-data", &mpegUserData) && mpegUserData != NULL) {
	bufmeta.setData(
	kKeyMpegUserData, 0, mpegUserData->data(), mpegUserData->size());
	}

	sp<ABuffer> ap;
	if (buffer->meta()->findBuffer("audio-presentation-info", &ap) && ap != NULL) {
	bufmeta.setData(
	kKeyAudioPresentationInfo, 0, ap->data(), ap->size());
	}

	int32_t cryptoMode;
	if (buffer->meta()->findInt32("cryptoMode", &cryptoMode)) {
	int32_t cryptoKey;
	int32_t pesOffset;
	sp<ABuffer> clearBytesBuffer, encBytesBuffer;

	CHECK(buffer->meta()->findInt32("cryptoKey", &cryptoKey));
	CHECK(buffer->meta()->findBuffer("clearBytes", &clearBytesBuffer)
	&& clearBytesBuffer != NULL);
	CHECK(buffer->meta()->findBuffer("encBytes", &encBytesBuffer)
	&& encBytesBuffer != NULL);
	CHECK(buffer->meta()->findInt32("pesOffset", &pesOffset)
	&& (pesOffset >= 0) && (pesOffset < 65536));

	bufmeta.setInt32(kKeyCryptoMode, cryptoMode);

	uint8_t array[16] = {0};
	bufmeta.setData(kKeyCryptoIV, 0, array, 16);

	array[0] = (uint8_t) (cryptoKey & 0xff);
	// array[1] contains PES header flag, which we don't use.
	// array[2~3] contain the PES offset.
	array[2] = (uint8_t) (pesOffset & 0xff);
	array[3] = (uint8_t) ((pesOffset >> 8) & 0xff);

	bufmeta.setData(kKeyCryptoKey, 0, array, 16);

	bufmeta.setData(kKeyPlainSizes, 0,
	clearBytesBuffer->data(), clearBytesBuffer->size());

	bufmeta.setData(kKeyEncryptedSizes, 0,
	encBytesBuffer->data(), encBytesBuffer->size());
	}


	*out = mediaBuffer;
	return OK;
	}

	return mEOSResult;
	}

	bool AnotherPacketSource::wasFormatChange(
	int32_t discontinuityType) const {
	if (mIsAudio) {
	return (discontinuityType & ATSParser::DISCONTINUITY_AUDIO_FORMAT) != 0;
	}

	if (mIsVideo) {
	return (discontinuityType & ATSParser::DISCONTINUITY_VIDEO_FORMAT) != 0;
	}

	return false;
	}

	void AnotherPacketSource::queueAccessUnit(const sp<ABuffer> &buffer) {
	int32_t damaged;
	if (buffer->meta()->findInt32("damaged", &damaged) && damaged) {
	// LOG(VERBOSE) << "discarding damaged AU";
	return;
	}

	Mutex::Autolock autoLock(mLock);
	mBuffers.push_back(buffer);
	mCondition.signal();

	int32_t discontinuity;
	if (buffer->meta()->findInt32("discontinuity", &discontinuity)){
	ALOGV("queueing a discontinuity with queueAccessUnit");

	mLastQueuedTimeUs = 0LL;
	mEOSResult = OK;
	mLatestEnqueuedMeta = NULL;

	mDiscontinuitySegments.push_back(DiscontinuitySegment());
	return;
	}

	int64_t lastQueuedTimeUs;
	CHECK(buffer->meta()->findInt64("timeUs", &lastQueuedTimeUs));
	mLastQueuedTimeUs = lastQueuedTimeUs;
	ALOGV("queueAccessUnit timeUs=%" PRIi64 " us (%.2f secs)",
	mLastQueuedTimeUs, mLastQueuedTimeUs / 1E6);

	// CHECK(!mDiscontinuitySegments.empty());
	DiscontinuitySegment &tailSeg = *(--mDiscontinuitySegments.end());
	if (lastQueuedTimeUs > tailSeg.mMaxEnqueTimeUs) {
	tailSeg.mMaxEnqueTimeUs = lastQueuedTimeUs;
	}
	if (tailSeg.mMaxDequeTimeUs == -1) {
	tailSeg.mMaxDequeTimeUs = lastQueuedTimeUs;
	}

	if (mLatestEnqueuedMeta == NULL) {
	mLatestEnqueuedMeta = buffer->meta()->dup();
	} else {
	int64_t latestTimeUs = 0;
	int64_t frameDeltaUs = 0;
	CHECK(mLatestEnqueuedMeta->findInt64("timeUs", &latestTimeUs));
	if (lastQueuedTimeUs > latestTimeUs) {
	mLatestEnqueuedMeta = buffer->meta()->dup();
	frameDeltaUs = lastQueuedTimeUs - latestTimeUs;
	mLatestEnqueuedMeta->setInt64("durationUs", frameDeltaUs);
	} else if (!mLatestEnqueuedMeta->findInt64("durationUs", &frameDeltaUs)) {
	// For B frames
	frameDeltaUs = latestTimeUs - lastQueuedTimeUs;
	mLatestEnqueuedMeta->setInt64("durationUs", frameDeltaUs);
	}
	}
	}

	void AnotherPacketSource::clear() {
	Mutex::Autolock autoLock(mLock);

	mBuffers.clear();
	mEOSResult = OK;

	mDiscontinuitySegments.clear();
	mDiscontinuitySegments.push_back(DiscontinuitySegment());

	mFormat = NULL;
	mLatestEnqueuedMeta = NULL;

	mEstimatedBufferDurationUs = -1;
	}

	void AnotherPacketSource::queueDiscontinuity(
	ATSParser::DiscontinuityType type,
	const sp<AMessage> &extra,
	bool discard) {
	Mutex::Autolock autoLock(mLock);

	if (discard) {
	// Leave only discontinuities in the queue.
	List<sp<ABuffer> >::iterator it = mBuffers.begin();
	while (it != mBuffers.end()) {
	sp<ABuffer> oldBuffer = *it;

	int32_t oldDiscontinuityType;
	if (!oldBuffer->meta()->findInt32(
	"discontinuity", &oldDiscontinuityType)) {
	it = mBuffers.erase(it);
	continue;
	}

	++it;
	}

	for (List<DiscontinuitySegment>::iterator it2 = mDiscontinuitySegments.begin();
	it2 != mDiscontinuitySegments.end();
	++it2) {
	DiscontinuitySegment &seg = *it2;
	seg.clear();
	}

	}

	mEOSResult = OK;
	mLastQueuedTimeUs = 0;
	mLatestEnqueuedMeta = NULL;

	if (type == ATSParser::DISCONTINUITY_NONE) {
	return;
	}

	mDiscontinuitySegments.push_back(DiscontinuitySegment());

	sp<ABuffer> buffer = new ABuffer(0);
	buffer->meta()->setInt32("discontinuity", static_cast<int32_t>(type));
	buffer->meta()->setMessage("extra", extra);

	mBuffers.push_back(buffer);
	mCondition.signal();
	}

	void AnotherPacketSource::signalEOS(status_t result) {
	CHECK(result != OK);

	Mutex::Autolock autoLock(mLock);
	mEOSResult = result;
	mCondition.signal();
	}

	bool AnotherPacketSource::hasBufferAvailable(status_t *finalResult) {
	Mutex::Autolock autoLock(mLock);
	*finalResult = OK;
	if (!mEnabled) {
	return false;
	}
	if (!mBuffers.empty()) {
	return true;
	}

	*finalResult = mEOSResult;
	return false;
	}

	bool AnotherPacketSource::hasDataBufferAvailable(status_t *finalResult) {
	Mutex::Autolock autoLock(mLock);
	*finalResult = OK;
	if (!mEnabled) {
	return false;
	}
	List<sp<ABuffer> >::iterator it;
	for (it = mBuffers.begin(); it != mBuffers.end(); it++) {
	int32_t discontinuity;
	if (!(*it)->meta()->findInt32("discontinuity", &discontinuity)) {
	return true;
	}
	}

	*finalResult = mEOSResult;
	return false;
	}

	size_t AnotherPacketSource::getAvailableBufferCount(status_t *finalResult) {
	Mutex::Autolock autoLock(mLock);

	*finalResult = OK;
	if (!mEnabled) {
	return 0;
	}
	if (!mBuffers.empty()) {
	return mBuffers.size();
	}
	*finalResult = mEOSResult;
	return 0;
	}

	int64_t AnotherPacketSource::getBufferedDurationUs(status_t *finalResult) {
	Mutex::Autolock autoLock(mLock);
	*finalResult = mEOSResult;

	int64_t durationUs = 0;
	for (List<DiscontinuitySegment>::iterator it = mDiscontinuitySegments.begin();
	it != mDiscontinuitySegments.end();
	++it) {
	const DiscontinuitySegment &seg = *it;
	// dequeued access units should be a subset of enqueued access units
	// CHECK(seg.maxEnqueTimeUs >= seg.mMaxDequeTimeUs);
	durationUs += (seg.mMaxEnqueTimeUs - seg.mMaxDequeTimeUs);
	}

	return durationUs;
	}

	int64_t AnotherPacketSource::getEstimatedBufferDurationUs() {
	Mutex::Autolock autoLock(mLock);
	if (mEstimatedBufferDurationUs >= 0) {
	return mEstimatedBufferDurationUs;
	}

	SortedVector<int64_t> maxTimesUs;
	List<sp<ABuffer> >::iterator it;
	int64_t t1 = 0, t2 = 0;
	for (it = mBuffers.begin(); it != mBuffers.end(); ++it) {
	int64_t timeUs = 0;
	const sp<ABuffer> &buffer = *it;
	if (!buffer->meta()->findInt64("timeUs", &timeUs)) {
	continue;
	}
	maxTimesUs.add(timeUs);
	while (maxTimesUs.size() > 2) {
	maxTimesUs.removeAt(0);
	t1 = maxTimesUs.itemAt(0);
	t2 = maxTimesUs.itemAt(1);
	}
	}
	return mEstimatedBufferDurationUs = t2 - t1;
	}

	status_t AnotherPacketSource::nextBufferTime(int64_t *timeUs) {
	*timeUs = 0;

	Mutex::Autolock autoLock(mLock);

	if (mBuffers.empty()) {
	return mEOSResult != OK ? mEOSResult : -EWOULDBLOCK;
	}

	sp<ABuffer> buffer = *mBuffers.begin();
	CHECK(buffer->meta()->findInt64("timeUs", timeUs));

	return OK;
	}

	bool AnotherPacketSource::isFinished(int64_t duration) const {
	if (duration > 0) {
	int64_t diff = duration - mLastQueuedTimeUs;
	if (diff < kNearEOSMarkUs && diff > -kNearEOSMarkUs) {
	ALOGV("Detecting EOS due to near end");
	return true;
	}
	}
	return (mEOSResult != OK);
	}

	sp<AMessage> AnotherPacketSource::getLatestEnqueuedMeta() {
	Mutex::Autolock autoLock(mLock);
	return mLatestEnqueuedMeta;
	}

	sp<AMessage> AnotherPacketSource::getLatestDequeuedMeta() {
	Mutex::Autolock autoLock(mLock);
	return mLatestDequeuedMeta;
	}

	void AnotherPacketSource::enable(bool enable) {
	Mutex::Autolock autoLock(mLock);
	mEnabled = enable;
	}

	/*
	* returns the sample meta that's delayUs after queue head
	* (NULL if such sample is unavailable)
	*/
	sp<AMessage> AnotherPacketSource::getMetaAfterLastDequeued(int64_t delayUs) {
	Mutex::Autolock autoLock(mLock);
	int64_t firstUs = -1;
	int64_t lastUs = -1;
	int64_t durationUs = 0;

	List<sp<ABuffer> >::iterator it;
	for (it = mBuffers.begin(); it != mBuffers.end(); ++it) {
	const sp<ABuffer> &buffer = *it;
	int32_t discontinuity;
	if (buffer->meta()->findInt32("discontinuity", &discontinuity)) {
	durationUs += lastUs - firstUs;
	firstUs = -1;
	lastUs = -1;
	continue;
	}
	int64_t timeUs;
	if (buffer->meta()->findInt64("timeUs", &timeUs)) {
	if (firstUs < 0) {
	firstUs = timeUs;
	}
	if (lastUs < 0 \|\| timeUs > lastUs) {
	lastUs = timeUs;
	}
	if (durationUs + (lastUs - firstUs) >= delayUs) {
	return buffer->meta();
	}
	}
	}
	return NULL;
	}

	/*
	* removes samples with time equal or after meta
	*/
	void AnotherPacketSource::trimBuffersAfterMeta(
	const sp<AMessage> &meta) {
	if (meta == NULL) {
	ALOGW("trimming with NULL meta, ignoring");
	return;
	}

	Mutex::Autolock autoLock(mLock);
	if (mBuffers.empty()) {
	return;
	}

	HLSTime stopTime(meta);
	ALOGV("trimBuffersAfterMeta: discontinuitySeq %d, timeUs %lld",
	stopTime.mSeq, (long long)stopTime.mTimeUs);

	List<sp<ABuffer> >::iterator it;
	List<DiscontinuitySegment >::iterator it2;
	sp<AMessage> newLatestEnqueuedMeta = NULL;
	int64_t newLastQueuedTimeUs = 0;
	for (it = mBuffers.begin(), it2 = mDiscontinuitySegments.begin(); it != mBuffers.end(); ++it) {
	const sp<ABuffer> &buffer = *it;
	int32_t discontinuity;
	if (buffer->meta()->findInt32("discontinuity", &discontinuity)) {
	// CHECK(it2 != mDiscontinuitySegments.end());
	++it2;
	continue;
	}

	HLSTime curTime(buffer->meta());
	if (!(curTime < stopTime)) {
	ALOGV("trimming from %lld (inclusive) to end",
	(long long)curTime.mTimeUs);
	break;
	}
	newLatestEnqueuedMeta = buffer->meta();
	newLastQueuedTimeUs = curTime.mTimeUs;
	}

	mBuffers.erase(it, mBuffers.end());
	mLatestEnqueuedMeta = newLatestEnqueuedMeta;
	mLastQueuedTimeUs = newLastQueuedTimeUs;

	DiscontinuitySegment &seg = *it2;
	if (newLatestEnqueuedMeta != NULL) {
	seg.mMaxEnqueTimeUs = newLastQueuedTimeUs;
	} else {
	seg.clear();
	}
	mDiscontinuitySegments.erase(++it2, mDiscontinuitySegments.end());
	}

	/*
	* removes samples with time equal or before meta;
	* returns first sample left in the queue.
	*
	* (for AVC, if trim happens, the samples left will always start
	* at next IDR.)
	*/
	sp<AMessage> AnotherPacketSource::trimBuffersBeforeMeta(
	const sp<AMessage> &meta) {
	HLSTime startTime(meta);
	ALOGV("trimBuffersBeforeMeta: discontinuitySeq %d, timeUs %lld",
	startTime.mSeq, (long long)startTime.mTimeUs);

	sp<AMessage> firstMeta;
	int64_t firstTimeUs = -1;
	Mutex::Autolock autoLock(mLock);
	if (mBuffers.empty()) {
	return NULL;
	}

	sp<MetaData> format;
	bool isAvc = false;

	List<sp<ABuffer> >::iterator it;
	for (it = mBuffers.begin(); it != mBuffers.end(); ++it) {
	const sp<ABuffer> &buffer = *it;
	int32_t discontinuity;
	if (buffer->meta()->findInt32("discontinuity", &discontinuity)) {
	mDiscontinuitySegments.erase(mDiscontinuitySegments.begin());
	// CHECK(!mDiscontinuitySegments.empty());
	format = NULL;
	isAvc = false;
	continue;
	}
	if (format == NULL) {
	sp<RefBase> object;
	if (buffer->meta()->findObject("format", &object)) {
	const char* mime;
	format = static_cast<MetaData*>(object.get());
	isAvc = format != NULL
	&& format->findCString(kKeyMIMEType, &mime)
	&& !strcasecmp(mime, MEDIA_MIMETYPE_VIDEO_AVC);
	}
	}
	if (isAvc && !IsIDR(buffer->data(), buffer->size())) {
	continue;
	}

	HLSTime curTime(buffer->meta());
	if (startTime < curTime) {
	ALOGV("trimming from beginning to %lld (not inclusive)",
	(long long)curTime.mTimeUs);
	firstMeta = buffer->meta();
	firstTimeUs = curTime.mTimeUs;
	break;
	}
	}
	mBuffers.erase(mBuffers.begin(), it);
	mLatestDequeuedMeta = NULL;

	// CHECK(!mDiscontinuitySegments.empty());
	DiscontinuitySegment &seg = *mDiscontinuitySegments.begin();
	if (firstTimeUs >= 0) {
	seg.mMaxDequeTimeUs = firstTimeUs;
	} else {
	seg.clear();
	}

	return firstMeta;
	}

	} // namespace android