blob: c1c3e2b708873949ba735c6a5db99ce909c76b63 [file] [log] [blame]
/*
* 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 "LiveSession"
#include <utils/Log.h>
#include "LiveSession.h"
#include "M3UParser.h"
#include "PlaylistFetcher.h"
#include "include/HTTPBase.h"
#include "mpeg2ts/AnotherPacketSource.h"
#include <cutils/properties.h>
#include <media/stagefright/foundation/hexdump.h>
#include <media/stagefright/foundation/ABuffer.h>
#include <media/stagefright/foundation/ADebug.h>
#include <media/stagefright/foundation/AMessage.h>
#include <media/stagefright/DataSource.h>
#include <media/stagefright/FileSource.h>
#include <media/stagefright/MediaErrors.h>
#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>
namespace android {
LiveSession::LiveSession(
const sp<AMessage> &notify, uint32_t flags, bool uidValid, uid_t uid)
: mNotify(notify),
mFlags(flags),
mUIDValid(uidValid),
mUID(uid),
mInPreparationPhase(true),
mHTTPDataSource(
HTTPBase::Create(
(mFlags & kFlagIncognito)
? HTTPBase::kFlagIncognito
: 0)),
mPrevBandwidthIndex(-1),
mStreamMask(0),
mNewStreamMask(0),
mSwapMask(0),
mCheckBandwidthGeneration(0),
mSwitchGeneration(0),
mLastDequeuedTimeUs(0ll),
mRealTimeBaseUs(0ll),
mReconfigurationInProgress(false),
mSwitchInProgress(false),
mDisconnectReplyID(0),
mSeekReplyID(0) {
if (mUIDValid) {
mHTTPDataSource->setUID(mUID);
}
mStreams[kAudioIndex] = StreamItem("audio");
mStreams[kVideoIndex] = StreamItem("video");
mStreams[kSubtitleIndex] = StreamItem("subtitle");
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 -EWOULDBLOCK to avoid halting the decoder
// when switching between audio/video and audio only.
return -EWOULDBLOCK;
}
sp<AnotherPacketSource> packetSource = mPacketSources.valueFor(stream);
status_t finalResult;
if (!packetSource->hasBufferAvailable(&finalResult)) {
return finalResult == OK ? -EAGAIN : finalResult;
}
status_t err = packetSource->dequeueAccessUnit(accessUnit);
const char *streamStr;
switch (stream) {
case STREAMTYPE_AUDIO:
streamStr = "audio";
break;
case STREAMTYPE_VIDEO:
streamStr = "video";
break;
case STREAMTYPE_SUBTITLES:
streamStr = "subs";
break;
default:
TRESPASS();
}
if (err == INFO_DISCONTINUITY) {
int32_t type;
CHECK((*accessUnit)->meta()->findInt32("discontinuity", &type));
sp<AMessage> extra;
if (!(*accessUnit)->meta()->findMessage("extra", &extra)) {
extra.clear();
}
ALOGI("[%s] read discontinuity of type %d, extra = %s",
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;
CHECK((*accessUnit)->meta()->findInt64("timeUs", &timeUs));
ALOGV("[%s] read buffer at time %lld us", streamStr, timeUs);
mLastDequeuedTimeUs = timeUs;
mRealTimeBaseUs = ALooper::GetNowUs() - timeUs;
} else if (stream == STREAMTYPE_SUBTITLES) {
(*accessUnit)->meta()->setInt32(
"trackIndex", mPlaylist->getSelectedIndex());
(*accessUnit)->meta()->setInt64("baseUs", mRealTimeBaseUs);
}
} else {
ALOGI("[%s] encountered error %d", streamStr, err);
}
return err;
}
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;
}
sp<AnotherPacketSource> packetSource = mPacketSources.valueFor(stream);
sp<MetaData> meta = packetSource->getFormat();
if (meta == NULL) {
return -EAGAIN;
}
return convertMetaDataToMessage(meta, format);
}
void LiveSession::connectAsync(
const char *url, const KeyedVector<String8, String8> *headers) {
sp<AMessage> msg = new AMessage(kWhatConnect, id());
msg->setString("url", url);
if (headers != NULL) {
msg->setPointer(
"headers",
new KeyedVector<String8, String8>(*headers));
}
msg->post();
}
status_t LiveSession::disconnect() {
sp<AMessage> msg = new AMessage(kWhatDisconnect, id());
sp<AMessage> response;
status_t err = msg->postAndAwaitResponse(&response);
return err;
}
status_t LiveSession::seekTo(int64_t timeUs) {
sp<AMessage> msg = new AMessage(kWhatSeek, id());
msg->setInt64("timeUs", timeUs);
sp<AMessage> response;
status_t err = msg->postAndAwaitResponse(&response);
uint32_t replyID;
CHECK(response == mSeekReply && 0 != mSeekReplyID);
mSeekReply.clear();
mSeekReplyID = 0;
return err;
}
void LiveSession::onMessageReceived(const sp<AMessage> &msg) {
switch (msg->what()) {
case kWhatConnect:
{
onConnect(msg);
break;
}
case kWhatDisconnect:
{
CHECK(msg->senderAwaitsResponse(&mDisconnectReplyID));
if (mReconfigurationInProgress) {
break;
}
finishDisconnect();
break;
}
case kWhatSeek:
{
CHECK(msg->senderAwaitsResponse(&mSeekReplyID));
status_t err = onSeek(msg);
mSeekReply = new AMessage;
mSeekReply->setInt32("err", err);
break;
}
case kWhatFetcherNotify:
{
int32_t what;
CHECK(msg->findInt32("what", &what));
switch (what) {
case PlaylistFetcher::kWhatStarted:
break;
case PlaylistFetcher::kWhatPaused:
case PlaylistFetcher::kWhatStopped:
{
if (what == PlaylistFetcher::kWhatStopped) {
AString uri;
CHECK(msg->findString("uri", &uri));
if (mFetcherInfos.removeItem(uri) < 0) {
// ignore duplicated kWhatStopped messages.
break;
}
tryToFinishBandwidthSwitch();
}
if (mContinuation != NULL) {
CHECK_GT(mContinuationCounter, 0);
if (--mContinuationCounter == 0) {
mContinuation->post();
if (mSeekReplyID != 0) {
CHECK(mSeekReply != NULL);
mSeekReply->postReply(mSeekReplyID);
}
}
}
break;
}
case PlaylistFetcher::kWhatDurationUpdate:
{
AString uri;
CHECK(msg->findString("uri", &uri));
int64_t durationUs;
CHECK(msg->findInt64("durationUs", &durationUs));
FetcherInfo *info = &mFetcherInfos.editValueFor(uri);
info->mDurationUs = durationUs;
break;
}
case PlaylistFetcher::kWhatError:
{
status_t err;
CHECK(msg->findInt32("err", &err));
ALOGE("XXX Received error %d from PlaylistFetcher.", err);
if (mInPreparationPhase) {
postPrepared(err);
}
cancelBandwidthSwitch();
mPacketSources.valueFor(STREAMTYPE_AUDIO)->signalEOS(err);
mPacketSources.valueFor(STREAMTYPE_VIDEO)->signalEOS(err);
mPacketSources.valueFor(
STREAMTYPE_SUBTITLES)->signalEOS(err);
sp<AMessage> notify = mNotify->dup();
notify->setInt32("what", kWhatError);
notify->setInt32("err", err);
notify->post();
break;
}
case PlaylistFetcher::kWhatTemporarilyDoneFetching:
{
AString uri;
CHECK(msg->findString("uri", &uri));
FetcherInfo *info = &mFetcherInfos.editValueFor(uri);
info->mIsPrepared = true;
if (mInPreparationPhase) {
bool allFetchersPrepared = true;
for (size_t i = 0; i < mFetcherInfos.size(); ++i) {
if (!mFetcherInfos.valueAt(i).mIsPrepared) {
allFetchersPrepared = false;
break;
}
}
if (allFetchersPrepared) {
postPrepared(OK);
}
}
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();
}
break;
}
case kWhatCheckBandwidth:
{
int32_t generation;
CHECK(msg->findInt32("generation", &generation));
if (generation != mCheckBandwidthGeneration) {
break;
}
onCheckBandwidth();
break;
}
case kWhatChangeConfiguration:
{
onChangeConfiguration(msg);
break;
}
case kWhatChangeConfiguration2:
{
onChangeConfiguration2(msg);
break;
}
case kWhatChangeConfiguration3:
{
onChangeConfiguration3(msg);
break;
}
case kWhatFinishDisconnect2:
{
onFinishDisconnect2();
break;
}
case kWhatSwapped:
{
onSwapped(msg);
break;
}
default:
TRESPASS();
break;
}
}
// static
int LiveSession::SortByBandwidth(const BandwidthItem *a, const BandwidthItem *b) {
if (a->mBandwidth < b->mBandwidth) {
return -1;
} else if (a->mBandwidth == b->mBandwidth) {
return 0;
}
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));
KeyedVector<String8, String8> *headers = NULL;
if (!msg->findPointer("headers", (void **)&headers)) {
mExtraHeaders.clear();
} else {
mExtraHeaders = *headers;
delete headers;
headers = NULL;
}
#if 1
ALOGI("onConnect <URL suppressed>");
#else
ALOGI("onConnect %s", url.c_str());
#endif
mMasterURL = url;
bool dummy;
mPlaylist = fetchPlaylist(url.c_str(), NULL /* curPlaylistHash */, &dummy);
if (mPlaylist == NULL) {
ALOGE("unable to fetch master playlist '%s'.", url.c_str());
postPrepared(ERROR_IO);
return;
}
// We trust the content provider to make a reasonable choice of preferred
// initial bandwidth by listing it first in the variant playlist.
// At startup we really don't have a good estimate on the available
// network bandwidth since we haven't tranferred any data yet. Once
// we have we can make a better informed choice.
size_t initialBandwidth = 0;
size_t initialBandwidthIndex = 0;
if (mPlaylist->isVariantPlaylist()) {
for (size_t i = 0; i < mPlaylist->size(); ++i) {
BandwidthItem item;
item.mPlaylistIndex = i;
sp<AMessage> meta;
AString uri;
mPlaylist->itemAt(i, &uri, &meta);
unsigned long bandwidth;
CHECK(meta->findInt32("bandwidth", (int32_t *)&item.mBandwidth));
if (initialBandwidth == 0) {
initialBandwidth = item.mBandwidth;
}
mBandwidthItems.push(item);
}
CHECK_GT(mBandwidthItems.size(), 0u);
mBandwidthItems.sort(SortByBandwidth);
for (size_t i = 0; i < mBandwidthItems.size(); ++i) {
if (mBandwidthItems.itemAt(i).mBandwidth == initialBandwidth) {
initialBandwidthIndex = i;
break;
}
}
} else {
// dummy item.
BandwidthItem item;
item.mPlaylistIndex = 0;
item.mBandwidth = 0;
mBandwidthItems.push(item);
}
changeConfiguration(
0ll /* timeUs */, initialBandwidthIndex, true /* pickTrack */);
}
void LiveSession::finishDisconnect() {
// No reconfiguration is currently pending, make sure none will trigger
// 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();
}
sp<AMessage> msg = new AMessage(kWhatFinishDisconnect2, id());
mContinuationCounter = mFetcherInfos.size();
mContinuation = msg;
if (mContinuationCounter == 0) {
msg->post();
}
}
void LiveSession::onFinishDisconnect2() {
mContinuation.clear();
mPacketSources.valueFor(STREAMTYPE_AUDIO)->signalEOS(ERROR_END_OF_STREAM);
mPacketSources.valueFor(STREAMTYPE_VIDEO)->signalEOS(ERROR_END_OF_STREAM);
mPacketSources.valueFor(
STREAMTYPE_SUBTITLES)->signalEOS(ERROR_END_OF_STREAM);
sp<AMessage> response = new AMessage;
response->setInt32("err", OK);
response->postReply(mDisconnectReplyID);
mDisconnectReplyID = 0;
}
sp<PlaylistFetcher> LiveSession::addFetcher(const char *uri) {
ssize_t index = mFetcherInfos.indexOfKey(uri);
if (index >= 0) {
return NULL;
}
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);
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,
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;
}
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 (err != OK) {
return err;
}
*source = mHTTPDataSource;
}
}
status_t getSizeErr = (*source)->getSize(&size);
if (getSizeErr != OK) {
size = 65536;
}
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 && getSizeErr != OK) {
bufferRemaining = 32768;
ALOGV("increasing download buffer to %d bytes",
buffer->size() + bufferRemaining);
sp<ABuffer> copy = new ABuffer(buffer->size() + bufferRemaining);
memcpy(copy->data(), buffer->data(), buffer->size());
copy->setRange(0, buffer->size());
buffer = copy;
}
size_t maxBytesToRead = bufferRemaining;
if (range_length >= 0) {
int64_t bytesLeftInRange = range_length - buffer->size();
if (bytesLeftInRange < maxBytesToRead) {
maxBytesToRead = bytesLeftInRange;
if (bytesLeftInRange == 0) {
break;
}
}
}
// 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);
if (n < 0) {
return n;
}
if (n == 0) {
break;
}
buffer->setRange(0, buffer->size() + (size_t)n);
}
*out = buffer;
if (actualUrl != NULL) {
*actualUrl = (*source)->getUri();
if (actualUrl->isEmpty()) {
*actualUrl = url;
}
}
return OK;
}
sp<M3UParser> LiveSession::fetchPlaylist(
const char *url, uint8_t *curPlaylistHash, bool *unchanged) {
ALOGV("fetchPlaylist '%s'", url);
*unchanged = false;
sp<ABuffer> buffer;
String8 actualUrl;
status_t err = fetchFile(url, &buffer, 0, -1, 0, NULL, &actualUrl);
if (err != OK) {
return NULL;
}
// MD5 functionality is not available on the simulator, treat all
// playlists as changed.
#if defined(HAVE_ANDROID_OS)
uint8_t hash[16];
MD5_CTX m;
MD5_Init(&m);
MD5_Update(&m, buffer->data(), buffer->size());
MD5_Final(hash, &m);
if (curPlaylistHash != NULL && !memcmp(hash, curPlaylistHash, 16)) {
// playlist unchanged
*unchanged = true;
ALOGV("Playlist unchanged, refresh state is now %d",
(int)mRefreshState);
return NULL;
}
if (curPlaylistHash != NULL) {
memcpy(curPlaylistHash, hash, sizeof(hash));
}
#endif
sp<M3UParser> playlist =
new M3UParser(actualUrl.string(), buffer->data(), buffer->size());
if (playlist->initCheck() != OK) {
ALOGE("failed to parse .m3u8 playlist");
return NULL;
}
return playlist;
}
static double uniformRand() {
return (double)rand() / RAND_MAX;
}
size_t LiveSession::getBandwidthIndex() {
if (mBandwidthItems.size() == 0) {
return 0;
}
#if 1
char value[PROPERTY_VALUE_MAX];
ssize_t index = -1;
if (property_get("media.httplive.bw-index", value, NULL)) {
char *end;
index = strtol(value, &end, 10);
CHECK(end > value && *end == '\0');
if (index >= 0 && (size_t)index >= mBandwidthItems.size()) {
index = mBandwidthItems.size() - 1;
}
}
if (index < 0) {
int32_t bandwidthBps;
if (mHTTPDataSource != NULL
&& mHTTPDataSource->estimateBandwidth(&bandwidthBps)) {
ALOGV("bandwidth estimated at %.2f kbps", bandwidthBps / 1024.0f);
} else {
ALOGV("no bandwidth estimate.");
return 0; // Pick the lowest bandwidth stream by default.
}
char value[PROPERTY_VALUE_MAX];
if (property_get("media.httplive.max-bw", value, NULL)) {
char *end;
long maxBw = strtoul(value, &end, 10);
if (end > value && *end == '\0') {
if (maxBw > 0 && bandwidthBps > maxBw) {
ALOGV("bandwidth capped to %ld bps", maxBw);
bandwidthBps = maxBw;
}
}
}
// Consider only 80% of the available bandwidth usable.
bandwidthBps = (bandwidthBps * 8) / 10;
// Pick the highest bandwidth stream below or equal to estimated bandwidth.
index = mBandwidthItems.size() - 1;
while (index > 0 && mBandwidthItems.itemAt(index).mBandwidth
> (size_t)bandwidthBps) {
--index;
}
}
#elif 0
// Change bandwidth at random()
size_t index = uniformRand() * mBandwidthItems.size();
#elif 0
// There's a 50% chance to stay on the current bandwidth and
// a 50% chance to switch to the next higher bandwidth (wrapping around
// to lowest)
const size_t kMinIndex = 0;
static ssize_t mPrevBandwidthIndex = -1;
size_t index;
if (mPrevBandwidthIndex < 0) {
index = kMinIndex;
} else if (uniformRand() < 0.5) {
index = (size_t)mPrevBandwidthIndex;
} else {
index = mPrevBandwidthIndex + 1;
if (index == mBandwidthItems.size()) {
index = kMinIndex;
}
}
mPrevBandwidthIndex = index;
#elif 0
// Pick the highest bandwidth stream below or equal to 1.2 Mbit/sec
size_t index = mBandwidthItems.size() - 1;
while (index > 0 && mBandwidthItems.itemAt(index).mBandwidth > 1200000) {
--index;
}
#elif 1
char value[PROPERTY_VALUE_MAX];
size_t index;
if (property_get("media.httplive.bw-index", value, NULL)) {
char *end;
index = strtoul(value, &end, 10);
CHECK(end > value && *end == '\0');
if (index >= mBandwidthItems.size()) {
index = mBandwidthItems.size() - 1;
}
} else {
index = 0;
}
#else
size_t index = mBandwidthItems.size() - 1; // Highest bandwidth stream
#endif
CHECK_GE(index, 0);
return index;
}
status_t LiveSession::onSeek(const sp<AMessage> &msg) {
int64_t timeUs;
CHECK(msg->findInt64("timeUs", &timeUs));
if (!mReconfigurationInProgress) {
changeConfiguration(timeUs, getBandwidthIndex());
}
return OK;
}
status_t LiveSession::getDuration(int64_t *durationUs) const {
int64_t maxDurationUs = 0ll;
for (size_t i = 0; i < mFetcherInfos.size(); ++i) {
int64_t fetcherDurationUs = mFetcherInfos.valueAt(i).mDurationUs;
if (fetcherDurationUs >= 0ll && fetcherDurationUs > maxDurationUs) {
maxDurationUs = fetcherDurationUs;
}
}
*durationUs = maxDurationUs;
return OK;
}
bool LiveSession::isSeekable() const {
int64_t durationUs;
return getDuration(&durationUs) == OK && durationUs >= 0;
}
bool LiveSession::hasDynamicDuration() const {
return false;
}
status_t LiveSession::getTrackInfo(Parcel *reply) const {
return mPlaylist->getTrackInfo(reply);
}
status_t LiveSession::selectTrack(size_t index, bool select) {
status_t err = mPlaylist->selectTrack(index, select);
if (err == OK) {
(new AMessage(kWhatChangeConfiguration, id()))->post();
}
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;
mPrevBandwidthIndex = bandwidthIndex;
ALOGV("changeConfiguration => timeUs:%lld us, bwIndex:%d, pickTrack:%d",
timeUs, bandwidthIndex, pickTrack);
if (pickTrack) {
mPlaylist->pickRandomMediaItems();
}
CHECK_LT(bandwidthIndex, mBandwidthItems.size());
const BandwidthItem &item = mBandwidthItems.itemAt(bandwidthIndex);
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 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.
// Pause those that we'll reuse.
for (size_t i = 0; i < mFetcherInfos.size(); ++i) {
const AString &uri = mFetcherInfos.keyAt(i);
bool discardFetcher = true;
// If we're seeking all current fetchers are discarded.
if (timeUs < 0ll) {
// 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;
}
}
}
if (discardFetcher) {
mFetcherInfos.valueAt(i).mFetcher->stopAsync();
} else {
mFetcherInfos.valueAt(i).mFetcher->pauseAsync();
}
}
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);
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
// we'll decrement mContinuationCounter, once it reaches zero, i.e. all
// fetchers have completed their asynchronous operation, we'll post
// mContinuation, which then is handled below in onChangeConfiguration2.
mContinuationCounter = mFetcherInfos.size();
mContinuation = msg;
if (mContinuationCounter == 0) {
msg->post();
if (mSeekReplyID != 0) {
CHECK(mSeekReply != NULL);
mSeekReply->postReply(mSeekReplyID);
}
}
}
void LiveSession::onChangeConfiguration(const sp<AMessage> &msg) {
if (!mReconfigurationInProgress) {
changeConfiguration(-1ll /* timeUs */, getBandwidthIndex());
} else {
msg->post(1000000ll); // retry in 1 sec
}
}
void LiveSession::onChangeConfiguration2(const sp<AMessage> &msg) {
mContinuation.clear();
// All fetchers are either suspended or have been removed now.
uint32_t streamMask;
CHECK(msg->findInt32("streamMask", (int32_t *)&streamMask));
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,
// a decoder has to be shutdown if either
// 1) its streamtype was active before but now longer isn't.
// or
// 2) its streamtype was already active and still is but the URI
// has changed.
uint32_t changedMask = 0;
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) {
// If nothing changed as far as the audio/video decoders
// are concerned we can proceed.
onChangeConfiguration3(msg);
return;
}
// Something changed, inform the player which will shutdown the
// corresponding decoders and will post the reply once that's done.
// Handling the reply will continue executing below in
// onChangeConfiguration3.
sp<AMessage> notify = mNotify->dup();
notify->setInt32("what", kWhatStreamsChanged);
notify->setInt32("changedMask", changedMask);
msg->setWhat(kWhatChangeConfiguration3);
msg->setTarget(id());
notify->setMessage("reply", msg);
notify->post();
}
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, resumeMask;
CHECK(msg->findInt32("streamMask", (int32_t *)&streamMask));
CHECK(msg->findInt32("resumeMask", (int32_t *)&resumeMask));
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;
mNewStreamMask = streamMask;
// 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);
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));
}
}
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;
}
}
// streamMask now only contains the types that need a new fetcher created.
if (streamMask != 0) {
ALOGV("creating new fetchers for mask 0x%08x", streamMask);
}
// 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;
uri = mStreams[i].mUri;
sp<PlaylistFetcher> fetcher = addFetcher(uri.c_str());
CHECK(fetcher != NULL);
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;
}
}
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 &= ~indexToType(j);
}
}
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
// has completed.
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);
msg->post(10000000ll);
}
void LiveSession::cancelCheckBandwidthEvent() {
++mCheckBandwidthGeneration;
}
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 (canSwitchBandwidthTo(bandwidthIndex)) {
changeConfiguration(-1ll /* timeUs */, bandwidthIndex);
} else {
scheduleCheckBandwidthEvent();
}
// Handling the kWhatCheckBandwidth even here does _not_ automatically
// schedule another one on return, only an explicit call to
// scheduleCheckBandwidthEvent will do that.
// This ensures that only one configuration change is ongoing at any
// one time, once that completes it'll schedule another check bandwidth
// event.
}
void LiveSession::postPrepared(status_t err) {
CHECK(mInPreparationPhase);
sp<AMessage> notify = mNotify->dup();
if (err == OK || err == ERROR_END_OF_STREAM) {
notify->setInt32("what", kWhatPrepared);
} else {
notify->setInt32("what", kWhatPreparationFailed);
notify->setInt32("err", err);
}
notify->post();
mInPreparationPhase = false;
}
} // namespace android