PlaylistFetcher.cpp revision ceeabe15f4e7bc73efdfcafed917202de9d515cb 
1/*
2 * Copyright (C) 2012 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 *      http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17//#define LOG_NDEBUG 0
18#define LOG_TAG "PlaylistFetcher"
19#include <utils/Log.h>
20
21#include "PlaylistFetcher.h"
22
23#include "LiveDataSource.h"
24#include "LiveSession.h"
25#include "M3UParser.h"
26
27#include "include/avc_utils.h"
28#include "include/HTTPBase.h"
29#include "include/ID3.h"
30#include "mpeg2ts/AnotherPacketSource.h"
31
32#include <media/IStreamSource.h>
33#include <media/stagefright/foundation/ABitReader.h>
34#include <media/stagefright/foundation/ABuffer.h>
35#include <media/stagefright/foundation/ADebug.h>
36#include <media/stagefright/foundation/hexdump.h>
37#include <media/stagefright/FileSource.h>
38#include <media/stagefright/MediaDefs.h>
39#include <media/stagefright/MetaData.h>
40#include <media/stagefright/Utils.h>
41
42#include <ctype.h>
43#include <openssl/aes.h>
44#include <openssl/md5.h>
45
46namespace android {
47
48// static
49const int64_t PlaylistFetcher::kMinBufferedDurationUs = 10000000ll;
50const int64_t PlaylistFetcher::kMaxMonitorDelayUs = 3000000ll;
51const int32_t PlaylistFetcher::kNumSkipFrames = 10;
52
53PlaylistFetcher::PlaylistFetcher(
54        const sp<AMessage> &notify,
55        const sp<LiveSession> &session,
56        const char *uri)
57    : mNotify(notify),
58      mStartTimeUsNotify(notify->dup()),
59      mSession(session),
60      mURI(uri),
61      mStreamTypeMask(0),
62      mStartTimeUs(-1ll),
63      mMinStartTimeUs(0ll),
64      mStopParams(NULL),
65      mLastPlaylistFetchTimeUs(-1ll),
66      mSeqNumber(-1),
67      mNumRetries(0),
68      mStartup(true),
69      mPrepared(false),
70      mNextPTSTimeUs(-1ll),
71      mMonitorQueueGeneration(0),
72      mRefreshState(INITIAL_MINIMUM_RELOAD_DELAY),
73      mFirstPTSValid(false),
74      mAbsoluteTimeAnchorUs(0ll) {
75    memset(mPlaylistHash, 0, sizeof(mPlaylistHash));
76    mStartTimeUsNotify->setInt32("what", kWhatStartedAt);
77    mStartTimeUsNotify->setInt32("streamMask", 0);
78}
79
80PlaylistFetcher::~PlaylistFetcher() {
81}
82
83int64_t PlaylistFetcher::getSegmentStartTimeUs(int32_t seqNumber) const {
84    CHECK(mPlaylist != NULL);
85
86    int32_t firstSeqNumberInPlaylist;
87    if (mPlaylist->meta() == NULL || !mPlaylist->meta()->findInt32(
88                "media-sequence", &firstSeqNumberInPlaylist)) {
89        firstSeqNumberInPlaylist = 0;
90    }
91
92    int32_t lastSeqNumberInPlaylist =
93        firstSeqNumberInPlaylist + (int32_t)mPlaylist->size() - 1;
94
95    CHECK_GE(seqNumber, firstSeqNumberInPlaylist);
96    CHECK_LE(seqNumber, lastSeqNumberInPlaylist);
97
98    int64_t segmentStartUs = 0ll;
99    for (int32_t index = 0;
100            index < seqNumber - firstSeqNumberInPlaylist; ++index) {
101        sp<AMessage> itemMeta;
102        CHECK(mPlaylist->itemAt(
103                    index, NULL /* uri */, &itemMeta));
104
105        int64_t itemDurationUs;
106        CHECK(itemMeta->findInt64("durationUs", &itemDurationUs));
107
108        segmentStartUs += itemDurationUs;
109    }
110
111    return segmentStartUs;
112}
113
114int64_t PlaylistFetcher::delayUsToRefreshPlaylist() const {
115    int64_t nowUs = ALooper::GetNowUs();
116
117    if (mPlaylist == NULL || mLastPlaylistFetchTimeUs < 0ll) {
118        CHECK_EQ((int)mRefreshState, (int)INITIAL_MINIMUM_RELOAD_DELAY);
119        return 0ll;
120    }
121
122    if (mPlaylist->isComplete()) {
123        return (~0llu >> 1);
124    }
125
126    int32_t targetDurationSecs;
127    CHECK(mPlaylist->meta()->findInt32("target-duration", &targetDurationSecs));
128
129    int64_t targetDurationUs = targetDurationSecs * 1000000ll;
130
131    int64_t minPlaylistAgeUs;
132
133    switch (mRefreshState) {
134        case INITIAL_MINIMUM_RELOAD_DELAY:
135        {
136            size_t n = mPlaylist->size();
137            if (n > 0) {
138                sp<AMessage> itemMeta;
139                CHECK(mPlaylist->itemAt(n - 1, NULL /* uri */, &itemMeta));
140
141                int64_t itemDurationUs;
142                CHECK(itemMeta->findInt64("durationUs", &itemDurationUs));
143
144                minPlaylistAgeUs = itemDurationUs;
145                break;
146            }
147
148            // fall through
149        }
150
151        case FIRST_UNCHANGED_RELOAD_ATTEMPT:
152        {
153            minPlaylistAgeUs = targetDurationUs / 2;
154            break;
155        }
156
157        case SECOND_UNCHANGED_RELOAD_ATTEMPT:
158        {
159            minPlaylistAgeUs = (targetDurationUs * 3) / 2;
160            break;
161        }
162
163        case THIRD_UNCHANGED_RELOAD_ATTEMPT:
164        {
165            minPlaylistAgeUs = targetDurationUs * 3;
166            break;
167        }
168
169        default:
170            TRESPASS();
171            break;
172    }
173
174    int64_t delayUs = mLastPlaylistFetchTimeUs + minPlaylistAgeUs - nowUs;
175    return delayUs > 0ll ? delayUs : 0ll;
176}
177
178status_t PlaylistFetcher::decryptBuffer(
179        size_t playlistIndex, const sp<ABuffer> &buffer,
180        bool first) {
181    sp<AMessage> itemMeta;
182    bool found = false;
183    AString method;
184
185    for (ssize_t i = playlistIndex; i >= 0; --i) {
186        AString uri;
187        CHECK(mPlaylist->itemAt(i, &uri, &itemMeta));
188
189        if (itemMeta->findString("cipher-method", &method)) {
190            found = true;
191            break;
192        }
193    }
194
195    if (!found) {
196        method = "NONE";
197    }
198    buffer->meta()->setString("cipher-method", method.c_str());
199
200    if (method == "NONE") {
201        return OK;
202    } else if (!(method == "AES-128")) {
203        ALOGE("Unsupported cipher method '%s'", method.c_str());
204        return ERROR_UNSUPPORTED;
205    }
206
207    AString keyURI;
208    if (!itemMeta->findString("cipher-uri", &keyURI)) {
209        ALOGE("Missing key uri");
210        return ERROR_MALFORMED;
211    }
212
213    ssize_t index = mAESKeyForURI.indexOfKey(keyURI);
214
215    sp<ABuffer> key;
216    if (index >= 0) {
217        key = mAESKeyForURI.valueAt(index);
218    } else {
219        status_t err = mSession->fetchFile(keyURI.c_str(), &key);
220
221        if (err != OK) {
222            ALOGE("failed to fetch cipher key from '%s'.", keyURI.c_str());
223            return ERROR_IO;
224        } else if (key->size() != 16) {
225            ALOGE("key file '%s' wasn't 16 bytes in size.", keyURI.c_str());
226            return ERROR_MALFORMED;
227        }
228
229        mAESKeyForURI.add(keyURI, key);
230    }
231
232    AES_KEY aes_key;
233    if (AES_set_decrypt_key(key->data(), 128, &aes_key) != 0) {
234        ALOGE("failed to set AES decryption key.");
235        return UNKNOWN_ERROR;
236    }
237
238    size_t n = buffer->size();
239    if (!n) {
240        return OK;
241    }
242    CHECK(n % 16 == 0);
243
244    if (first) {
245        // If decrypting the first block in a file, read the iv from the manifest
246        // or derive the iv from the file's sequence number.
247
248        AString iv;
249        if (itemMeta->findString("cipher-iv", &iv)) {
250            if ((!iv.startsWith("0x") && !iv.startsWith("0X"))
251                    || iv.size() != 16 * 2 + 2) {
252                ALOGE("malformed cipher IV '%s'.", iv.c_str());
253                return ERROR_MALFORMED;
254            }
255
256            memset(mAESInitVec, 0, sizeof(mAESInitVec));
257            for (size_t i = 0; i < 16; ++i) {
258                char c1 = tolower(iv.c_str()[2 + 2 * i]);
259                char c2 = tolower(iv.c_str()[3 + 2 * i]);
260                if (!isxdigit(c1) || !isxdigit(c2)) {
261                    ALOGE("malformed cipher IV '%s'.", iv.c_str());
262                    return ERROR_MALFORMED;
263                }
264                uint8_t nibble1 = isdigit(c1) ? c1 - '0' : c1 - 'a' + 10;
265                uint8_t nibble2 = isdigit(c2) ? c2 - '0' : c2 - 'a' + 10;
266
267                mAESInitVec[i] = nibble1 << 4 | nibble2;
268            }
269        } else {
270            memset(mAESInitVec, 0, sizeof(mAESInitVec));
271            mAESInitVec[15] = mSeqNumber & 0xff;
272            mAESInitVec[14] = (mSeqNumber >> 8) & 0xff;
273            mAESInitVec[13] = (mSeqNumber >> 16) & 0xff;
274            mAESInitVec[12] = (mSeqNumber >> 24) & 0xff;
275        }
276    }
277
278    AES_cbc_encrypt(
279            buffer->data(), buffer->data(), buffer->size(),
280            &aes_key, mAESInitVec, AES_DECRYPT);
281
282    return OK;
283}
284
285status_t PlaylistFetcher::checkDecryptPadding(const sp<ABuffer> &buffer) {
286    status_t err;
287    AString method;
288    CHECK(buffer->meta()->findString("cipher-method", &method));
289    if (method == "NONE") {
290        return OK;
291    }
292
293    uint8_t padding = 0;
294    if (buffer->size() > 0) {
295        padding = buffer->data()[buffer->size() - 1];
296    }
297
298    if (padding > 16) {
299        return ERROR_MALFORMED;
300    }
301
302    for (size_t i = buffer->size() - padding; i < padding; i++) {
303        if (buffer->data()[i] != padding) {
304            return ERROR_MALFORMED;
305        }
306    }
307
308    buffer->setRange(buffer->offset(), buffer->size() - padding);
309    return OK;
310}
311
312void PlaylistFetcher::postMonitorQueue(int64_t delayUs, int64_t minDelayUs) {
313    int64_t maxDelayUs = delayUsToRefreshPlaylist();
314    if (maxDelayUs < minDelayUs) {
315        maxDelayUs = minDelayUs;
316    }
317    if (delayUs > maxDelayUs) {
318        ALOGV("Need to refresh playlist in %lld", maxDelayUs);
319        delayUs = maxDelayUs;
320    }
321    sp<AMessage> msg = new AMessage(kWhatMonitorQueue, id());
322    msg->setInt32("generation", mMonitorQueueGeneration);
323    msg->post(delayUs);
324}
325
326void PlaylistFetcher::cancelMonitorQueue() {
327    ++mMonitorQueueGeneration;
328}
329
330void PlaylistFetcher::startAsync(
331        const sp<AnotherPacketSource> &audioSource,
332        const sp<AnotherPacketSource> &videoSource,
333        const sp<AnotherPacketSource> &subtitleSource,
334        int64_t startTimeUs,
335        int64_t minStartTimeUs,
336        int32_t startSeqNumberHint) {
337    sp<AMessage> msg = new AMessage(kWhatStart, id());
338
339    uint32_t streamTypeMask = 0ul;
340
341    if (audioSource != NULL) {
342        msg->setPointer("audioSource", audioSource.get());
343        streamTypeMask |= LiveSession::STREAMTYPE_AUDIO;
344    }
345
346    if (videoSource != NULL) {
347        msg->setPointer("videoSource", videoSource.get());
348        streamTypeMask |= LiveSession::STREAMTYPE_VIDEO;
349    }
350
351    if (subtitleSource != NULL) {
352        msg->setPointer("subtitleSource", subtitleSource.get());
353        streamTypeMask |= LiveSession::STREAMTYPE_SUBTITLES;
354    }
355
356    msg->setInt32("streamTypeMask", streamTypeMask);
357    msg->setInt64("startTimeUs", startTimeUs);
358    msg->setInt64("minStartTimeUs", minStartTimeUs);
359    msg->setInt32("startSeqNumberHint", startSeqNumberHint);
360    msg->post();
361}
362
363void PlaylistFetcher::pauseAsync() {
364    (new AMessage(kWhatPause, id()))->post();
365}
366
367void PlaylistFetcher::stopAsync(bool selfTriggered) {
368    sp<AMessage> msg = new AMessage(kWhatStop, id());
369    msg->setInt32("selfTriggered", selfTriggered);
370    msg->post();
371}
372
373void PlaylistFetcher::resumeUntilAsync(const sp<AMessage> &params) {
374    AMessage* msg = new AMessage(kWhatResumeUntil, id());
375    msg->setMessage("params", params);
376    msg->post();
377}
378
379void PlaylistFetcher::onMessageReceived(const sp<AMessage> &msg) {
380    switch (msg->what()) {
381        case kWhatStart:
382        {
383            status_t err = onStart(msg);
384
385            sp<AMessage> notify = mNotify->dup();
386            notify->setInt32("what", kWhatStarted);
387            notify->setInt32("err", err);
388            notify->post();
389            break;
390        }
391
392        case kWhatPause:
393        {
394            onPause();
395
396            sp<AMessage> notify = mNotify->dup();
397            notify->setInt32("what", kWhatPaused);
398            notify->post();
399            break;
400        }
401
402        case kWhatStop:
403        {
404            onStop(msg);
405
406            sp<AMessage> notify = mNotify->dup();
407            notify->setInt32("what", kWhatStopped);
408            notify->post();
409            break;
410        }
411
412        case kWhatMonitorQueue:
413        case kWhatDownloadNext:
414        {
415            int32_t generation;
416            CHECK(msg->findInt32("generation", &generation));
417
418            if (generation != mMonitorQueueGeneration) {
419                // Stale event
420                break;
421            }
422
423            if (msg->what() == kWhatMonitorQueue) {
424                onMonitorQueue();
425            } else {
426                onDownloadNext();
427            }
428            break;
429        }
430
431        case kWhatResumeUntil:
432        {
433            onResumeUntil(msg);
434            break;
435        }
436
437        default:
438            TRESPASS();
439    }
440}
441
442status_t PlaylistFetcher::onStart(const sp<AMessage> &msg) {
443    mPacketSources.clear();
444
445    uint32_t streamTypeMask;
446    CHECK(msg->findInt32("streamTypeMask", (int32_t *)&streamTypeMask));
447
448    int64_t startTimeUs;
449    int32_t startSeqNumberHint;
450    CHECK(msg->findInt64("startTimeUs", &startTimeUs));
451    CHECK(msg->findInt64("minStartTimeUs", (int64_t *) &mMinStartTimeUs));
452    CHECK(msg->findInt32("startSeqNumberHint", &startSeqNumberHint));
453
454    if (streamTypeMask & LiveSession::STREAMTYPE_AUDIO) {
455        void *ptr;
456        CHECK(msg->findPointer("audioSource", &ptr));
457
458        mPacketSources.add(
459                LiveSession::STREAMTYPE_AUDIO,
460                static_cast<AnotherPacketSource *>(ptr));
461    }
462
463    if (streamTypeMask & LiveSession::STREAMTYPE_VIDEO) {
464        void *ptr;
465        CHECK(msg->findPointer("videoSource", &ptr));
466
467        mPacketSources.add(
468                LiveSession::STREAMTYPE_VIDEO,
469                static_cast<AnotherPacketSource *>(ptr));
470    }
471
472    if (streamTypeMask & LiveSession::STREAMTYPE_SUBTITLES) {
473        void *ptr;
474        CHECK(msg->findPointer("subtitleSource", &ptr));
475
476        mPacketSources.add(
477                LiveSession::STREAMTYPE_SUBTITLES,
478                static_cast<AnotherPacketSource *>(ptr));
479    }
480
481    mStreamTypeMask = streamTypeMask;
482    mStartTimeUs = startTimeUs;
483
484    if (mStartTimeUs >= 0ll) {
485        mSeqNumber = -1;
486        mStartup = true;
487        mPrepared = false;
488    }
489
490    if (startSeqNumberHint >= 0) {
491        mSeqNumber = startSeqNumberHint;
492    }
493
494    postMonitorQueue();
495
496    return OK;
497}
498
499void PlaylistFetcher::onPause() {
500    cancelMonitorQueue();
501}
502
503void PlaylistFetcher::onStop(const sp<AMessage> &msg) {
504    cancelMonitorQueue();
505
506    int32_t selfTriggered;
507    CHECK(msg->findInt32("selfTriggered", &selfTriggered));
508    if (!selfTriggered) {
509        // Self triggered stops only happen during switching, in which case we do not want
510        // to clear the discontinuities queued at the end of packet sources.
511        for (size_t i = 0; i < mPacketSources.size(); i++) {
512            sp<AnotherPacketSource> packetSource = mPacketSources.valueAt(i);
513            packetSource->clear();
514        }
515    }
516
517    mPacketSources.clear();
518    mStreamTypeMask = 0;
519}
520
521// Resume until we have reached the boundary timestamps listed in `msg`; when
522// the remaining time is too short (within a resume threshold) stop immediately
523// instead.
524status_t PlaylistFetcher::onResumeUntil(const sp<AMessage> &msg) {
525    sp<AMessage> params;
526    CHECK(msg->findMessage("params", &params));
527
528    bool stop = false;
529    for (size_t i = 0; i < mPacketSources.size(); i++) {
530        sp<AnotherPacketSource> packetSource = mPacketSources.valueAt(i);
531
532        const char *stopKey;
533        int streamType = mPacketSources.keyAt(i);
534        switch (streamType) {
535        case LiveSession::STREAMTYPE_VIDEO:
536            stopKey = "timeUsVideo";
537            break;
538
539        case LiveSession::STREAMTYPE_AUDIO:
540            stopKey = "timeUsAudio";
541            break;
542
543        case LiveSession::STREAMTYPE_SUBTITLES:
544            stopKey = "timeUsSubtitle";
545            break;
546
547        default:
548            TRESPASS();
549        }
550
551        // Don't resume if we would stop within a resume threshold.
552        int64_t latestTimeUs = 0, stopTimeUs = 0;
553        sp<AMessage> latestMeta = packetSource->getLatestMeta();
554        if (latestMeta != NULL
555                && (latestMeta->findInt64("timeUs", &latestTimeUs)
556                && params->findInt64(stopKey, &stopTimeUs))) {
557            int64_t diffUs = stopTimeUs - latestTimeUs;
558            if (diffUs < resumeThreshold(latestMeta)) {
559                stop = true;
560            }
561        }
562    }
563
564    if (stop) {
565        for (size_t i = 0; i < mPacketSources.size(); i++) {
566            mPacketSources.valueAt(i)->queueAccessUnit(mSession->createFormatChangeBuffer());
567        }
568        stopAsync(/* selfTriggered = */ true);
569        return OK;
570    }
571
572    mStopParams = params;
573    postMonitorQueue();
574
575    return OK;
576}
577
578void PlaylistFetcher::notifyError(status_t err) {
579    sp<AMessage> notify = mNotify->dup();
580    notify->setInt32("what", kWhatError);
581    notify->setInt32("err", err);
582    notify->post();
583}
584
585void PlaylistFetcher::queueDiscontinuity(
586        ATSParser::DiscontinuityType type, const sp<AMessage> &extra) {
587    for (size_t i = 0; i < mPacketSources.size(); ++i) {
588        mPacketSources.valueAt(i)->queueDiscontinuity(type, extra);
589    }
590}
591
592void PlaylistFetcher::onMonitorQueue() {
593    bool downloadMore = false;
594    refreshPlaylist();
595
596    int32_t targetDurationSecs;
597    int64_t targetDurationUs = kMinBufferedDurationUs;
598    if (mPlaylist != NULL) {
599        CHECK(mPlaylist->meta()->findInt32("target-duration", &targetDurationSecs));
600        targetDurationUs = targetDurationSecs * 1000000ll;
601    }
602
603    // buffer at least 3 times the target duration, or up to 10 seconds
604    int64_t durationToBufferUs = targetDurationUs * 3;
605    if (durationToBufferUs > kMinBufferedDurationUs)  {
606        durationToBufferUs = kMinBufferedDurationUs;
607    }
608
609    int64_t bufferedDurationUs = 0ll;
610    status_t finalResult = NOT_ENOUGH_DATA;
611    if (mStreamTypeMask == LiveSession::STREAMTYPE_SUBTITLES) {
612        sp<AnotherPacketSource> packetSource =
613            mPacketSources.valueFor(LiveSession::STREAMTYPE_SUBTITLES);
614
615        bufferedDurationUs =
616                packetSource->getBufferedDurationUs(&finalResult);
617        finalResult = OK;
618    } else {
619        // Use max stream duration to prevent us from waiting on a non-existent stream;
620        // when we cannot make out from the manifest what streams are included in a playlist
621        // we might assume extra streams.
622        for (size_t i = 0; i < mPacketSources.size(); ++i) {
623            if ((mStreamTypeMask & mPacketSources.keyAt(i)) == 0) {
624                continue;
625            }
626
627            int64_t bufferedStreamDurationUs =
628                mPacketSources.valueAt(i)->getBufferedDurationUs(&finalResult);
629            ALOGV("buffered %lld for stream %d",
630                    bufferedStreamDurationUs, mPacketSources.keyAt(i));
631            if (bufferedStreamDurationUs > bufferedDurationUs) {
632                bufferedDurationUs = bufferedStreamDurationUs;
633            }
634        }
635    }
636    downloadMore = (bufferedDurationUs < durationToBufferUs);
637
638    // signal start if buffered up at least the target size
639    if (!mPrepared && bufferedDurationUs > targetDurationUs && downloadMore) {
640        mPrepared = true;
641
642        ALOGV("prepared, buffered=%lld > %lld",
643                bufferedDurationUs, targetDurationUs);
644        sp<AMessage> msg = mNotify->dup();
645        msg->setInt32("what", kWhatTemporarilyDoneFetching);
646        msg->post();
647    }
648
649    if (finalResult == OK && downloadMore) {
650        ALOGV("monitoring, buffered=%lld < %lld",
651                bufferedDurationUs, durationToBufferUs);
652        // delay the next download slightly; hopefully this gives other concurrent fetchers
653        // a better chance to run.
654        // onDownloadNext();
655        sp<AMessage> msg = new AMessage(kWhatDownloadNext, id());
656        msg->setInt32("generation", mMonitorQueueGeneration);
657        msg->post(1000l);
658    } else {
659        // Nothing to do yet, try again in a second.
660
661        sp<AMessage> msg = mNotify->dup();
662        msg->setInt32("what", kWhatTemporarilyDoneFetching);
663        msg->post();
664
665        int64_t delayUs = mPrepared ? kMaxMonitorDelayUs : targetDurationUs / 2;
666        ALOGV("pausing for %lld, buffered=%lld > %lld",
667                delayUs, bufferedDurationUs, durationToBufferUs);
668        // :TRICKY: need to enforce minimum delay because the delay to
669        // refresh the playlist will become 0
670        postMonitorQueue(delayUs, mPrepared ? targetDurationUs * 2 : 0);
671    }
672}
673
674status_t PlaylistFetcher::refreshPlaylist() {
675    if (delayUsToRefreshPlaylist() <= 0) {
676        bool unchanged;
677        sp<M3UParser> playlist = mSession->fetchPlaylist(
678                mURI.c_str(), mPlaylistHash, &unchanged);
679
680        if (playlist == NULL) {
681            if (unchanged) {
682                // We succeeded in fetching the playlist, but it was
683                // unchanged from the last time we tried.
684
685                if (mRefreshState != THIRD_UNCHANGED_RELOAD_ATTEMPT) {
686                    mRefreshState = (RefreshState)(mRefreshState + 1);
687                }
688            } else {
689                ALOGE("failed to load playlist at url '%s'", mURI.c_str());
690                notifyError(ERROR_IO);
691                return ERROR_IO;
692            }
693        } else {
694            mRefreshState = INITIAL_MINIMUM_RELOAD_DELAY;
695            mPlaylist = playlist;
696
697            if (mPlaylist->isComplete() || mPlaylist->isEvent()) {
698                updateDuration();
699            }
700        }
701
702        mLastPlaylistFetchTimeUs = ALooper::GetNowUs();
703    }
704    return OK;
705}
706
707void PlaylistFetcher::onDownloadNext() {
708    if (refreshPlaylist() != OK) {
709        return;
710    }
711
712    int32_t firstSeqNumberInPlaylist;
713    if (mPlaylist->meta() == NULL || !mPlaylist->meta()->findInt32(
714                "media-sequence", &firstSeqNumberInPlaylist)) {
715        firstSeqNumberInPlaylist = 0;
716    }
717
718    bool seekDiscontinuity = false;
719    bool explicitDiscontinuity = false;
720
721    const int32_t lastSeqNumberInPlaylist =
722        firstSeqNumberInPlaylist + (int32_t)mPlaylist->size() - 1;
723
724    if (mStartup && mSeqNumber >= 0
725            && (mSeqNumber < firstSeqNumberInPlaylist || mSeqNumber > lastSeqNumberInPlaylist)) {
726        // in case we guessed wrong during reconfiguration, try fetching the latest content.
727        mSeqNumber = lastSeqNumberInPlaylist;
728    }
729
730    if (mSeqNumber < 0) {
731        CHECK_GE(mStartTimeUs, 0ll);
732
733        if (mPlaylist->isComplete() || mPlaylist->isEvent()) {
734            mSeqNumber = getSeqNumberForTime(mStartTimeUs);
735            ALOGV("Initial sequence number for time %lld is %ld from (%ld .. %ld)",
736                    mStartTimeUs, mSeqNumber, firstSeqNumberInPlaylist,
737                    lastSeqNumberInPlaylist);
738        } else {
739            // If this is a live session, start 3 segments from the end.
740            mSeqNumber = lastSeqNumberInPlaylist - 3;
741            if (mSeqNumber < firstSeqNumberInPlaylist) {
742                mSeqNumber = firstSeqNumberInPlaylist;
743            }
744            ALOGV("Initial sequence number for live event %ld from (%ld .. %ld)",
745                    mSeqNumber, firstSeqNumberInPlaylist,
746                    lastSeqNumberInPlaylist);
747        }
748
749        mStartTimeUs = -1ll;
750    }
751
752    if (mSeqNumber < firstSeqNumberInPlaylist
753            || mSeqNumber > lastSeqNumberInPlaylist) {
754        if (!mPlaylist->isComplete() && mNumRetries < kMaxNumRetries) {
755            ++mNumRetries;
756
757            if (mSeqNumber > lastSeqNumberInPlaylist) {
758                // refresh in increasing fraction (1/2, 1/3, ...) of the
759                // playlist's target duration or 3 seconds, whichever is less
760                int32_t targetDurationSecs;
761                CHECK(mPlaylist->meta()->findInt32(
762                        "target-duration", &targetDurationSecs));
763                int64_t delayUs = mPlaylist->size() * targetDurationSecs *
764                        1000000ll / (1 + mNumRetries);
765                if (delayUs > kMaxMonitorDelayUs) {
766                    delayUs = kMaxMonitorDelayUs;
767                }
768                ALOGV("sequence number high: %ld from (%ld .. %ld), monitor in %lld (retry=%d)",
769                        mSeqNumber, firstSeqNumberInPlaylist,
770                        lastSeqNumberInPlaylist, delayUs, mNumRetries);
771                postMonitorQueue(delayUs);
772                return;
773            }
774
775            // we've missed the boat, let's start from the lowest sequence
776            // number available and signal a discontinuity.
777
778            ALOGI("We've missed the boat, restarting playback."
779                  "  mStartup=%d, was  looking for %d in %d-%d",
780                    mStartup, mSeqNumber, firstSeqNumberInPlaylist,
781                    lastSeqNumberInPlaylist);
782            mSeqNumber = lastSeqNumberInPlaylist - 3;
783            if (mSeqNumber < firstSeqNumberInPlaylist) {
784                mSeqNumber = firstSeqNumberInPlaylist;
785            }
786            explicitDiscontinuity = true;
787
788            // fall through
789        } else {
790            ALOGE("Cannot find sequence number %d in playlist "
791                 "(contains %d - %d)",
792                 mSeqNumber, firstSeqNumberInPlaylist,
793                 firstSeqNumberInPlaylist + mPlaylist->size() - 1);
794
795            notifyError(ERROR_END_OF_STREAM);
796            return;
797        }
798    }
799
800    mNumRetries = 0;
801
802    AString uri;
803    sp<AMessage> itemMeta;
804    CHECK(mPlaylist->itemAt(
805                mSeqNumber - firstSeqNumberInPlaylist,
806                &uri,
807                &itemMeta));
808
809    int32_t val;
810    if (itemMeta->findInt32("discontinuity", &val) && val != 0) {
811        explicitDiscontinuity = true;
812    }
813
814    int64_t range_offset, range_length;
815    if (!itemMeta->findInt64("range-offset", &range_offset)
816            || !itemMeta->findInt64("range-length", &range_length)) {
817        range_offset = 0;
818        range_length = -1;
819    }
820
821    ALOGV("fetching segment %d from (%d .. %d)",
822          mSeqNumber, firstSeqNumberInPlaylist, lastSeqNumberInPlaylist);
823
824    ALOGV("fetching '%s'", uri.c_str());
825
826    sp<ABuffer> buffer;
827    status_t err = mSession->fetchFile(
828            uri.c_str(), &buffer, range_offset, range_length);
829
830    if (err != OK) {
831        ALOGE("failed to fetch .ts segment at url '%s'", uri.c_str());
832        notifyError(err);
833        return;
834    }
835
836    CHECK(buffer != NULL);
837
838    err = decryptBuffer(mSeqNumber - firstSeqNumberInPlaylist, buffer);
839    if (err == OK) {
840        err = checkDecryptPadding(buffer);
841    }
842
843    if (err != OK) {
844        ALOGE("decryptBuffer failed w/ error %d", err);
845
846        notifyError(err);
847        return;
848    }
849
850    if (mStartup || seekDiscontinuity || explicitDiscontinuity) {
851        // Signal discontinuity.
852
853        if (mPlaylist->isComplete() || mPlaylist->isEvent()) {
854            // If this was a live event this made no sense since
855            // we don't have access to all the segment before the current
856            // one.
857            mNextPTSTimeUs = getSegmentStartTimeUs(mSeqNumber);
858        }
859
860        if (seekDiscontinuity || explicitDiscontinuity) {
861            ALOGI("queueing discontinuity (seek=%d, explicit=%d)",
862                 seekDiscontinuity, explicitDiscontinuity);
863
864            queueDiscontinuity(
865                    explicitDiscontinuity
866                        ? ATSParser::DISCONTINUITY_FORMATCHANGE
867                        : ATSParser::DISCONTINUITY_SEEK,
868                    NULL /* extra */);
869        }
870    }
871
872    err = extractAndQueueAccessUnits(buffer, itemMeta);
873
874    if (err == -EAGAIN) {
875        // bad starting sequence number hint
876        postMonitorQueue();
877        return;
878    }
879
880    if (err == ERROR_OUT_OF_RANGE) {
881        // reached stopping point
882        stopAsync(/* selfTriggered = */ true);
883        return;
884    }
885
886    if (err != OK) {
887        notifyError(err);
888        return;
889    }
890
891    ++mSeqNumber;
892
893    postMonitorQueue();
894
895    mStartup = false;
896}
897
898int32_t PlaylistFetcher::getSeqNumberForTime(int64_t timeUs) const {
899    int32_t firstSeqNumberInPlaylist;
900    if (mPlaylist->meta() == NULL || !mPlaylist->meta()->findInt32(
901                "media-sequence", &firstSeqNumberInPlaylist)) {
902        firstSeqNumberInPlaylist = 0;
903    }
904
905    size_t index = 0;
906    int64_t segmentStartUs = 0;
907    while (index < mPlaylist->size()) {
908        sp<AMessage> itemMeta;
909        CHECK(mPlaylist->itemAt(
910                    index, NULL /* uri */, &itemMeta));
911
912        int64_t itemDurationUs;
913        CHECK(itemMeta->findInt64("durationUs", &itemDurationUs));
914
915        if (timeUs < segmentStartUs + itemDurationUs) {
916            break;
917        }
918
919        segmentStartUs += itemDurationUs;
920        ++index;
921    }
922
923    if (index >= mPlaylist->size()) {
924        index = mPlaylist->size() - 1;
925    }
926
927    return firstSeqNumberInPlaylist + index;
928}
929
930status_t PlaylistFetcher::extractAndQueueAccessUnits(
931        const sp<ABuffer> &buffer, const sp<AMessage> &itemMeta) {
932    if (buffer->size() > 0 && buffer->data()[0] == 0x47) {
933        // Let's assume this is an MPEG2 transport stream.
934
935        if ((buffer->size() % 188) != 0) {
936            ALOGE("MPEG2 transport stream is not an even multiple of 188 "
937                  "bytes in length.");
938            return ERROR_MALFORMED;
939        }
940
941        if (mTSParser == NULL) {
942            // Use TS_TIMESTAMPS_ARE_ABSOLUTE so pts carry over between fetchers.
943            mTSParser = new ATSParser(ATSParser::TS_TIMESTAMPS_ARE_ABSOLUTE);
944        }
945
946        if (mNextPTSTimeUs >= 0ll) {
947            sp<AMessage> extra = new AMessage;
948            // Since we are using absolute timestamps, signal an offset of 0 to prevent
949            // ATSParser from skewing the timestamps of access units.
950            extra->setInt64(IStreamListener::kKeyMediaTimeUs, 0);
951
952            mTSParser->signalDiscontinuity(
953                    ATSParser::DISCONTINUITY_SEEK, extra);
954
955            mNextPTSTimeUs = -1ll;
956        }
957
958        size_t offset = 0;
959        while (offset < buffer->size()) {
960            status_t err = mTSParser->feedTSPacket(buffer->data() + offset, 188);
961
962            if (err != OK) {
963                return err;
964            }
965
966            offset += 188;
967        }
968
969        status_t err = OK;
970        for (size_t i = mPacketSources.size(); i-- > 0;) {
971            sp<AnotherPacketSource> packetSource = mPacketSources.valueAt(i);
972
973            const char *key;
974            ATSParser::SourceType type;
975            const LiveSession::StreamType stream = mPacketSources.keyAt(i);
976            switch (stream) {
977
978                case LiveSession::STREAMTYPE_VIDEO:
979                    type = ATSParser::VIDEO;
980                    key = "timeUsVideo";
981                    break;
982
983                case LiveSession::STREAMTYPE_AUDIO:
984                    type = ATSParser::AUDIO;
985                    key = "timeUsAudio";
986                    break;
987
988                case LiveSession::STREAMTYPE_SUBTITLES:
989                {
990                    ALOGE("MPEG2 Transport streams do not contain subtitles.");
991                    return ERROR_MALFORMED;
992                    break;
993                }
994
995                default:
996                    TRESPASS();
997            }
998
999            sp<AnotherPacketSource> source =
1000                static_cast<AnotherPacketSource *>(
1001                        mTSParser->getSource(type).get());
1002
1003            if (source == NULL) {
1004                ALOGW("MPEG2 Transport stream does not contain %s data.",
1005                      type == ATSParser::VIDEO ? "video" : "audio");
1006
1007                mStreamTypeMask &= ~mPacketSources.keyAt(i);
1008                mPacketSources.removeItemsAt(i);
1009                continue;
1010            }
1011
1012            int64_t timeUs;
1013            sp<ABuffer> accessUnit;
1014            status_t finalResult;
1015            while (source->hasBufferAvailable(&finalResult)
1016                    && source->dequeueAccessUnit(&accessUnit) == OK) {
1017
1018                CHECK(accessUnit->meta()->findInt64("timeUs", &timeUs));
1019                if (mMinStartTimeUs > 0) {
1020                    if (timeUs < mMinStartTimeUs) {
1021                        // TODO untested path
1022                        // try a later ts
1023                        int32_t targetDuration;
1024                        mPlaylist->meta()->findInt32("target-duration", &targetDuration);
1025                        int32_t incr = (mMinStartTimeUs - timeUs) / 1000000 / targetDuration;
1026                        if (incr == 0) {
1027                            // increment mSeqNumber by at least one
1028                            incr = 1;
1029                        }
1030                        mSeqNumber += incr;
1031                        err = -EAGAIN;
1032                        break;
1033                    } else {
1034                        int64_t startTimeUs;
1035                        if (mStartTimeUsNotify != NULL
1036                                && !mStartTimeUsNotify->findInt64(key, &startTimeUs)) {
1037                            mStartTimeUsNotify->setInt64(key, timeUs);
1038
1039                            uint32_t streamMask = 0;
1040                            mStartTimeUsNotify->findInt32("streamMask", (int32_t *) &streamMask);
1041                            streamMask |= mPacketSources.keyAt(i);
1042                            mStartTimeUsNotify->setInt32("streamMask", streamMask);
1043
1044                            if (streamMask == mStreamTypeMask) {
1045                                mStartTimeUsNotify->post();
1046                                mStartTimeUsNotify.clear();
1047                            }
1048                        }
1049                    }
1050                }
1051
1052                if (mStopParams != NULL) {
1053                    // Queue discontinuity in original stream.
1054                    int64_t stopTimeUs;
1055                    if (!mStopParams->findInt64(key, &stopTimeUs) || timeUs >= stopTimeUs) {
1056                        packetSource->queueAccessUnit(mSession->createFormatChangeBuffer());
1057                        mStreamTypeMask &= ~stream;
1058                        mPacketSources.removeItemsAt(i);
1059                        break;
1060                    }
1061                }
1062
1063                // Note that we do NOT dequeue any discontinuities except for format change.
1064
1065                // for simplicity, store a reference to the format in each unit
1066                sp<MetaData> format = source->getFormat();
1067                if (format != NULL) {
1068                    accessUnit->meta()->setObject("format", format);
1069                }
1070
1071                // Stash the sequence number so we can hint future fetchers where to start at.
1072                accessUnit->meta()->setInt32("seq", mSeqNumber);
1073                packetSource->queueAccessUnit(accessUnit);
1074            }
1075
1076            if (err != OK) {
1077                break;
1078            }
1079        }
1080
1081        if (err != OK) {
1082            for (size_t i = mPacketSources.size(); i-- > 0;) {
1083                sp<AnotherPacketSource> packetSource = mPacketSources.valueAt(i);
1084                packetSource->clear();
1085            }
1086            return err;
1087        }
1088
1089        if (!mStreamTypeMask) {
1090            // Signal gap is filled between original and new stream.
1091            ALOGV("ERROR OUT OF RANGE");
1092            return ERROR_OUT_OF_RANGE;
1093        }
1094
1095        return OK;
1096    } else if (buffer->size() >= 7 && !memcmp("WEBVTT\n", buffer->data(), 7)) {
1097        if (mStreamTypeMask != LiveSession::STREAMTYPE_SUBTITLES) {
1098            ALOGE("This stream only contains subtitles.");
1099            return ERROR_MALFORMED;
1100        }
1101
1102        const sp<AnotherPacketSource> packetSource =
1103            mPacketSources.valueFor(LiveSession::STREAMTYPE_SUBTITLES);
1104
1105        int64_t durationUs;
1106        CHECK(itemMeta->findInt64("durationUs", &durationUs));
1107        buffer->meta()->setInt64("timeUs", getSegmentStartTimeUs(mSeqNumber));
1108        buffer->meta()->setInt64("durationUs", durationUs);
1109        buffer->meta()->setInt32("seq", mSeqNumber);
1110
1111        packetSource->queueAccessUnit(buffer);
1112        return OK;
1113    }
1114
1115    if (mNextPTSTimeUs >= 0ll) {
1116        mFirstPTSValid = false;
1117        mAbsoluteTimeAnchorUs = mNextPTSTimeUs;
1118        mNextPTSTimeUs = -1ll;
1119    }
1120
1121    // This better be an ISO 13818-7 (AAC) or ISO 13818-1 (MPEG) audio
1122    // stream prefixed by an ID3 tag.
1123
1124    bool firstID3Tag = true;
1125    uint64_t PTS = 0;
1126
1127    for (;;) {
1128        // Make sure to skip all ID3 tags preceding the audio data.
1129        // At least one must be present to provide the PTS timestamp.
1130
1131        ID3 id3(buffer->data(), buffer->size(), true /* ignoreV1 */);
1132        if (!id3.isValid()) {
1133            if (firstID3Tag) {
1134                ALOGE("Unable to parse ID3 tag.");
1135                return ERROR_MALFORMED;
1136            } else {
1137                break;
1138            }
1139        }
1140
1141        if (firstID3Tag) {
1142            bool found = false;
1143
1144            ID3::Iterator it(id3, "PRIV");
1145            while (!it.done()) {
1146                size_t length;
1147                const uint8_t *data = it.getData(&length);
1148
1149                static const char *kMatchName =
1150                    "com.apple.streaming.transportStreamTimestamp";
1151                static const size_t kMatchNameLen = strlen(kMatchName);
1152
1153                if (length == kMatchNameLen + 1 + 8
1154                        && !strncmp((const char *)data, kMatchName, kMatchNameLen)) {
1155                    found = true;
1156                    PTS = U64_AT(&data[kMatchNameLen + 1]);
1157                }
1158
1159                it.next();
1160            }
1161
1162            if (!found) {
1163                ALOGE("Unable to extract transportStreamTimestamp from ID3 tag.");
1164                return ERROR_MALFORMED;
1165            }
1166        }
1167
1168        // skip the ID3 tag
1169        buffer->setRange(
1170                buffer->offset() + id3.rawSize(), buffer->size() - id3.rawSize());
1171
1172        firstID3Tag = false;
1173    }
1174
1175    if (!mFirstPTSValid) {
1176        mFirstPTSValid = true;
1177        mFirstPTS = PTS;
1178    }
1179    PTS -= mFirstPTS;
1180
1181    int64_t timeUs = (PTS * 100ll) / 9ll + mAbsoluteTimeAnchorUs;
1182
1183    if (mStreamTypeMask != LiveSession::STREAMTYPE_AUDIO) {
1184        ALOGW("This stream only contains audio data!");
1185
1186        mStreamTypeMask &= LiveSession::STREAMTYPE_AUDIO;
1187
1188        if (mStreamTypeMask == 0) {
1189            return OK;
1190        }
1191    }
1192
1193    sp<AnotherPacketSource> packetSource =
1194        mPacketSources.valueFor(LiveSession::STREAMTYPE_AUDIO);
1195
1196    if (packetSource->getFormat() == NULL && buffer->size() >= 7) {
1197        ABitReader bits(buffer->data(), buffer->size());
1198
1199        // adts_fixed_header
1200
1201        CHECK_EQ(bits.getBits(12), 0xfffu);
1202        bits.skipBits(3);  // ID, layer
1203        bool protection_absent = bits.getBits(1) != 0;
1204
1205        unsigned profile = bits.getBits(2);
1206        CHECK_NE(profile, 3u);
1207        unsigned sampling_freq_index = bits.getBits(4);
1208        bits.getBits(1);  // private_bit
1209        unsigned channel_configuration = bits.getBits(3);
1210        CHECK_NE(channel_configuration, 0u);
1211        bits.skipBits(2);  // original_copy, home
1212
1213        sp<MetaData> meta = MakeAACCodecSpecificData(
1214                profile, sampling_freq_index, channel_configuration);
1215
1216        meta->setInt32(kKeyIsADTS, true);
1217
1218        packetSource->setFormat(meta);
1219    }
1220
1221    int64_t numSamples = 0ll;
1222    int32_t sampleRate;
1223    CHECK(packetSource->getFormat()->findInt32(kKeySampleRate, &sampleRate));
1224
1225    size_t offset = 0;
1226    while (offset < buffer->size()) {
1227        const uint8_t *adtsHeader = buffer->data() + offset;
1228        CHECK_LT(offset + 5, buffer->size());
1229
1230        unsigned aac_frame_length =
1231            ((adtsHeader[3] & 3) << 11)
1232            | (adtsHeader[4] << 3)
1233            | (adtsHeader[5] >> 5);
1234
1235        CHECK_LE(offset + aac_frame_length, buffer->size());
1236
1237        sp<ABuffer> unit = new ABuffer(aac_frame_length);
1238        memcpy(unit->data(), adtsHeader, aac_frame_length);
1239
1240        int64_t unitTimeUs = timeUs + numSamples * 1000000ll / sampleRate;
1241        unit->meta()->setInt64("timeUs", unitTimeUs);
1242
1243        // Each AAC frame encodes 1024 samples.
1244        numSamples += 1024;
1245
1246        unit->meta()->setInt32("seq", mSeqNumber);
1247        packetSource->queueAccessUnit(unit);
1248
1249        offset += aac_frame_length;
1250    }
1251
1252    return OK;
1253}
1254
1255void PlaylistFetcher::updateDuration() {
1256    int64_t durationUs = 0ll;
1257    for (size_t index = 0; index < mPlaylist->size(); ++index) {
1258        sp<AMessage> itemMeta;
1259        CHECK(mPlaylist->itemAt(
1260                    index, NULL /* uri */, &itemMeta));
1261
1262        int64_t itemDurationUs;
1263        CHECK(itemMeta->findInt64("durationUs", &itemDurationUs));
1264
1265        durationUs += itemDurationUs;
1266    }
1267
1268    sp<AMessage> msg = mNotify->dup();
1269    msg->setInt32("what", kWhatDurationUpdate);
1270    msg->setInt64("durationUs", durationUs);
1271    msg->post();
1272}
1273
1274int64_t PlaylistFetcher::resumeThreshold(const sp<AMessage> &msg) {
1275    int64_t durationUs, threshold;
1276    if (msg->findInt64("durationUs", &durationUs)) {
1277        return kNumSkipFrames * durationUs;
1278    }
1279
1280    sp<RefBase> obj;
1281    msg->findObject("format", &obj);
1282    MetaData *format = static_cast<MetaData *>(obj.get());
1283
1284    const char *mime;
1285    CHECK(format->findCString(kKeyMIMEType, &mime));
1286    bool audio = !strncasecmp(mime, "audio/", 6);
1287    if (audio) {
1288        // Assumes 1000 samples per frame.
1289        int32_t sampleRate;
1290        CHECK(format->findInt32(kKeySampleRate, &sampleRate));
1291        return kNumSkipFrames  /* frames */ * 1000 /* samples */
1292                * (1000000 / sampleRate) /* sample duration (us) */;
1293    } else {
1294        int32_t frameRate;
1295        if (format->findInt32(kKeyFrameRate, &frameRate) && frameRate > 0) {
1296            return kNumSkipFrames * (1000000 / frameRate);
1297        }
1298    }
1299
1300    return 500000ll;
1301}
1302
1303}  // namespace android
1304