summaryrefslogtreecommitdiffstats
path: root/media
diff options
context:
space:
mode:
authorLajos Molnar <lajos@google.com>2014-03-07 18:31:25 +0000
committerAndroid Git Automerger <android-git-automerger@android.com>2014-03-07 18:31:25 +0000
commit3eee65fa79e382b065ff5299bdc81f3a5b85be9a (patch)
tree06ae58990f32e0f8d51a90a22cce7733290156ad /media
parentc202aed28f3a24a7fbc74f7a65d806ef8deefb0b (diff)
parent49ea13379fb15ddb73183ebafa3a377342ef932f (diff)
downloadframeworks_av-3eee65fa79e382b065ff5299bdc81f3a5b85be9a.zip
frameworks_av-3eee65fa79e382b065ff5299bdc81f3a5b85be9a.tar.gz
frameworks_av-3eee65fa79e382b065ff5299bdc81f3a5b85be9a.tar.bz2
am 49ea1337: Merge changes I787e1c05,I72d3a5e1,I0a5cc65f,I75fc2a25,I2c2be08d, ... into klp-dev
* commit '49ea13379fb15ddb73183ebafa3a377342ef932f': LiveSession: Use the actual, possibly redirected url as base in the M3U M3UParser: Skip query strings when looking for the last slash in a URL ChromiumHTTPDataSource: Keep track of the redirected URL Initial HLS seamless switch implementation. NuPlayer side support for seamless format switch. LiveSession refactor PlaylistFetcher: Add support for block-by-block decryption. LiveSession: Add support for block-by-block fetchFile.
Diffstat (limited to 'media')
-rw-r--r--media/libmediaplayerservice/nuplayer/NuPlayer.cpp9
-rw-r--r--media/libmediaplayerservice/nuplayer/NuPlayerDecoder.cpp61
-rw-r--r--media/libmediaplayerservice/nuplayer/NuPlayerDecoder.h5
-rw-r--r--media/libstagefright/httplive/LiveSession.cpp557
-rw-r--r--media/libstagefright/httplive/LiveSession.h83
-rw-r--r--media/libstagefright/httplive/M3UParser.cpp12
-rw-r--r--media/libstagefright/httplive/M3UParser.h6
-rw-r--r--media/libstagefright/httplive/PlaylistFetcher.cpp346
-rw-r--r--media/libstagefright/httplive/PlaylistFetcher.h39
9 files changed, 864 insertions, 254 deletions
diff --git a/media/libmediaplayerservice/nuplayer/NuPlayer.cpp b/media/libmediaplayerservice/nuplayer/NuPlayer.cpp
index 3669a5b..25d55a3 100644
--- a/media/libmediaplayerservice/nuplayer/NuPlayer.cpp
+++ b/media/libmediaplayerservice/nuplayer/NuPlayer.cpp
@@ -1011,7 +1011,14 @@ status_t NuPlayer::feedDecoderInputData(bool audio, const sp<AMessage> &msg) {
&NuPlayer::performScanSources));
}
- flushDecoder(audio, formatChange);
+ sp<AMessage> newFormat = mSource->getFormat(audio);
+ sp<Decoder> &decoder = audio ? mAudioDecoder : mVideoDecoder;
+ if (formatChange && !decoder->supportsSeamlessFormatChange(newFormat)) {
+ flushDecoder(audio, /* needShutdown = */ true);
+ } else {
+ flushDecoder(audio, /* needShutdown = */ false);
+ err = OK;
+ }
} else {
// This stream is unaffected by the discontinuity
diff --git a/media/libmediaplayerservice/nuplayer/NuPlayerDecoder.cpp b/media/libmediaplayerservice/nuplayer/NuPlayerDecoder.cpp
index 22f699e..2423fd5 100644
--- a/media/libmediaplayerservice/nuplayer/NuPlayerDecoder.cpp
+++ b/media/libmediaplayerservice/nuplayer/NuPlayerDecoder.cpp
@@ -67,6 +67,7 @@ void NuPlayer::Decoder::configure(const sp<AMessage> &format) {
// queue.
bool needDedicatedLooper = !strncasecmp(mime.c_str(), "video/", 6);
+ mFormat = format;
mCodec = new ACodec;
if (needDedicatedLooper && mCodecLooper == NULL) {
@@ -147,5 +148,65 @@ void NuPlayer::Decoder::initiateShutdown() {
}
}
+bool NuPlayer::Decoder::supportsSeamlessAudioFormatChange(const sp<AMessage> &targetFormat) const {
+ if (targetFormat == NULL) {
+ return true;
+ }
+
+ AString mime;
+ if (!targetFormat->findString("mime", &mime)) {
+ return false;
+ }
+
+ if (!strcasecmp(mime.c_str(), MEDIA_MIMETYPE_AUDIO_AAC)) {
+ // field-by-field comparison
+ const char * keys[] = { "channel-count", "sample-rate", "is-adts" };
+ for (unsigned int i = 0; i < sizeof(keys) / sizeof(keys[0]); i++) {
+ int32_t oldVal, newVal;
+ if (!mFormat->findInt32(keys[i], &oldVal) || !targetFormat->findInt32(keys[i], &newVal)
+ || oldVal != newVal) {
+ return false;
+ }
+ }
+
+ sp<ABuffer> oldBuf, newBuf;
+ if (mFormat->findBuffer("csd-0", &oldBuf) && targetFormat->findBuffer("csd-0", &newBuf)) {
+ if (oldBuf->size() != newBuf->size()) {
+ return false;
+ }
+ return !memcmp(oldBuf->data(), newBuf->data(), oldBuf->size());
+ }
+ }
+ return false;
+}
+
+bool NuPlayer::Decoder::supportsSeamlessFormatChange(const sp<AMessage> &targetFormat) const {
+ if (mFormat == NULL) {
+ return false;
+ }
+
+ if (targetFormat == NULL) {
+ return true;
+ }
+
+ AString oldMime, newMime;
+ if (!mFormat->findString("mime", &oldMime)
+ || !targetFormat->findString("mime", &newMime)
+ || !(oldMime == newMime)) {
+ return false;
+ }
+
+ bool audio = !strncasecmp(oldMime.c_str(), "audio/", strlen("audio/"));
+ bool seamless;
+ if (audio) {
+ seamless = supportsSeamlessAudioFormatChange(targetFormat);
+ } else {
+ seamless = mCodec != NULL && mCodec->isConfiguredForAdaptivePlayback();
+ }
+
+ ALOGV("%s seamless support for %s", seamless ? "yes" : "no", oldMime.c_str());
+ return seamless;
+}
+
} // namespace android
diff --git a/media/libmediaplayerservice/nuplayer/NuPlayerDecoder.h b/media/libmediaplayerservice/nuplayer/NuPlayerDecoder.h
index a876148..78ea74a 100644
--- a/media/libmediaplayerservice/nuplayer/NuPlayerDecoder.h
+++ b/media/libmediaplayerservice/nuplayer/NuPlayerDecoder.h
@@ -36,6 +36,8 @@ struct NuPlayer::Decoder : public AHandler {
void signalResume();
void initiateShutdown();
+ bool supportsSeamlessFormatChange(const sp<AMessage> &to) const;
+
protected:
virtual ~Decoder();
@@ -49,6 +51,7 @@ private:
sp<AMessage> mNotify;
sp<NativeWindowWrapper> mNativeWindow;
+ sp<AMessage> mFormat;
sp<ACodec> mCodec;
sp<ALooper> mCodecLooper;
@@ -59,6 +62,8 @@ private:
void onFillThisBuffer(const sp<AMessage> &msg);
+ bool supportsSeamlessAudioFormatChange(const sp<AMessage> &targetFormat) const;
+
DISALLOW_EVIL_CONSTRUCTORS(Decoder);
};
diff --git a/media/libstagefright/httplive/LiveSession.cpp b/media/libstagefright/httplive/LiveSession.cpp
index 233db44..80aec06 100644
--- a/media/libstagefright/httplive/LiveSession.cpp
+++ b/media/libstagefright/httplive/LiveSession.cpp
@@ -37,6 +37,8 @@
#include <media/stagefright/MetaData.h>
#include <media/stagefright/Utils.h>
+#include <utils/Mutex.h>
+
#include <ctype.h>
#include <openssl/aes.h>
#include <openssl/md5.h>
@@ -57,32 +59,56 @@ LiveSession::LiveSession(
: 0)),
mPrevBandwidthIndex(-1),
mStreamMask(0),
+ mNewStreamMask(0),
+ mSwapMask(0),
mCheckBandwidthGeneration(0),
+ mSwitchGeneration(0),
mLastDequeuedTimeUs(0ll),
mRealTimeBaseUs(0ll),
mReconfigurationInProgress(false),
+ mSwitchInProgress(false),
mDisconnectReplyID(0) {
if (mUIDValid) {
mHTTPDataSource->setUID(mUID);
}
- mPacketSources.add(
- STREAMTYPE_AUDIO, new AnotherPacketSource(NULL /* meta */));
-
- mPacketSources.add(
- STREAMTYPE_VIDEO, new AnotherPacketSource(NULL /* meta */));
+ mStreams[kAudioIndex] = StreamItem("audio");
+ mStreams[kVideoIndex] = StreamItem("video");
+ mStreams[kSubtitleIndex] = StreamItem("subtitle");
- mPacketSources.add(
- STREAMTYPE_SUBTITLES, new AnotherPacketSource(NULL /* meta */));
+ for (size_t i = 0; i < kMaxStreams; ++i) {
+ mPacketSources.add(indexToType(i), new AnotherPacketSource(NULL /* meta */));
+ mPacketSources2.add(indexToType(i), new AnotherPacketSource(NULL /* meta */));
+ }
}
LiveSession::~LiveSession() {
}
+sp<ABuffer> LiveSession::createFormatChangeBuffer(bool swap) {
+ ABuffer *discontinuity = new ABuffer(0);
+ discontinuity->meta()->setInt32("discontinuity", ATSParser::DISCONTINUITY_FORMATCHANGE);
+ discontinuity->meta()->setInt32("swapPacketSource", swap);
+ discontinuity->meta()->setInt32("switchGeneration", mSwitchGeneration);
+ discontinuity->meta()->setInt64("timeUs", -1);
+ return discontinuity;
+}
+
+void LiveSession::swapPacketSource(StreamType stream) {
+ sp<AnotherPacketSource> &aps = mPacketSources.editValueFor(stream);
+ sp<AnotherPacketSource> &aps2 = mPacketSources2.editValueFor(stream);
+ sp<AnotherPacketSource> tmp = aps;
+ aps = aps2;
+ aps2 = tmp;
+ aps2->clear();
+}
+
status_t LiveSession::dequeueAccessUnit(
StreamType stream, sp<ABuffer> *accessUnit) {
if (!(mStreamMask & stream)) {
- return UNKNOWN_ERROR;
+ // return -EWOULDBLOCK to avoid halting the decoder
+ // when switching between audio/video and audio only.
+ return -EWOULDBLOCK;
}
sp<AnotherPacketSource> packetSource = mPacketSources.valueFor(stream);
@@ -122,6 +148,25 @@ status_t LiveSession::dequeueAccessUnit(
streamStr,
type,
extra == NULL ? "NULL" : extra->debugString().c_str());
+
+ int32_t swap;
+ if (type == ATSParser::DISCONTINUITY_FORMATCHANGE
+ && (*accessUnit)->meta()->findInt32("swapPacketSource", &swap)
+ && swap) {
+
+ int32_t switchGeneration;
+ CHECK((*accessUnit)->meta()->findInt32("switchGeneration", &switchGeneration));
+ {
+ Mutex::Autolock lock(mSwapMutex);
+ if (switchGeneration == mSwitchGeneration) {
+ swapPacketSource(stream);
+ sp<AMessage> msg = new AMessage(kWhatSwapped, id());
+ msg->setInt32("stream", stream);
+ msg->setInt32("switchGeneration", switchGeneration);
+ msg->post();
+ }
+ }
+ }
} else if (err == OK) {
if (stream == STREAMTYPE_AUDIO || stream == STREAMTYPE_VIDEO) {
int64_t timeUs;
@@ -143,6 +188,7 @@ status_t LiveSession::dequeueAccessUnit(
}
status_t LiveSession::getStreamFormat(StreamType stream, sp<AMessage> *format) {
+ // No swapPacketSource race condition; called from the same thread as dequeueAccessUnit.
if (!(mStreamMask & stream)) {
return UNKNOWN_ERROR;
}
@@ -239,7 +285,12 @@ void LiveSession::onMessageReceived(const sp<AMessage> &msg) {
if (what == PlaylistFetcher::kWhatStopped) {
AString uri;
CHECK(msg->findString("uri", &uri));
- mFetcherInfos.removeItem(uri);
+ if (mFetcherInfos.removeItem(uri) < 0) {
+ // ignore duplicated kWhatStopped messages.
+ break;
+ }
+
+ tryToFinishBandwidthSwitch();
}
if (mContinuation != NULL) {
@@ -275,6 +326,8 @@ void LiveSession::onMessageReceived(const sp<AMessage> &msg) {
postPrepared(err);
}
+ cancelBandwidthSwitch();
+
mPacketSources.valueFor(STREAMTYPE_AUDIO)->signalEOS(err);
mPacketSources.valueFor(STREAMTYPE_VIDEO)->signalEOS(err);
@@ -313,6 +366,27 @@ void LiveSession::onMessageReceived(const sp<AMessage> &msg) {
break;
}
+ case PlaylistFetcher::kWhatStartedAt:
+ {
+ int32_t switchGeneration;
+ CHECK(msg->findInt32("switchGeneration", &switchGeneration));
+
+ if (switchGeneration != mSwitchGeneration) {
+ break;
+ }
+
+ // Resume fetcher for the original variant; the resumed fetcher should
+ // continue until the timestamps found in msg, which is stored by the
+ // new fetcher to indicate where the new variant has started buffering.
+ for (size_t i = 0; i < mFetcherInfos.size(); i++) {
+ const FetcherInfo info = mFetcherInfos.valueAt(i);
+ if (info.mToBeRemoved) {
+ info.mFetcher->resumeUntilAsync(msg);
+ }
+ }
+ break;
+ }
+
default:
TRESPASS();
}
@@ -357,6 +431,11 @@ void LiveSession::onMessageReceived(const sp<AMessage> &msg) {
break;
}
+ case kWhatSwapped:
+ {
+ onSwapped(msg);
+ break;
+ }
default:
TRESPASS();
break;
@@ -374,6 +453,12 @@ int LiveSession::SortByBandwidth(const BandwidthItem *a, const BandwidthItem *b)
return 1;
}
+// static
+LiveSession::StreamType LiveSession::indexToType(int idx) {
+ CHECK(idx >= 0 && idx < kMaxStreams);
+ return (StreamType)(1 << idx);
+}
+
void LiveSession::onConnect(const sp<AMessage> &msg) {
AString url;
CHECK(msg->findString("url", &url));
@@ -461,6 +546,10 @@ void LiveSession::finishDisconnect() {
// during disconnection either.
cancelCheckBandwidthEvent();
+ // Protect mPacketSources from a swapPacketSource race condition through disconnect.
+ // (finishDisconnect, onFinishDisconnect2)
+ cancelBandwidthSwitch();
+
for (size_t i = 0; i < mFetcherInfos.size(); ++i) {
mFetcherInfos.valueAt(i).mFetcher->stopAsync();
}
@@ -500,11 +589,13 @@ sp<PlaylistFetcher> LiveSession::addFetcher(const char *uri) {
sp<AMessage> notify = new AMessage(kWhatFetcherNotify, id());
notify->setString("uri", uri);
+ notify->setInt32("switchGeneration", mSwitchGeneration);
FetcherInfo info;
info.mFetcher = new PlaylistFetcher(notify, this, uri);
info.mDurationUs = -1ll;
info.mIsPrepared = false;
+ info.mToBeRemoved = false;
looper()->registerHandler(info.mFetcher);
mFetcherInfos.add(uri, info);
@@ -512,53 +603,81 @@ sp<PlaylistFetcher> LiveSession::addFetcher(const char *uri) {
return info.mFetcher;
}
+/*
+ * Illustration of parameters:
+ *
+ * 0 `range_offset`
+ * +------------+-------------------------------------------------------+--+--+
+ * | | | next block to fetch | | |
+ * | | `source` handle => `out` buffer | | | |
+ * | `url` file |<--------- buffer size --------->|<--- `block_size` -->| | |
+ * | |<----------- `range_length` / buffer capacity ----------->| |
+ * |<------------------------------ file_size ------------------------------->|
+ *
+ * Special parameter values:
+ * - range_length == -1 means entire file
+ * - block_size == 0 means entire range
+ *
+ */
status_t LiveSession::fetchFile(
const char *url, sp<ABuffer> *out,
- int64_t range_offset, int64_t range_length) {
- *out = NULL;
+ int64_t range_offset, int64_t range_length,
+ uint32_t block_size, /* download block size */
+ sp<DataSource> *source, /* to return and reuse source */
+ String8 *actualUrl) {
+ off64_t size;
+ sp<DataSource> temp_source;
+ if (source == NULL) {
+ source = &temp_source;
+ }
- sp<DataSource> source;
+ if (*source == NULL) {
+ if (!strncasecmp(url, "file://", 7)) {
+ *source = new FileSource(url + 7);
+ } else if (strncasecmp(url, "http://", 7)
+ && strncasecmp(url, "https://", 8)) {
+ return ERROR_UNSUPPORTED;
+ } else {
+ KeyedVector<String8, String8> headers = mExtraHeaders;
+ if (range_offset > 0 || range_length >= 0) {
+ headers.add(
+ String8("Range"),
+ String8(
+ StringPrintf(
+ "bytes=%lld-%s",
+ range_offset,
+ range_length < 0
+ ? "" : StringPrintf("%lld",
+ range_offset + range_length - 1).c_str()).c_str()));
+ }
+ status_t err = mHTTPDataSource->connect(url, &headers);
- if (!strncasecmp(url, "file://", 7)) {
- source = new FileSource(url + 7);
- } else if (strncasecmp(url, "http://", 7)
- && strncasecmp(url, "https://", 8)) {
- return ERROR_UNSUPPORTED;
- } else {
- KeyedVector<String8, String8> headers = mExtraHeaders;
- if (range_offset > 0 || range_length >= 0) {
- headers.add(
- String8("Range"),
- String8(
- StringPrintf(
- "bytes=%lld-%s",
- range_offset,
- range_length < 0
- ? "" : StringPrintf("%lld", range_offset + range_length - 1).c_str()).c_str()));
- }
- status_t err = mHTTPDataSource->connect(url, &headers);
+ if (err != OK) {
+ return err;
+ }
- if (err != OK) {
- return err;
+ *source = mHTTPDataSource;
}
-
- source = mHTTPDataSource;
}
- off64_t size;
- status_t err = source->getSize(&size);
-
- if (err != OK) {
+ status_t getSizeErr = (*source)->getSize(&size);
+ if (getSizeErr != OK) {
size = 65536;
}
- sp<ABuffer> buffer = new ABuffer(size);
- buffer->setRange(0, 0);
+ sp<ABuffer> buffer = *out != NULL ? *out : new ABuffer(size);
+ if (*out == NULL) {
+ buffer->setRange(0, 0);
+ }
+ // adjust range_length if only reading partial block
+ if (block_size > 0 && (range_length == -1 || buffer->size() + block_size < range_length)) {
+ range_length = buffer->size() + block_size;
+ }
for (;;) {
+ // Only resize when we don't know the size.
size_t bufferRemaining = buffer->capacity() - buffer->size();
-
- if (bufferRemaining == 0) {
+ if (bufferRemaining == 0 && getSizeErr != OK) {
bufferRemaining = 32768;
ALOGV("increasing download buffer to %d bytes",
@@ -583,7 +702,9 @@ status_t LiveSession::fetchFile(
}
}
- ssize_t n = source->readAt(
+ // The DataSource is responsible for informing us of error (n < 0) or eof (n == 0)
+ // to help us break out of the loop.
+ ssize_t n = (*source)->readAt(
buffer->size(), buffer->data() + buffer->size(),
maxBytesToRead);
@@ -599,6 +720,12 @@ status_t LiveSession::fetchFile(
}
*out = buffer;
+ if (actualUrl != NULL) {
+ *actualUrl = (*source)->getUri();
+ if (actualUrl->isEmpty()) {
+ *actualUrl = url;
+ }
+ }
return OK;
}
@@ -610,7 +737,8 @@ sp<M3UParser> LiveSession::fetchPlaylist(
*unchanged = false;
sp<ABuffer> buffer;
- status_t err = fetchFile(url, &buffer);
+ String8 actualUrl;
+ status_t err = fetchFile(url, &buffer, 0, -1, 0, NULL, &actualUrl);
if (err != OK) {
return NULL;
@@ -641,7 +769,7 @@ sp<M3UParser> LiveSession::fetchPlaylist(
#endif
sp<M3UParser> playlist =
- new M3UParser(url, buffer->data(), buffer->size());
+ new M3UParser(actualUrl.string(), buffer->data(), buffer->size());
if (playlist->initCheck() != OK) {
ALOGE("failed to parse .m3u8 playlist");
@@ -807,8 +935,25 @@ status_t LiveSession::selectTrack(size_t index, bool select) {
return err;
}
+bool LiveSession::canSwitchUp() {
+ // Allow upwards bandwidth switch when a stream has buffered at least 10 seconds.
+ status_t err = OK;
+ for (size_t i = 0; i < mPacketSources.size(); ++i) {
+ sp<AnotherPacketSource> source = mPacketSources.valueAt(i);
+ int64_t dur = source->getBufferedDurationUs(&err);
+ if (err == OK && dur > 10000000) {
+ return true;
+ }
+ }
+ return false;
+}
+
void LiveSession::changeConfiguration(
int64_t timeUs, size_t bandwidthIndex, bool pickTrack) {
+ // Protect mPacketSources from a swapPacketSource race condition through reconfiguration.
+ // (changeConfiguration, onChangeConfiguration2, onChangeConfiguration3).
+ cancelBandwidthSwitch();
+
CHECK(!mReconfigurationInProgress);
mReconfigurationInProgress = true;
@@ -824,21 +969,14 @@ void LiveSession::changeConfiguration(
CHECK_LT(bandwidthIndex, mBandwidthItems.size());
const BandwidthItem &item = mBandwidthItems.itemAt(bandwidthIndex);
- uint32_t streamMask = 0;
-
- AString audioURI;
- if (mPlaylist->getAudioURI(item.mPlaylistIndex, &audioURI)) {
- streamMask |= STREAMTYPE_AUDIO;
- }
-
- AString videoURI;
- if (mPlaylist->getVideoURI(item.mPlaylistIndex, &videoURI)) {
- streamMask |= STREAMTYPE_VIDEO;
- }
+ uint32_t streamMask = 0; // streams that should be fetched by the new fetcher
+ uint32_t resumeMask = 0; // streams that should be fetched by the original fetcher
- AString subtitleURI;
- if (mPlaylist->getSubtitleURI(item.mPlaylistIndex, &subtitleURI)) {
- streamMask |= STREAMTYPE_SUBTITLES;
+ AString URIs[kMaxStreams];
+ for (size_t i = 0; i < kMaxStreams; ++i) {
+ if (mPlaylist->getTypeURI(item.mPlaylistIndex, mStreams[i].mType, &URIs[i])) {
+ streamMask |= indexToType(i);
+ }
}
// Step 1, stop and discard fetchers that are no longer needed.
@@ -850,10 +988,15 @@ void LiveSession::changeConfiguration(
// If we're seeking all current fetchers are discarded.
if (timeUs < 0ll) {
- if (((streamMask & STREAMTYPE_AUDIO) && uri == audioURI)
- || ((streamMask & STREAMTYPE_VIDEO) && uri == videoURI)
- || ((streamMask & STREAMTYPE_SUBTITLES) && uri == subtitleURI)) {
- discardFetcher = false;
+ // delay fetcher removal
+ discardFetcher = false;
+
+ for (size_t j = 0; j < kMaxStreams; ++j) {
+ StreamType type = indexToType(j);
+ if ((streamMask & type) && uri == URIs[j]) {
+ resumeMask |= type;
+ streamMask &= ~type;
+ }
}
}
@@ -864,17 +1007,20 @@ void LiveSession::changeConfiguration(
}
}
- sp<AMessage> msg = new AMessage(kWhatChangeConfiguration2, id());
+ sp<AMessage> msg;
+ if (timeUs < 0ll) {
+ // skip onChangeConfiguration2 (decoder destruction) if switching.
+ msg = new AMessage(kWhatChangeConfiguration3, id());
+ } else {
+ msg = new AMessage(kWhatChangeConfiguration2, id());
+ }
msg->setInt32("streamMask", streamMask);
+ msg->setInt32("resumeMask", resumeMask);
msg->setInt64("timeUs", timeUs);
- if (streamMask & STREAMTYPE_AUDIO) {
- msg->setString("audioURI", audioURI.c_str());
- }
- if (streamMask & STREAMTYPE_VIDEO) {
- msg->setString("videoURI", videoURI.c_str());
- }
- if (streamMask & STREAMTYPE_SUBTITLES) {
- msg->setString("subtitleURI", subtitleURI.c_str());
+ for (size_t i = 0; i < kMaxStreams; ++i) {
+ if (streamMask & indexToType(i)) {
+ msg->setString(mStreams[i].uriKey().c_str(), URIs[i].c_str());
+ }
}
// Every time a fetcher acknowledges the stopAsync or pauseAsync request
@@ -905,18 +1051,13 @@ void LiveSession::onChangeConfiguration2(const sp<AMessage> &msg) {
uint32_t streamMask;
CHECK(msg->findInt32("streamMask", (int32_t *)&streamMask));
- AString audioURI, videoURI, subtitleURI;
- if (streamMask & STREAMTYPE_AUDIO) {
- CHECK(msg->findString("audioURI", &audioURI));
- ALOGV("audioURI = '%s'", audioURI.c_str());
- }
- if (streamMask & STREAMTYPE_VIDEO) {
- CHECK(msg->findString("videoURI", &videoURI));
- ALOGV("videoURI = '%s'", videoURI.c_str());
- }
- if (streamMask & STREAMTYPE_SUBTITLES) {
- CHECK(msg->findString("subtitleURI", &subtitleURI));
- ALOGV("subtitleURI = '%s'", subtitleURI.c_str());
+ AString URIs[kMaxStreams];
+ for (size_t i = 0; i < kMaxStreams; ++i) {
+ if (streamMask & indexToType(i)) {
+ const AString &uriKey = mStreams[i].uriKey();
+ CHECK(msg->findString(uriKey.c_str(), &URIs[i]));
+ ALOGV("%s = '%s'", uriKey.c_str(), URIs[i].c_str());
+ }
}
// Determine which decoders to shutdown on the player side,
@@ -926,15 +1067,12 @@ void LiveSession::onChangeConfiguration2(const sp<AMessage> &msg) {
// 2) its streamtype was already active and still is but the URI
// has changed.
uint32_t changedMask = 0;
- if (((mStreamMask & streamMask & STREAMTYPE_AUDIO)
- && !(audioURI == mAudioURI))
- || (mStreamMask & ~streamMask & STREAMTYPE_AUDIO)) {
- changedMask |= STREAMTYPE_AUDIO;
- }
- if (((mStreamMask & streamMask & STREAMTYPE_VIDEO)
- && !(videoURI == mVideoURI))
- || (mStreamMask & ~streamMask & STREAMTYPE_VIDEO)) {
- changedMask |= STREAMTYPE_VIDEO;
+ for (size_t i = 0; i < kMaxStreams && i != kSubtitleIndex; ++i) {
+ if (((mStreamMask & streamMask & indexToType(i))
+ && !(URIs[i] == mStreams[i].mUri))
+ || (mStreamMask & ~streamMask & indexToType(i))) {
+ changedMask |= indexToType(i);
+ }
}
if (changedMask == 0) {
@@ -960,68 +1098,54 @@ void LiveSession::onChangeConfiguration2(const sp<AMessage> &msg) {
}
void LiveSession::onChangeConfiguration3(const sp<AMessage> &msg) {
+ mContinuation.clear();
// All remaining fetchers are still suspended, the player has shutdown
// any decoders that needed it.
- uint32_t streamMask;
+ uint32_t streamMask, resumeMask;
CHECK(msg->findInt32("streamMask", (int32_t *)&streamMask));
+ CHECK(msg->findInt32("resumeMask", (int32_t *)&resumeMask));
- AString audioURI, videoURI, subtitleURI;
- if (streamMask & STREAMTYPE_AUDIO) {
- CHECK(msg->findString("audioURI", &audioURI));
- }
- if (streamMask & STREAMTYPE_VIDEO) {
- CHECK(msg->findString("videoURI", &videoURI));
- }
- if (streamMask & STREAMTYPE_SUBTITLES) {
- CHECK(msg->findString("subtitleURI", &subtitleURI));
+ for (size_t i = 0; i < kMaxStreams; ++i) {
+ if (streamMask & indexToType(i)) {
+ CHECK(msg->findString(mStreams[i].uriKey().c_str(), &mStreams[i].mUri));
+ }
}
int64_t timeUs;
+ bool switching = false;
CHECK(msg->findInt64("timeUs", &timeUs));
if (timeUs < 0ll) {
timeUs = mLastDequeuedTimeUs;
+ switching = true;
}
mRealTimeBaseUs = ALooper::GetNowUs() - timeUs;
- mStreamMask = streamMask;
- mAudioURI = audioURI;
- mVideoURI = videoURI;
- mSubtitleURI = subtitleURI;
+ mNewStreamMask = streamMask;
- // Resume all existing fetchers and assign them packet sources.
+ // Of all existing fetchers:
+ // * Resume fetchers that are still needed and assign them original packet sources.
+ // * Mark otherwise unneeded fetchers for removal.
+ ALOGV("resuming fetchers for mask 0x%08x", resumeMask);
for (size_t i = 0; i < mFetcherInfos.size(); ++i) {
const AString &uri = mFetcherInfos.keyAt(i);
- uint32_t resumeMask = 0;
-
- sp<AnotherPacketSource> audioSource;
- if ((streamMask & STREAMTYPE_AUDIO) && uri == audioURI) {
- audioSource = mPacketSources.valueFor(STREAMTYPE_AUDIO);
- resumeMask |= STREAMTYPE_AUDIO;
- }
-
- sp<AnotherPacketSource> videoSource;
- if ((streamMask & STREAMTYPE_VIDEO) && uri == videoURI) {
- videoSource = mPacketSources.valueFor(STREAMTYPE_VIDEO);
- resumeMask |= STREAMTYPE_VIDEO;
+ sp<AnotherPacketSource> sources[kMaxStreams];
+ for (size_t j = 0; j < kMaxStreams; ++j) {
+ if ((resumeMask & indexToType(j)) && uri == mStreams[j].mUri) {
+ sources[j] = mPacketSources.valueFor(indexToType(j));
+ }
}
- sp<AnotherPacketSource> subtitleSource;
- if ((streamMask & STREAMTYPE_SUBTITLES) && uri == subtitleURI) {
- subtitleSource = mPacketSources.valueFor(STREAMTYPE_SUBTITLES);
- resumeMask |= STREAMTYPE_SUBTITLES;
+ FetcherInfo &info = mFetcherInfos.editValueAt(i);
+ if (sources[kAudioIndex] != NULL || sources[kVideoIndex] != NULL
+ || sources[kSubtitleIndex] != NULL) {
+ info.mFetcher->startAsync(
+ sources[kAudioIndex], sources[kVideoIndex], sources[kSubtitleIndex]);
+ } else {
+ info.mToBeRemoved = true;
}
-
- CHECK_NE(resumeMask, 0u);
-
- ALOGV("resuming fetchers for mask 0x%08x", resumeMask);
-
- streamMask &= ~resumeMask;
-
- mFetcherInfos.valueAt(i).mFetcher->startAsync(
- audioSource, videoSource, subtitleSource);
}
// streamMask now only contains the types that need a new fetcher created.
@@ -1030,52 +1154,65 @@ void LiveSession::onChangeConfiguration3(const sp<AMessage> &msg) {
ALOGV("creating new fetchers for mask 0x%08x", streamMask);
}
- while (streamMask != 0) {
- StreamType streamType = (StreamType)(streamMask & ~(streamMask - 1));
+ // Find out when the original fetchers have buffered up to and start the new fetchers
+ // at a later timestamp.
+ for (size_t i = 0; i < kMaxStreams; i++) {
+ if (!(indexToType(i) & streamMask)) {
+ continue;
+ }
AString uri;
- switch (streamType) {
- case STREAMTYPE_AUDIO:
- uri = audioURI;
- break;
- case STREAMTYPE_VIDEO:
- uri = videoURI;
- break;
- case STREAMTYPE_SUBTITLES:
- uri = subtitleURI;
- break;
- default:
- TRESPASS();
- }
+ uri = mStreams[i].mUri;
sp<PlaylistFetcher> fetcher = addFetcher(uri.c_str());
CHECK(fetcher != NULL);
- sp<AnotherPacketSource> audioSource;
- if ((streamMask & STREAMTYPE_AUDIO) && uri == audioURI) {
- audioSource = mPacketSources.valueFor(STREAMTYPE_AUDIO);
- audioSource->clear();
-
- streamMask &= ~STREAMTYPE_AUDIO;
- }
-
- sp<AnotherPacketSource> videoSource;
- if ((streamMask & STREAMTYPE_VIDEO) && uri == videoURI) {
- videoSource = mPacketSources.valueFor(STREAMTYPE_VIDEO);
- videoSource->clear();
-
- streamMask &= ~STREAMTYPE_VIDEO;
- }
+ int32_t latestSeq = -1;
+ int64_t latestTimeUs = 0ll;
+ sp<AnotherPacketSource> sources[kMaxStreams];
+
+ // TRICKY: looping from i as earlier streams are already removed from streamMask
+ for (size_t j = i; j < kMaxStreams; ++j) {
+ if ((streamMask & indexToType(j)) && uri == mStreams[j].mUri) {
+ sources[j] = mPacketSources.valueFor(indexToType(j));
+
+ if (!switching) {
+ sources[j]->clear();
+ } else {
+ int32_t type, seq;
+ int64_t srcTimeUs;
+ sp<AMessage> meta = sources[j]->getLatestMeta();
+
+ if (meta != NULL && !meta->findInt32("discontinuity", &type)) {
+ CHECK(meta->findInt32("seq", &seq));
+ if (seq > latestSeq) {
+ latestSeq = seq;
+ }
+ CHECK(meta->findInt64("timeUs", &srcTimeUs));
+ if (srcTimeUs > latestTimeUs) {
+ latestTimeUs = srcTimeUs;
+ }
+ }
- sp<AnotherPacketSource> subtitleSource;
- if ((streamMask & STREAMTYPE_SUBTITLES) && uri == subtitleURI) {
- subtitleSource = mPacketSources.valueFor(STREAMTYPE_SUBTITLES);
- subtitleSource->clear();
+ sources[j] = mPacketSources2.valueFor(indexToType(j));
+ sources[j]->clear();
+ uint32_t extraStreams = mNewStreamMask & (~mStreamMask);
+ if (extraStreams & indexToType(j)) {
+ sources[j]->queueAccessUnit(createFormatChangeBuffer(/* swap = */ false));
+ }
+ }
- streamMask &= ~STREAMTYPE_SUBTITLES;
+ streamMask &= ~indexToType(j);
+ }
}
- fetcher->startAsync(audioSource, videoSource, subtitleSource, timeUs);
+ fetcher->startAsync(
+ sources[kAudioIndex],
+ sources[kVideoIndex],
+ sources[kSubtitleIndex],
+ timeUs,
+ latestTimeUs /* min start time(us) */,
+ latestSeq >= 0 ? latestSeq + 1 : -1 /* starting sequence number hint */ );
}
// All fetchers have now been started, the configuration change
@@ -1084,14 +1221,61 @@ void LiveSession::onChangeConfiguration3(const sp<AMessage> &msg) {
scheduleCheckBandwidthEvent();
ALOGV("XXX configuration change completed.");
-
mReconfigurationInProgress = false;
+ if (switching) {
+ mSwitchInProgress = true;
+ } else {
+ mStreamMask = mNewStreamMask;
+ }
if (mDisconnectReplyID != 0) {
finishDisconnect();
}
}
+void LiveSession::onSwapped(const sp<AMessage> &msg) {
+ int32_t switchGeneration;
+ CHECK(msg->findInt32("switchGeneration", &switchGeneration));
+ if (switchGeneration != mSwitchGeneration) {
+ return;
+ }
+
+ int32_t stream;
+ CHECK(msg->findInt32("stream", &stream));
+ mSwapMask |= stream;
+ if (mSwapMask != mStreamMask) {
+ return;
+ }
+
+ // Check if new variant contains extra streams.
+ uint32_t extraStreams = mNewStreamMask & (~mStreamMask);
+ while (extraStreams) {
+ StreamType extraStream = (StreamType) (extraStreams & ~(extraStreams - 1));
+ swapPacketSource(extraStream);
+ extraStreams &= ~extraStream;
+ }
+
+ tryToFinishBandwidthSwitch();
+}
+
+// Mark switch done when:
+// 1. all old buffers are swapped out, AND
+// 2. all old fetchers are removed.
+void LiveSession::tryToFinishBandwidthSwitch() {
+ bool needToRemoveFetchers = false;
+ for (size_t i = 0; i < mFetcherInfos.size(); ++i) {
+ if (mFetcherInfos.valueAt(i).mToBeRemoved) {
+ needToRemoveFetchers = true;
+ break;
+ }
+ }
+ if (!needToRemoveFetchers && mSwapMask == mStreamMask) {
+ mStreamMask = mNewStreamMask;
+ mSwitchInProgress = false;
+ mSwapMask = 0;
+ }
+}
+
void LiveSession::scheduleCheckBandwidthEvent() {
sp<AMessage> msg = new AMessage(kWhatCheckBandwidth, id());
msg->setInt32("generation", mCheckBandwidthGeneration);
@@ -1102,16 +1286,37 @@ void LiveSession::cancelCheckBandwidthEvent() {
++mCheckBandwidthGeneration;
}
-void LiveSession::onCheckBandwidth() {
- if (mReconfigurationInProgress) {
- scheduleCheckBandwidthEvent();
- return;
+void LiveSession::cancelBandwidthSwitch() {
+ Mutex::Autolock lock(mSwapMutex);
+ mSwitchGeneration++;
+ mSwitchInProgress = false;
+ mSwapMask = 0;
+}
+
+bool LiveSession::canSwitchBandwidthTo(size_t bandwidthIndex) {
+ if (mReconfigurationInProgress || mSwitchInProgress) {
+ return false;
}
+ if (mPrevBandwidthIndex < 0) {
+ return true;
+ }
+
+ if (bandwidthIndex == (size_t)mPrevBandwidthIndex) {
+ return false;
+ } else if (bandwidthIndex > (size_t)mPrevBandwidthIndex) {
+ return canSwitchUp();
+ } else {
+ return true;
+ }
+}
+
+void LiveSession::onCheckBandwidth() {
size_t bandwidthIndex = getBandwidthIndex();
- if (mPrevBandwidthIndex < 0
- || bandwidthIndex != (size_t)mPrevBandwidthIndex) {
+ if (canSwitchBandwidthTo(bandwidthIndex)) {
changeConfiguration(-1ll /* timeUs */, bandwidthIndex);
+ } else {
+ scheduleCheckBandwidthEvent();
}
// Handling the kWhatCheckBandwidth even here does _not_ automatically
diff --git a/media/libstagefright/httplive/LiveSession.h b/media/libstagefright/httplive/LiveSession.h
index 99b480a8..376d451 100644
--- a/media/libstagefright/httplive/LiveSession.h
+++ b/media/libstagefright/httplive/LiveSession.h
@@ -42,10 +42,17 @@ struct LiveSession : public AHandler {
const sp<AMessage> &notify,
uint32_t flags = 0, bool uidValid = false, uid_t uid = 0);
+ enum StreamIndex {
+ kAudioIndex = 0,
+ kVideoIndex = 1,
+ kSubtitleIndex = 2,
+ kMaxStreams = 3,
+ };
+
enum StreamType {
- STREAMTYPE_AUDIO = 1,
- STREAMTYPE_VIDEO = 2,
- STREAMTYPE_SUBTITLES = 4,
+ STREAMTYPE_AUDIO = 1 << kAudioIndex,
+ STREAMTYPE_VIDEO = 1 << kVideoIndex,
+ STREAMTYPE_SUBTITLES = 1 << kSubtitleIndex,
};
status_t dequeueAccessUnit(StreamType stream, sp<ABuffer> *accessUnit);
@@ -74,6 +81,11 @@ struct LiveSession : public AHandler {
kWhatPreparationFailed,
};
+ // create a format-change discontinuity
+ //
+ // swap:
+ // whether is format-change discontinuity should trigger a buffer swap
+ sp<ABuffer> createFormatChangeBuffer(bool swap = true);
protected:
virtual ~LiveSession();
@@ -92,6 +104,7 @@ private:
kWhatChangeConfiguration2 = 'chC2',
kWhatChangeConfiguration3 = 'chC3',
kWhatFinishDisconnect2 = 'fin2',
+ kWhatSwapped = 'swap',
};
struct BandwidthItem {
@@ -103,8 +116,22 @@ private:
sp<PlaylistFetcher> mFetcher;
int64_t mDurationUs;
bool mIsPrepared;
+ bool mToBeRemoved;
};
+ struct StreamItem {
+ const char *mType;
+ AString mUri;
+ StreamItem() : mType("") {}
+ StreamItem(const char *type) : mType(type) {}
+ AString uriKey() {
+ AString key(mType);
+ key.append("URI");
+ return key;
+ }
+ };
+ StreamItem mStreams[kMaxStreams];
+
sp<AMessage> mNotify;
uint32_t mFlags;
bool mUIDValid;
@@ -123,12 +150,28 @@ private:
sp<M3UParser> mPlaylist;
KeyedVector<AString, FetcherInfo> mFetcherInfos;
- AString mAudioURI, mVideoURI, mSubtitleURI;
uint32_t mStreamMask;
+ // Masks used during reconfiguration:
+ // mNewStreamMask: streams in the variant playlist we're switching to;
+ // we don't want to immediately overwrite the original value.
+ uint32_t mNewStreamMask;
+
+ // mSwapMask: streams that have started to playback content in the new variant playlist;
+ // we use this to track reconfiguration progress.
+ uint32_t mSwapMask;
+
KeyedVector<StreamType, sp<AnotherPacketSource> > mPacketSources;
+ // A second set of packet sources that buffer content for the variant we're switching to.
+ KeyedVector<StreamType, sp<AnotherPacketSource> > mPacketSources2;
+
+ // A mutex used to serialize two sets of events:
+ // * the swapping of packet sources in dequeueAccessUnit on the player thread, AND
+ // * a forced bandwidth switch termination in cancelSwitch on the live looper.
+ Mutex mSwapMutex;
int32_t mCheckBandwidthGeneration;
+ int32_t mSwitchGeneration;
size_t mContinuationCounter;
sp<AMessage> mContinuation;
@@ -137,6 +180,7 @@ private:
int64_t mRealTimeBaseUs;
bool mReconfigurationInProgress;
+ bool mSwitchInProgress;
uint32_t mDisconnectReplyID;
sp<PlaylistFetcher> addFetcher(const char *uri);
@@ -145,9 +189,26 @@ private:
status_t onSeek(const sp<AMessage> &msg);
void onFinishDisconnect2();
+ // If given a non-zero block_size (default 0), it is used to cap the number of
+ // bytes read in from the DataSource. If given a non-NULL buffer, new content
+ // is read into the end.
+ //
+ // The DataSource we read from is responsible for signaling error or EOF to help us
+ // break out of the read loop. The DataSource can be returned to the caller, so
+ // that the caller can reuse it for subsequent fetches (within the initially
+ // requested range).
+ //
+ // For reused HTTP sources, the caller must download a file sequentially without
+ // any overlaps or gaps to prevent reconnection.
status_t fetchFile(
const char *url, sp<ABuffer> *out,
- int64_t range_offset = 0, int64_t range_length = -1);
+ /* request/open a file starting at range_offset for range_length bytes */
+ int64_t range_offset = 0, int64_t range_length = -1,
+ /* download block size */
+ uint32_t block_size = 0,
+ /* reuse DataSource if doing partial fetch */
+ sp<DataSource> *source = NULL,
+ String8 *actualUrl = NULL);
sp<M3UParser> fetchPlaylist(
const char *url, uint8_t *curPlaylistHash, bool *unchanged);
@@ -155,22 +216,34 @@ private:
size_t getBandwidthIndex();
static int SortByBandwidth(const BandwidthItem *, const BandwidthItem *);
+ static StreamType indexToType(int idx);
void changeConfiguration(
int64_t timeUs, size_t bandwidthIndex, bool pickTrack = false);
void onChangeConfiguration(const sp<AMessage> &msg);
void onChangeConfiguration2(const sp<AMessage> &msg);
void onChangeConfiguration3(const sp<AMessage> &msg);
+ void onSwapped(const sp<AMessage> &msg);
+ void tryToFinishBandwidthSwitch();
void scheduleCheckBandwidthEvent();
void cancelCheckBandwidthEvent();
+ // cancelBandwidthSwitch is atomic wrt swapPacketSource; call it to prevent packet sources
+ // from being swapped out on stale discontinuities while manipulating
+ // mPacketSources/mPacketSources2.
+ void cancelBandwidthSwitch();
+
+ bool canSwitchBandwidthTo(size_t bandwidthIndex);
void onCheckBandwidth();
void finishDisconnect();
void postPrepared(status_t err);
+ void swapPacketSource(StreamType stream);
+ bool canSwitchUp();
+
DISALLOW_EVIL_CONSTRUCTORS(LiveSession);
};
diff --git a/media/libstagefright/httplive/M3UParser.cpp b/media/libstagefright/httplive/M3UParser.cpp
index a5a18d9..4c2c1f4 100644
--- a/media/libstagefright/httplive/M3UParser.cpp
+++ b/media/libstagefright/httplive/M3UParser.cpp
@@ -388,18 +388,6 @@ bool M3UParser::getTypeURI(size_t index, const char *key, AString *uri) const {
return true;
}
-bool M3UParser::getAudioURI(size_t index, AString *uri) const {
- return getTypeURI(index, "audio", uri);
-}
-
-bool M3UParser::getVideoURI(size_t index, AString *uri) const {
- return getTypeURI(index, "video", uri);
-}
-
-bool M3UParser::getSubtitleURI(size_t index, AString *uri) const {
- return getTypeURI(index, "subtitles", uri);
-}
-
static bool MakeURL(const char *baseURL, const char *url, AString *out) {
out->clear();
diff --git a/media/libstagefright/httplive/M3UParser.h b/media/libstagefright/httplive/M3UParser.h
index b93b0e5..ccd6556 100644
--- a/media/libstagefright/httplive/M3UParser.h
+++ b/media/libstagefright/httplive/M3UParser.h
@@ -45,9 +45,7 @@ struct M3UParser : public RefBase {
status_t getTrackInfo(Parcel* reply) const;
ssize_t getSelectedIndex() const;
- bool getAudioURI(size_t index, AString *uri) const;
- bool getVideoURI(size_t index, AString *uri) const;
- bool getSubtitleURI(size_t index, AString *uri) const;
+ bool getTypeURI(size_t index, const char *key, AString *uri) const;
protected:
virtual ~M3UParser();
@@ -95,8 +93,6 @@ private:
status_t parseMedia(const AString &line);
- bool getTypeURI(size_t index, const char *key, AString *uri) const;
-
static status_t ParseInt32(const char *s, int32_t *x);
static status_t ParseDouble(const char *s, double *x);
diff --git a/media/libstagefright/httplive/PlaylistFetcher.cpp b/media/libstagefright/httplive/PlaylistFetcher.cpp
index f095987..0eac8b3 100644
--- a/media/libstagefright/httplive/PlaylistFetcher.cpp
+++ b/media/libstagefright/httplive/PlaylistFetcher.cpp
@@ -48,16 +48,20 @@ namespace android {
// static
const int64_t PlaylistFetcher::kMinBufferedDurationUs = 10000000ll;
const int64_t PlaylistFetcher::kMaxMonitorDelayUs = 3000000ll;
+const int32_t PlaylistFetcher::kNumSkipFrames = 10;
PlaylistFetcher::PlaylistFetcher(
const sp<AMessage> &notify,
const sp<LiveSession> &session,
const char *uri)
: mNotify(notify),
+ mStartTimeUsNotify(notify->dup()),
mSession(session),
mURI(uri),
mStreamTypeMask(0),
mStartTimeUs(-1ll),
+ mMinStartTimeUs(0ll),
+ mStopParams(NULL),
mLastPlaylistFetchTimeUs(-1ll),
mSeqNumber(-1),
mNumRetries(0),
@@ -69,6 +73,8 @@ PlaylistFetcher::PlaylistFetcher(
mFirstPTSValid(false),
mAbsoluteTimeAnchorUs(0ll) {
memset(mPlaylistHash, 0, sizeof(mPlaylistHash));
+ mStartTimeUsNotify->setInt32("what", kWhatStartedAt);
+ mStartTimeUsNotify->setInt32("streamMask", 0);
}
PlaylistFetcher::~PlaylistFetcher() {
@@ -170,7 +176,8 @@ int64_t PlaylistFetcher::delayUsToRefreshPlaylist() const {
}
status_t PlaylistFetcher::decryptBuffer(
- size_t playlistIndex, const sp<ABuffer> &buffer) {
+ size_t playlistIndex, const sp<ABuffer> &buffer,
+ bool first) {
sp<AMessage> itemMeta;
bool found = false;
AString method;
@@ -188,6 +195,7 @@ status_t PlaylistFetcher::decryptBuffer(
if (!found) {
method = "NONE";
}
+ buffer->meta()->setString("cipher-method", method.c_str());
if (method == "NONE") {
return OK;
@@ -227,59 +235,77 @@ status_t PlaylistFetcher::decryptBuffer(
return UNKNOWN_ERROR;
}
- unsigned char aes_ivec[16];
+ size_t n = buffer->size();
+ if (!n) {
+ return OK;
+ }
+ CHECK(n % 16 == 0);
- AString iv;
- if (itemMeta->findString("cipher-iv", &iv)) {
- if ((!iv.startsWith("0x") && !iv.startsWith("0X"))
- || iv.size() != 16 * 2 + 2) {
- ALOGE("malformed cipher IV '%s'.", iv.c_str());
- return ERROR_MALFORMED;
- }
+ if (first) {
+ // If decrypting the first block in a file, read the iv from the manifest
+ // or derive the iv from the file's sequence number.
- memset(aes_ivec, 0, sizeof(aes_ivec));
- for (size_t i = 0; i < 16; ++i) {
- char c1 = tolower(iv.c_str()[2 + 2 * i]);
- char c2 = tolower(iv.c_str()[3 + 2 * i]);
- if (!isxdigit(c1) || !isxdigit(c2)) {
+ AString iv;
+ if (itemMeta->findString("cipher-iv", &iv)) {
+ if ((!iv.startsWith("0x") && !iv.startsWith("0X"))
+ || iv.size() != 16 * 2 + 2) {
ALOGE("malformed cipher IV '%s'.", iv.c_str());
return ERROR_MALFORMED;
}
- uint8_t nibble1 = isdigit(c1) ? c1 - '0' : c1 - 'a' + 10;
- uint8_t nibble2 = isdigit(c2) ? c2 - '0' : c2 - 'a' + 10;
- aes_ivec[i] = nibble1 << 4 | nibble2;
+ memset(mAESInitVec, 0, sizeof(mAESInitVec));
+ for (size_t i = 0; i < 16; ++i) {
+ char c1 = tolower(iv.c_str()[2 + 2 * i]);
+ char c2 = tolower(iv.c_str()[3 + 2 * i]);
+ if (!isxdigit(c1) || !isxdigit(c2)) {
+ ALOGE("malformed cipher IV '%s'.", iv.c_str());
+ return ERROR_MALFORMED;
+ }
+ uint8_t nibble1 = isdigit(c1) ? c1 - '0' : c1 - 'a' + 10;
+ uint8_t nibble2 = isdigit(c2) ? c2 - '0' : c2 - 'a' + 10;
+
+ mAESInitVec[i] = nibble1 << 4 | nibble2;
+ }
+ } else {
+ memset(mAESInitVec, 0, sizeof(mAESInitVec));
+ mAESInitVec[15] = mSeqNumber & 0xff;
+ mAESInitVec[14] = (mSeqNumber >> 8) & 0xff;
+ mAESInitVec[13] = (mSeqNumber >> 16) & 0xff;
+ mAESInitVec[12] = (mSeqNumber >> 24) & 0xff;
}
- } else {
- memset(aes_ivec, 0, sizeof(aes_ivec));
- aes_ivec[15] = mSeqNumber & 0xff;
- aes_ivec[14] = (mSeqNumber >> 8) & 0xff;
- aes_ivec[13] = (mSeqNumber >> 16) & 0xff;
- aes_ivec[12] = (mSeqNumber >> 24) & 0xff;
}
AES_cbc_encrypt(
buffer->data(), buffer->data(), buffer->size(),
- &aes_key, aes_ivec, AES_DECRYPT);
-
- // hexdump(buffer->data(), buffer->size());
+ &aes_key, mAESInitVec, AES_DECRYPT);
- size_t n = buffer->size();
- CHECK_GT(n, 0u);
+ return OK;
+}
- size_t pad = buffer->data()[n - 1];
+status_t PlaylistFetcher::checkDecryptPadding(const sp<ABuffer> &buffer) {
+ status_t err;
+ AString method;
+ CHECK(buffer->meta()->findString("cipher-method", &method));
+ if (method == "NONE") {
+ return OK;
+ }
- CHECK_GT(pad, 0u);
- CHECK_LE(pad, 16u);
- CHECK_GE((size_t)n, pad);
- for (size_t i = 0; i < pad; ++i) {
- CHECK_EQ((unsigned)buffer->data()[n - 1 - i], pad);
+ uint8_t padding = 0;
+ if (buffer->size() > 0) {
+ padding = buffer->data()[buffer->size() - 1];
}
- n -= pad;
+ if (padding > 16) {
+ return ERROR_MALFORMED;
+ }
- buffer->setRange(buffer->offset(), n);
+ for (size_t i = buffer->size() - padding; i < padding; i++) {
+ if (buffer->data()[i] != padding) {
+ return ERROR_MALFORMED;
+ }
+ }
+ buffer->setRange(buffer->offset(), buffer->size() - padding);
return OK;
}
@@ -305,7 +331,9 @@ void PlaylistFetcher::startAsync(
const sp<AnotherPacketSource> &audioSource,
const sp<AnotherPacketSource> &videoSource,
const sp<AnotherPacketSource> &subtitleSource,
- int64_t startTimeUs) {
+ int64_t startTimeUs,
+ int64_t minStartTimeUs,
+ int32_t startSeqNumberHint) {
sp<AMessage> msg = new AMessage(kWhatStart, id());
uint32_t streamTypeMask = 0ul;
@@ -327,6 +355,8 @@ void PlaylistFetcher::startAsync(
msg->setInt32("streamTypeMask", streamTypeMask);
msg->setInt64("startTimeUs", startTimeUs);
+ msg->setInt64("minStartTimeUs", minStartTimeUs);
+ msg->setInt32("startSeqNumberHint", startSeqNumberHint);
msg->post();
}
@@ -338,6 +368,12 @@ void PlaylistFetcher::stopAsync() {
(new AMessage(kWhatStop, id()))->post();
}
+void PlaylistFetcher::resumeUntilAsync(const sp<AMessage> &params) {
+ AMessage* msg = new AMessage(kWhatResumeUntil, id());
+ msg->setMessage("params", params);
+ msg->post();
+}
+
void PlaylistFetcher::onMessageReceived(const sp<AMessage> &msg) {
switch (msg->what()) {
case kWhatStart:
@@ -372,6 +408,7 @@ void PlaylistFetcher::onMessageReceived(const sp<AMessage> &msg) {
}
case kWhatMonitorQueue:
+ case kWhatDownloadNext:
{
int32_t generation;
CHECK(msg->findInt32("generation", &generation));
@@ -381,7 +418,17 @@ void PlaylistFetcher::onMessageReceived(const sp<AMessage> &msg) {
break;
}
- onMonitorQueue();
+ if (msg->what() == kWhatMonitorQueue) {
+ onMonitorQueue();
+ } else {
+ onDownloadNext();
+ }
+ break;
+ }
+
+ case kWhatResumeUntil:
+ {
+ onResumeUntil(msg);
break;
}
@@ -397,7 +444,10 @@ status_t PlaylistFetcher::onStart(const sp<AMessage> &msg) {
CHECK(msg->findInt32("streamTypeMask", (int32_t *)&streamTypeMask));
int64_t startTimeUs;
+ int32_t startSeqNumberHint;
CHECK(msg->findInt64("startTimeUs", &startTimeUs));
+ CHECK(msg->findInt64("minStartTimeUs", (int64_t *) &mMinStartTimeUs));
+ CHECK(msg->findInt32("startSeqNumberHint", &startSeqNumberHint));
if (streamTypeMask & LiveSession::STREAMTYPE_AUDIO) {
void *ptr;
@@ -435,6 +485,10 @@ status_t PlaylistFetcher::onStart(const sp<AMessage> &msg) {
mPrepared = false;
}
+ if (startSeqNumberHint >= 0) {
+ mSeqNumber = startSeqNumberHint;
+ }
+
postMonitorQueue();
return OK;
@@ -442,20 +496,70 @@ status_t PlaylistFetcher::onStart(const sp<AMessage> &msg) {
void PlaylistFetcher::onPause() {
cancelMonitorQueue();
+}
+
+void PlaylistFetcher::onStop() {
+ cancelMonitorQueue();
mPacketSources.clear();
mStreamTypeMask = 0;
}
-void PlaylistFetcher::onStop() {
- cancelMonitorQueue();
+// Resume until we have reached the boundary timestamps listed in `msg`; when
+// the remaining time is too short (within a resume threshold) stop immediately
+// instead.
+status_t PlaylistFetcher::onResumeUntil(const sp<AMessage> &msg) {
+ sp<AMessage> params;
+ CHECK(msg->findMessage("params", &params));
+
+ bool stop = false;
+ for (size_t i = 0; i < mPacketSources.size(); i++) {
+ sp<AnotherPacketSource> packetSource = mPacketSources.valueAt(i);
+
+ const char *stopKey;
+ int streamType = mPacketSources.keyAt(i);
+ switch (streamType) {
+ case LiveSession::STREAMTYPE_VIDEO:
+ stopKey = "timeUsVideo";
+ break;
- for (size_t i = 0; i < mPacketSources.size(); ++i) {
- mPacketSources.valueAt(i)->clear();
+ case LiveSession::STREAMTYPE_AUDIO:
+ stopKey = "timeUsAudio";
+ break;
+
+ case LiveSession::STREAMTYPE_SUBTITLES:
+ stopKey = "timeUsSubtitle";
+ break;
+
+ default:
+ TRESPASS();
+ }
+
+ // Don't resume if we would stop within a resume threshold.
+ int64_t latestTimeUs = 0, stopTimeUs = 0;
+ sp<AMessage> latestMeta = packetSource->getLatestMeta();
+ if (latestMeta != NULL
+ && (latestMeta->findInt64("timeUs", &latestTimeUs)
+ && params->findInt64(stopKey, &stopTimeUs))) {
+ int64_t diffUs = stopTimeUs - latestTimeUs;
+ if (diffUs < resumeThreshold(latestMeta)) {
+ stop = true;
+ }
+ }
}
- mPacketSources.clear();
- mStreamTypeMask = 0;
+ if (stop) {
+ for (size_t i = 0; i < mPacketSources.size(); i++) {
+ mPacketSources.valueAt(i)->queueAccessUnit(mSession->createFormatChangeBuffer());
+ }
+ stopAsync();
+ return OK;
+ }
+
+ mStopParams = params;
+ postMonitorQueue();
+
+ return OK;
}
void PlaylistFetcher::notifyError(status_t err) {
@@ -499,8 +603,9 @@ void PlaylistFetcher::onMonitorQueue() {
packetSource->getBufferedDurationUs(&finalResult);
finalResult = OK;
} else {
- bool first = true;
-
+ // Use max stream duration to prevent us from waiting on a non-existent stream;
+ // when we cannot make out from the manifest what streams are included in a playlist
+ // we might assume extra streams.
for (size_t i = 0; i < mPacketSources.size(); ++i) {
if ((mStreamTypeMask & mPacketSources.keyAt(i)) == 0) {
continue;
@@ -508,9 +613,10 @@ void PlaylistFetcher::onMonitorQueue() {
int64_t bufferedStreamDurationUs =
mPacketSources.valueAt(i)->getBufferedDurationUs(&finalResult);
- if (first || bufferedStreamDurationUs < bufferedDurationUs) {
+ ALOGV("buffered %lld for stream %d",
+ bufferedStreamDurationUs, mPacketSources.keyAt(i));
+ if (bufferedStreamDurationUs > bufferedDurationUs) {
bufferedDurationUs = bufferedStreamDurationUs;
- first = false;
}
}
}
@@ -530,7 +636,12 @@ void PlaylistFetcher::onMonitorQueue() {
if (finalResult == OK && downloadMore) {
ALOGV("monitoring, buffered=%lld < %lld",
bufferedDurationUs, durationToBufferUs);
- onDownloadNext();
+ // delay the next download slightly; hopefully this gives other concurrent fetchers
+ // a better chance to run.
+ // onDownloadNext();
+ sp<AMessage> msg = new AMessage(kWhatDownloadNext, id());
+ msg->setInt32("generation", mMonitorQueueGeneration);
+ msg->post(1000l);
} else {
// Nothing to do yet, try again in a second.
@@ -597,6 +708,12 @@ void PlaylistFetcher::onDownloadNext() {
const int32_t lastSeqNumberInPlaylist =
firstSeqNumberInPlaylist + (int32_t)mPlaylist->size() - 1;
+ if (mStartup && mSeqNumber >= 0
+ && (mSeqNumber < firstSeqNumberInPlaylist || mSeqNumber > lastSeqNumberInPlaylist)) {
+ // in case we guessed wrong during reconfiguration, try fetching the latest content.
+ mSeqNumber = lastSeqNumberInPlaylist;
+ }
+
if (mSeqNumber < 0) {
CHECK_GE(mStartTimeUs, 0ll);
@@ -706,6 +823,9 @@ void PlaylistFetcher::onDownloadNext() {
CHECK(buffer != NULL);
err = decryptBuffer(mSeqNumber - firstSeqNumberInPlaylist, buffer);
+ if (err == OK) {
+ err = checkDecryptPadding(buffer);
+ }
if (err != OK) {
ALOGE("decryptBuffer failed w/ error %d", err);
@@ -738,6 +858,18 @@ void PlaylistFetcher::onDownloadNext() {
err = extractAndQueueAccessUnits(buffer, itemMeta);
+ if (err == -EAGAIN) {
+ // bad starting sequence number hint
+ postMonitorQueue();
+ return;
+ }
+
+ if (err == ERROR_OUT_OF_RANGE) {
+ // reached stopping point
+ stopAsync();
+ return;
+ }
+
if (err != OK) {
notifyError(err);
return;
@@ -794,12 +926,15 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits(
}
if (mTSParser == NULL) {
- mTSParser = new ATSParser;
+ // Use TS_TIMESTAMPS_ARE_ABSOLUTE so pts carry over between fetchers.
+ mTSParser = new ATSParser(ATSParser::TS_TIMESTAMPS_ARE_ABSOLUTE);
}
if (mNextPTSTimeUs >= 0ll) {
sp<AMessage> extra = new AMessage;
- extra->setInt64(IStreamListener::kKeyMediaTimeUs, mNextPTSTimeUs);
+ // Since we are using absolute timestamps, signal an offset of 0 to prevent
+ // ATSParser from skewing the timestamps of access units.
+ extra->setInt64(IStreamListener::kKeyMediaTimeUs, 0);
mTSParser->signalDiscontinuity(
ATSParser::DISCONTINUITY_SEEK, extra);
@@ -818,17 +953,23 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits(
offset += 188;
}
+ status_t err = OK;
for (size_t i = mPacketSources.size(); i-- > 0;) {
sp<AnotherPacketSource> packetSource = mPacketSources.valueAt(i);
+ const char *key;
ATSParser::SourceType type;
- switch (mPacketSources.keyAt(i)) {
+ const LiveSession::StreamType stream = mPacketSources.keyAt(i);
+ switch (stream) {
+
case LiveSession::STREAMTYPE_VIDEO:
type = ATSParser::VIDEO;
+ key = "timeUsVideo";
break;
case LiveSession::STREAMTYPE_AUDIO:
type = ATSParser::AUDIO;
+ key = "timeUsAudio";
break;
case LiveSession::STREAMTYPE_SUBTITLES:
@@ -855,19 +996,87 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits(
continue;
}
+ int64_t timeUs;
sp<ABuffer> accessUnit;
status_t finalResult;
while (source->hasBufferAvailable(&finalResult)
&& source->dequeueAccessUnit(&accessUnit) == OK) {
- // Note that we do NOT dequeue any discontinuities.
+
+ CHECK(accessUnit->meta()->findInt64("timeUs", &timeUs));
+ if (mMinStartTimeUs > 0) {
+ if (timeUs < mMinStartTimeUs) {
+ // TODO untested path
+ // try a later ts
+ int32_t targetDuration;
+ mPlaylist->meta()->findInt32("target-duration", &targetDuration);
+ int32_t incr = (mMinStartTimeUs - timeUs) / 1000000 / targetDuration;
+ if (incr == 0) {
+ // increment mSeqNumber by at least one
+ incr = 1;
+ }
+ mSeqNumber += incr;
+ err = -EAGAIN;
+ break;
+ } else {
+ int64_t startTimeUs;
+ if (mStartTimeUsNotify != NULL
+ && !mStartTimeUsNotify->findInt64(key, &startTimeUs)) {
+ mStartTimeUsNotify->setInt64(key, timeUs);
+
+ uint32_t streamMask = 0;
+ mStartTimeUsNotify->findInt32("streamMask", (int32_t *) &streamMask);
+ streamMask |= mPacketSources.keyAt(i);
+ mStartTimeUsNotify->setInt32("streamMask", streamMask);
+
+ if (streamMask == mStreamTypeMask) {
+ mStartTimeUsNotify->post();
+ mStartTimeUsNotify.clear();
+ }
+ }
+ }
+ }
+
+ if (mStopParams != NULL) {
+ // Queue discontinuity in original stream.
+ int64_t stopTimeUs;
+ if (!mStopParams->findInt64(key, &stopTimeUs) || timeUs >= stopTimeUs) {
+ packetSource->queueAccessUnit(mSession->createFormatChangeBuffer());
+ mStreamTypeMask &= ~stream;
+ mPacketSources.removeItemsAt(i);
+ break;
+ }
+ }
+
+ // Note that we do NOT dequeue any discontinuities except for format change.
// for simplicity, store a reference to the format in each unit
sp<MetaData> format = source->getFormat();
if (format != NULL) {
accessUnit->meta()->setObject("format", format);
}
+
+ // Stash the sequence number so we can hint future fetchers where to start at.
+ accessUnit->meta()->setInt32("seq", mSeqNumber);
packetSource->queueAccessUnit(accessUnit);
}
+
+ if (err != OK) {
+ break;
+ }
+ }
+
+ if (err != OK) {
+ for (size_t i = mPacketSources.size(); i-- > 0;) {
+ sp<AnotherPacketSource> packetSource = mPacketSources.valueAt(i);
+ packetSource->clear();
+ }
+ return err;
+ }
+
+ if (!mStreamTypeMask) {
+ // Signal gap is filled between original and new stream.
+ ALOGV("ERROR OUT OF RANGE");
+ return ERROR_OUT_OF_RANGE;
}
return OK;
@@ -884,6 +1093,7 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits(
CHECK(itemMeta->findInt64("durationUs", &durationUs));
buffer->meta()->setInt64("timeUs", getSegmentStartTimeUs(mSeqNumber));
buffer->meta()->setInt64("durationUs", durationUs);
+ buffer->meta()->setInt32("seq", mSeqNumber);
packetSource->queueAccessUnit(buffer);
return OK;
@@ -1020,6 +1230,7 @@ status_t PlaylistFetcher::extractAndQueueAccessUnits(
// Each AAC frame encodes 1024 samples.
numSamples += 1024;
+ unit->meta()->setInt32("seq", mSeqNumber);
packetSource->queueAccessUnit(unit);
offset += aac_frame_length;
@@ -1047,4 +1258,33 @@ void PlaylistFetcher::updateDuration() {
msg->post();
}
+int64_t PlaylistFetcher::resumeThreshold(const sp<AMessage> &msg) {
+ int64_t durationUs, threshold;
+ if (msg->findInt64("durationUs", &durationUs)) {
+ return kNumSkipFrames * durationUs;
+ }
+
+ sp<RefBase> obj;
+ msg->findObject("format", &obj);
+ MetaData *format = static_cast<MetaData *>(obj.get());
+
+ const char *mime;
+ CHECK(format->findCString(kKeyMIMEType, &mime));
+ bool audio = !strncasecmp(mime, "audio/", 6);
+ if (audio) {
+ // Assumes 1000 samples per frame.
+ int32_t sampleRate;
+ CHECK(format->findInt32(kKeySampleRate, &sampleRate));
+ return kNumSkipFrames /* frames */ * 1000 /* samples */
+ * (1000000 / sampleRate) /* sample duration (us) */;
+ } else {
+ int32_t frameRate;
+ if (format->findInt32(kKeyFrameRate, &frameRate) && frameRate > 0) {
+ return kNumSkipFrames * (1000000 / frameRate);
+ }
+ }
+
+ return 500000ll;
+}
+
} // namespace android
diff --git a/media/libstagefright/httplive/PlaylistFetcher.h b/media/libstagefright/httplive/PlaylistFetcher.h
index 78dea20..2e0349f 100644
--- a/media/libstagefright/httplive/PlaylistFetcher.h
+++ b/media/libstagefright/httplive/PlaylistFetcher.h
@@ -43,6 +43,7 @@ struct PlaylistFetcher : public AHandler {
kWhatTemporarilyDoneFetching,
kWhatPrepared,
kWhatPreparationFailed,
+ kWhatStartedAt,
};
PlaylistFetcher(
@@ -56,12 +57,16 @@ struct PlaylistFetcher : public AHandler {
const sp<AnotherPacketSource> &audioSource,
const sp<AnotherPacketSource> &videoSource,
const sp<AnotherPacketSource> &subtitleSource,
- int64_t startTimeUs = -1ll);
+ int64_t startTimeUs = -1ll,
+ int64_t minStartTimeUs = 0ll /* start after this timestamp */,
+ int32_t startSeqNumberHint = -1 /* try starting at this sequence number */);
void pauseAsync();
void stopAsync();
+ void resumeUntilAsync(const sp<AMessage> &params);
+
protected:
virtual ~PlaylistFetcher();
virtual void onMessageReceived(const sp<AMessage> &msg);
@@ -76,17 +81,25 @@ private:
kWhatPause = 'paus',
kWhatStop = 'stop',
kWhatMonitorQueue = 'moni',
+ kWhatResumeUntil = 'rsme',
+ kWhatDownloadNext = 'dlnx',
};
static const int64_t kMinBufferedDurationUs;
static const int64_t kMaxMonitorDelayUs;
+ static const int32_t kNumSkipFrames;
+ // notifications to mSession
sp<AMessage> mNotify;
+ sp<AMessage> mStartTimeUsNotify;
+
sp<LiveSession> mSession;
AString mURI;
uint32_t mStreamTypeMask;
int64_t mStartTimeUs;
+ int64_t mMinStartTimeUs; // start fetching no earlier than this value
+ sp<AMessage> mStopParams; // message containing the latest timestamps we should fetch.
KeyedVector<LiveSession::StreamType, sp<AnotherPacketSource> >
mPacketSources;
@@ -119,8 +132,23 @@ private:
uint64_t mFirstPTS;
int64_t mAbsoluteTimeAnchorUs;
+ // Stores the initialization vector to decrypt the next block of cipher text, which can
+ // either be derived from the sequence number, read from the manifest, or copied from
+ // the last block of cipher text (cipher-block chaining).
+ unsigned char mAESInitVec[16];
+
+ // Set first to true if decrypting the first segment of a playlist segment. When
+ // first is true, reset the initialization vector based on the available
+ // information in the manifest; otherwise, use the initialization vector as
+ // updated by the last call to AES_cbc_encrypt.
+ //
+ // For the input to decrypt correctly, decryptBuffer must be called on
+ // consecutive byte ranges on block boundaries, e.g. 0..15, 16..47, 48..63,
+ // and so on.
status_t decryptBuffer(
- size_t playlistIndex, const sp<ABuffer> &buffer);
+ size_t playlistIndex, const sp<ABuffer> &buffer,
+ bool first = true);
+ status_t checkDecryptPadding(const sp<ABuffer> &buffer);
void postMonitorQueue(int64_t delayUs = 0, int64_t minDelayUs = 0);
void cancelMonitorQueue();
@@ -138,6 +166,9 @@ private:
void onMonitorQueue();
void onDownloadNext();
+ // Resume a fetcher to continue until the stopping point stored in msg.
+ status_t onResumeUntil(const sp<AMessage> &msg);
+
status_t extractAndQueueAccessUnits(
const sp<ABuffer> &buffer, const sp<AMessage> &itemMeta);
@@ -150,6 +181,10 @@ private:
void updateDuration();
+ // Before resuming a fetcher in onResume, check the remaining duration is longer than that
+ // returned by resumeThreshold.
+ int64_t resumeThreshold(const sp<AMessage> &msg);
+
DISALLOW_EVIL_CONSTRUCTORS(PlaylistFetcher);
};