LiveSession.cpp revision 7314fa17093d514199fedcb55ac41136a1b31cb3 
1/*
2 * Copyright (C) 2010 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 "LiveSession"
19#include <utils/Log.h>
20
21#include "include/LiveSession.h"
22
23#include "LiveDataSource.h"
24
25#include "include/M3UParser.h"
26#include "include/NuHTTPDataSource.h"
27
28#include <cutils/properties.h>
29#include <media/stagefright/foundation/hexdump.h>
30#include <media/stagefright/foundation/ABuffer.h>
31#include <media/stagefright/foundation/ADebug.h>
32#include <media/stagefright/foundation/AMessage.h>
33#include <media/stagefright/DataSource.h>
34#include <media/stagefright/FileSource.h>
35#include <media/stagefright/MediaErrors.h>
36
37#include <ctype.h>
38#include <openssl/aes.h>
39
40namespace android {
41
42const int64_t LiveSession::kMaxPlaylistAgeUs = 15000000ll;
43
44LiveSession::LiveSession(uint32_t flags)
45    : mFlags(flags),
46      mDataSource(new LiveDataSource),
47      mHTTPDataSource(
48              new NuHTTPDataSource(
49                  (mFlags & kFlagIncognito)
50                    ? NuHTTPDataSource::kFlagIncognito
51                    : 0)),
52      mPrevBandwidthIndex(-1),
53      mLastPlaylistFetchTimeUs(-1),
54      mSeqNumber(-1),
55      mSeekTimeUs(-1),
56      mNumRetries(0),
57      mDurationUs(-1),
58      mSeekDone(false),
59      mDisconnectPending(false),
60      mMonitorQueueGeneration(0) {
61}
62
63LiveSession::~LiveSession() {
64}
65
66sp<DataSource> LiveSession::getDataSource() {
67    return mDataSource;
68}
69
70void LiveSession::connect(const char *url) {
71    sp<AMessage> msg = new AMessage(kWhatConnect, id());
72    msg->setString("url", url);
73    msg->post();
74}
75
76void LiveSession::disconnect() {
77    Mutex::Autolock autoLock(mLock);
78    mDisconnectPending = true;
79
80    mHTTPDataSource->disconnect();
81
82    (new AMessage(kWhatDisconnect, id()))->post();
83}
84
85void LiveSession::seekTo(int64_t timeUs) {
86    Mutex::Autolock autoLock(mLock);
87    mSeekDone = false;
88
89    sp<AMessage> msg = new AMessage(kWhatSeek, id());
90    msg->setInt64("timeUs", timeUs);
91    msg->post();
92
93    while (!mSeekDone) {
94        mCondition.wait(mLock);
95    }
96}
97
98void LiveSession::onMessageReceived(const sp<AMessage> &msg) {
99    switch (msg->what()) {
100        case kWhatConnect:
101            onConnect(msg);
102            break;
103
104        case kWhatDisconnect:
105            onDisconnect();
106            break;
107
108        case kWhatMonitorQueue:
109        {
110            int32_t generation;
111            CHECK(msg->findInt32("generation", &generation));
112
113            if (generation != mMonitorQueueGeneration) {
114                // Stale event
115                break;
116            }
117
118            onMonitorQueue();
119            break;
120        }
121
122        case kWhatSeek:
123            onSeek(msg);
124            break;
125
126        default:
127            TRESPASS();
128            break;
129    }
130}
131
132// static
133int LiveSession::SortByBandwidth(const BandwidthItem *a, const BandwidthItem *b) {
134    if (a->mBandwidth < b->mBandwidth) {
135        return -1;
136    } else if (a->mBandwidth == b->mBandwidth) {
137        return 0;
138    }
139
140    return 1;
141}
142
143void LiveSession::onConnect(const sp<AMessage> &msg) {
144    AString url;
145    CHECK(msg->findString("url", &url));
146
147    if (!(mFlags & kFlagIncognito)) {
148        LOGI("onConnect '%s'", url.c_str());
149    } else {
150        LOGI("onConnect <URL suppressed>");
151    }
152
153    mMasterURL = url;
154
155    sp<M3UParser> playlist = fetchPlaylist(url.c_str());
156
157    if (playlist == NULL) {
158        LOGE("unable to fetch master playlist '%s'.", url.c_str());
159
160        mDataSource->queueEOS(ERROR_IO);
161        return;
162    }
163
164    if (playlist->isVariantPlaylist()) {
165        for (size_t i = 0; i < playlist->size(); ++i) {
166            BandwidthItem item;
167
168            sp<AMessage> meta;
169            playlist->itemAt(i, &item.mURI, &meta);
170
171            unsigned long bandwidth;
172            CHECK(meta->findInt32("bandwidth", (int32_t *)&item.mBandwidth));
173
174            mBandwidthItems.push(item);
175        }
176
177        CHECK_GT(mBandwidthItems.size(), 0u);
178
179        mBandwidthItems.sort(SortByBandwidth);
180    }
181
182    postMonitorQueue();
183}
184
185void LiveSession::onDisconnect() {
186    LOGI("onDisconnect");
187
188    mDataSource->queueEOS(ERROR_END_OF_STREAM);
189
190    Mutex::Autolock autoLock(mLock);
191    mDisconnectPending = false;
192}
193
194status_t LiveSession::fetchFile(const char *url, sp<ABuffer> *out) {
195    *out = NULL;
196
197    sp<DataSource> source;
198
199    if (!strncasecmp(url, "file://", 7)) {
200        source = new FileSource(url + 7);
201    } else if (strncasecmp(url, "http://", 7)
202            && strncasecmp(url, "https://", 8)) {
203        return ERROR_UNSUPPORTED;
204    } else {
205        {
206            Mutex::Autolock autoLock(mLock);
207
208            if (mDisconnectPending) {
209                return ERROR_IO;
210            }
211        }
212
213        status_t err = mHTTPDataSource->connect(url);
214
215        if (err != OK) {
216            return err;
217        }
218
219        source = mHTTPDataSource;
220    }
221
222    off64_t size;
223    status_t err = source->getSize(&size);
224
225    if (err != OK) {
226        size = 65536;
227    }
228
229    sp<ABuffer> buffer = new ABuffer(size);
230    buffer->setRange(0, 0);
231
232    for (;;) {
233        size_t bufferRemaining = buffer->capacity() - buffer->size();
234
235        if (bufferRemaining == 0) {
236            bufferRemaining = 32768;
237
238            LOGV("increasing download buffer to %d bytes",
239                 buffer->size() + bufferRemaining);
240
241            sp<ABuffer> copy = new ABuffer(buffer->size() + bufferRemaining);
242            memcpy(copy->data(), buffer->data(), buffer->size());
243            copy->setRange(0, buffer->size());
244
245            buffer = copy;
246        }
247
248        ssize_t n = source->readAt(
249                buffer->size(), buffer->data() + buffer->size(),
250                bufferRemaining);
251
252        if (n < 0) {
253            return n;
254        }
255
256        if (n == 0) {
257            break;
258        }
259
260        buffer->setRange(0, buffer->size() + (size_t)n);
261    }
262
263    *out = buffer;
264
265    return OK;
266}
267
268sp<M3UParser> LiveSession::fetchPlaylist(const char *url) {
269    sp<ABuffer> buffer;
270    status_t err = fetchFile(url, &buffer);
271
272    if (err != OK) {
273        return NULL;
274    }
275
276    sp<M3UParser> playlist =
277        new M3UParser(url, buffer->data(), buffer->size());
278
279    if (playlist->initCheck() != OK) {
280        return NULL;
281    }
282
283    return playlist;
284}
285
286static double uniformRand() {
287    return (double)rand() / RAND_MAX;
288}
289
290size_t LiveSession::getBandwidthIndex() {
291    if (mBandwidthItems.size() == 0) {
292        return 0;
293    }
294
295#if 1
296    int32_t bandwidthBps;
297    if (mHTTPDataSource != NULL
298            && mHTTPDataSource->estimateBandwidth(&bandwidthBps)) {
299        LOGV("bandwidth estimated at %.2f kbps", bandwidthBps / 1024.0f);
300    } else {
301        LOGV("no bandwidth estimate.");
302        return 0;  // Pick the lowest bandwidth stream by default.
303    }
304
305    char value[PROPERTY_VALUE_MAX];
306    if (property_get("media.httplive.max-bw", value, NULL)) {
307        char *end;
308        long maxBw = strtoul(value, &end, 10);
309        if (end > value && *end == '\0') {
310            if (maxBw > 0 && bandwidthBps > maxBw) {
311                LOGV("bandwidth capped to %ld bps", maxBw);
312                bandwidthBps = maxBw;
313            }
314        }
315    }
316
317    // Consider only 80% of the available bandwidth usable.
318    bandwidthBps = (bandwidthBps * 8) / 10;
319
320    // Pick the highest bandwidth stream below or equal to estimated bandwidth.
321
322    size_t index = mBandwidthItems.size() - 1;
323    while (index > 0 && mBandwidthItems.itemAt(index).mBandwidth
324                            > (size_t)bandwidthBps) {
325        --index;
326    }
327#elif 0
328    // Change bandwidth at random()
329    size_t index = uniformRand() * mBandwidthItems.size();
330#elif 0
331    // There's a 50% chance to stay on the current bandwidth and
332    // a 50% chance to switch to the next higher bandwidth (wrapping around
333    // to lowest)
334    const size_t kMinIndex = 0;
335
336    size_t index;
337    if (mPrevBandwidthIndex < 0) {
338        index = kMinIndex;
339    } else if (uniformRand() < 0.5) {
340        index = (size_t)mPrevBandwidthIndex;
341    } else {
342        index = mPrevBandwidthIndex + 1;
343        if (index == mBandwidthItems.size()) {
344            index = kMinIndex;
345        }
346    }
347#elif 0
348    // Pick the highest bandwidth stream below or equal to 1.2 Mbit/sec
349
350    size_t index = mBandwidthItems.size() - 1;
351    while (index > 0 && mBandwidthItems.itemAt(index).mBandwidth > 1200000) {
352        --index;
353    }
354#else
355    size_t index = mBandwidthItems.size() - 1;  // Highest bandwidth stream
356#endif
357
358    return index;
359}
360
361void LiveSession::onDownloadNext() {
362    size_t bandwidthIndex = getBandwidthIndex();
363
364rinse_repeat:
365    int64_t nowUs = ALooper::GetNowUs();
366
367    if (mLastPlaylistFetchTimeUs < 0
368            || (ssize_t)bandwidthIndex != mPrevBandwidthIndex
369            || (!mPlaylist->isComplete()
370                && mLastPlaylistFetchTimeUs + kMaxPlaylistAgeUs <= nowUs)) {
371        AString url;
372        if (mBandwidthItems.size() > 0) {
373            url = mBandwidthItems.editItemAt(bandwidthIndex).mURI;
374        } else {
375            url = mMasterURL;
376        }
377
378        bool firstTime = (mPlaylist == NULL);
379
380        mPlaylist = fetchPlaylist(url.c_str());
381        if (mPlaylist == NULL) {
382            LOGE("failed to load playlist at url '%s'", url.c_str());
383            mDataSource->queueEOS(ERROR_IO);
384            return;
385        }
386
387        if (firstTime) {
388            Mutex::Autolock autoLock(mLock);
389
390            int32_t targetDuration;
391            if (!mPlaylist->isComplete()
392                    || !mPlaylist->meta()->findInt32(
393                    "target-duration", &targetDuration)) {
394                mDurationUs = -1;
395            } else {
396                mDurationUs = 1000000ll * targetDuration * mPlaylist->size();
397            }
398        }
399
400        mLastPlaylistFetchTimeUs = ALooper::GetNowUs();
401    }
402
403    int32_t firstSeqNumberInPlaylist;
404    if (mPlaylist->meta() == NULL || !mPlaylist->meta()->findInt32(
405                "media-sequence", &firstSeqNumberInPlaylist)) {
406        firstSeqNumberInPlaylist = 0;
407    }
408
409    bool explicitDiscontinuity = false;
410    bool bandwidthChanged = false;
411
412    if (mSeekTimeUs >= 0) {
413        int32_t targetDuration;
414        if (mPlaylist->isComplete() &&
415                mPlaylist->meta()->findInt32(
416                    "target-duration", &targetDuration)) {
417            int64_t seekTimeSecs = (mSeekTimeUs + 500000ll) / 1000000ll;
418            int64_t index = seekTimeSecs / targetDuration;
419
420            if (index >= 0 && index < mPlaylist->size()) {
421                int32_t newSeqNumber = firstSeqNumberInPlaylist + index;
422
423                if (newSeqNumber != mSeqNumber) {
424                    LOGI("seeking to seq no %d", newSeqNumber);
425
426                    mSeqNumber = newSeqNumber;
427
428                    mDataSource->reset();
429
430                    // reseting the data source will have had the
431                    // side effect of discarding any previously queued
432                    // bandwidth change discontinuity.
433                    // Therefore we'll need to treat these explicit
434                    // discontinuities as involving a bandwidth change
435                    // even if they aren't directly.
436                    explicitDiscontinuity = true;
437                    bandwidthChanged = true;
438                }
439            }
440        }
441
442        mSeekTimeUs = -1;
443
444        Mutex::Autolock autoLock(mLock);
445        mSeekDone = true;
446        mCondition.broadcast();
447    }
448
449    if (mSeqNumber < 0) {
450        if (mPlaylist->isComplete()) {
451            mSeqNumber = firstSeqNumberInPlaylist;
452        } else {
453            mSeqNumber = firstSeqNumberInPlaylist + mPlaylist->size() / 2;
454        }
455    }
456
457    int32_t lastSeqNumberInPlaylist =
458        firstSeqNumberInPlaylist + (int32_t)mPlaylist->size() - 1;
459
460    if (mSeqNumber < firstSeqNumberInPlaylist
461            || mSeqNumber > lastSeqNumberInPlaylist) {
462        if (mPrevBandwidthIndex != (ssize_t)bandwidthIndex) {
463            // Go back to the previous bandwidth.
464
465            LOGI("new bandwidth does not have the sequence number "
466                 "we're looking for, switching back to previous bandwidth");
467
468            mLastPlaylistFetchTimeUs = -1;
469            bandwidthIndex = mPrevBandwidthIndex;
470            goto rinse_repeat;
471        }
472
473        if (!mPlaylist->isComplete()
474                && mSeqNumber > lastSeqNumberInPlaylist
475                && mNumRetries < kMaxNumRetries) {
476            ++mNumRetries;
477
478            mLastPlaylistFetchTimeUs = -1;
479            postMonitorQueue(3000000ll);
480            return;
481        }
482
483        LOGE("Cannot find sequence number %d in playlist "
484             "(contains %d - %d)",
485             mSeqNumber, firstSeqNumberInPlaylist,
486             firstSeqNumberInPlaylist + mPlaylist->size() - 1);
487
488        mDataSource->queueEOS(ERROR_END_OF_STREAM);
489        return;
490    }
491
492    mNumRetries = 0;
493
494    AString uri;
495    sp<AMessage> itemMeta;
496    CHECK(mPlaylist->itemAt(
497                mSeqNumber - firstSeqNumberInPlaylist,
498                &uri,
499                &itemMeta));
500
501    int32_t val;
502    if (itemMeta->findInt32("discontinuity", &val) && val != 0) {
503        explicitDiscontinuity = true;
504    }
505
506    sp<ABuffer> buffer;
507    status_t err = fetchFile(uri.c_str(), &buffer);
508    if (err != OK) {
509        LOGE("failed to fetch .ts segment at url '%s'", uri.c_str());
510        mDataSource->queueEOS(err);
511        return;
512    }
513
514    CHECK(buffer != NULL);
515
516    err = decryptBuffer(mSeqNumber - firstSeqNumberInPlaylist, buffer);
517
518    if (err != OK) {
519        LOGE("decryptBuffer failed w/ error %d", err);
520
521        mDataSource->queueEOS(err);
522        return;
523    }
524
525    if (buffer->size() == 0 || buffer->data()[0] != 0x47) {
526        // Not a transport stream???
527
528        LOGE("This doesn't look like a transport stream...");
529
530        mBandwidthItems.removeAt(bandwidthIndex);
531
532        if (mBandwidthItems.isEmpty()) {
533            mDataSource->queueEOS(ERROR_UNSUPPORTED);
534            return;
535        }
536
537        LOGI("Retrying with a different bandwidth stream.");
538
539        mLastPlaylistFetchTimeUs = -1;
540        bandwidthIndex = getBandwidthIndex();
541        mPrevBandwidthIndex = bandwidthIndex;
542        mSeqNumber = -1;
543
544        goto rinse_repeat;
545    }
546
547    if ((size_t)mPrevBandwidthIndex != bandwidthIndex) {
548        bandwidthChanged = true;
549    }
550
551    if (mPrevBandwidthIndex < 0) {
552        // Don't signal a bandwidth change at the very beginning of
553        // playback.
554        bandwidthChanged = false;
555    }
556
557    if (explicitDiscontinuity || bandwidthChanged) {
558        // Signal discontinuity.
559
560        LOGI("queueing discontinuity (explicit=%d, bandwidthChanged=%d)",
561             explicitDiscontinuity, bandwidthChanged);
562
563        sp<ABuffer> tmp = new ABuffer(188);
564        memset(tmp->data(), 0, tmp->size());
565        tmp->data()[1] = bandwidthChanged;
566
567        mDataSource->queueBuffer(tmp);
568    }
569
570    mDataSource->queueBuffer(buffer);
571
572    mPrevBandwidthIndex = bandwidthIndex;
573    ++mSeqNumber;
574
575    postMonitorQueue();
576}
577
578void LiveSession::onMonitorQueue() {
579    if (mSeekTimeUs >= 0
580            || mDataSource->countQueuedBuffers() < kMaxNumQueuedFragments) {
581        onDownloadNext();
582    } else {
583        postMonitorQueue(1000000ll);
584    }
585}
586
587status_t LiveSession::decryptBuffer(
588        size_t playlistIndex, const sp<ABuffer> &buffer) {
589    sp<AMessage> itemMeta;
590    bool found = false;
591    AString method;
592
593    for (ssize_t i = playlistIndex; i >= 0; --i) {
594        AString uri;
595        CHECK(mPlaylist->itemAt(i, &uri, &itemMeta));
596
597        if (itemMeta->findString("cipher-method", &method)) {
598            found = true;
599            break;
600        }
601    }
602
603    if (!found) {
604        method = "NONE";
605    }
606
607    if (method == "NONE") {
608        return OK;
609    } else if (!(method == "AES-128")) {
610        LOGE("Unsupported cipher method '%s'", method.c_str());
611        return ERROR_UNSUPPORTED;
612    }
613
614    AString keyURI;
615    if (!itemMeta->findString("cipher-uri", &keyURI)) {
616        LOGE("Missing key uri");
617        return ERROR_MALFORMED;
618    }
619
620    ssize_t index = mAESKeyForURI.indexOfKey(keyURI);
621
622    sp<ABuffer> key;
623    if (index >= 0) {
624        key = mAESKeyForURI.valueAt(index);
625    } else {
626        key = new ABuffer(16);
627
628        sp<NuHTTPDataSource> keySource = new NuHTTPDataSource;
629        status_t err = keySource->connect(keyURI.c_str());
630
631        if (err == OK) {
632            size_t offset = 0;
633            while (offset < 16) {
634                ssize_t n = keySource->readAt(
635                        offset, key->data() + offset, 16 - offset);
636                if (n <= 0) {
637                    err = ERROR_IO;
638                    break;
639                }
640
641                offset += n;
642            }
643        }
644
645        if (err != OK) {
646            LOGE("failed to fetch cipher key from '%s'.", keyURI.c_str());
647            return ERROR_IO;
648        }
649
650        mAESKeyForURI.add(keyURI, key);
651    }
652
653    AES_KEY aes_key;
654    if (AES_set_decrypt_key(key->data(), 128, &aes_key) != 0) {
655        LOGE("failed to set AES decryption key.");
656        return UNKNOWN_ERROR;
657    }
658
659    unsigned char aes_ivec[16];
660
661    AString iv;
662    if (itemMeta->findString("cipher-iv", &iv)) {
663        if ((!iv.startsWith("0x") && !iv.startsWith("0X"))
664                || iv.size() != 16 * 2 + 2) {
665            LOGE("malformed cipher IV '%s'.", iv.c_str());
666            return ERROR_MALFORMED;
667        }
668
669        memset(aes_ivec, 0, sizeof(aes_ivec));
670        for (size_t i = 0; i < 16; ++i) {
671            char c1 = tolower(iv.c_str()[2 + 2 * i]);
672            char c2 = tolower(iv.c_str()[3 + 2 * i]);
673            if (!isxdigit(c1) || !isxdigit(c2)) {
674                LOGE("malformed cipher IV '%s'.", iv.c_str());
675                return ERROR_MALFORMED;
676            }
677            uint8_t nibble1 = isdigit(c1) ? c1 - '0' : c1 - 'a' + 10;
678            uint8_t nibble2 = isdigit(c2) ? c2 - '0' : c2 - 'a' + 10;
679
680            aes_ivec[i] = nibble1 << 4 | nibble2;
681        }
682    } else {
683        memset(aes_ivec, 0, sizeof(aes_ivec));
684        aes_ivec[15] = mSeqNumber & 0xff;
685        aes_ivec[14] = (mSeqNumber >> 8) & 0xff;
686        aes_ivec[13] = (mSeqNumber >> 16) & 0xff;
687        aes_ivec[12] = (mSeqNumber >> 24) & 0xff;
688    }
689
690    AES_cbc_encrypt(
691            buffer->data(), buffer->data(), buffer->size(),
692            &aes_key, aes_ivec, AES_DECRYPT);
693
694    // hexdump(buffer->data(), buffer->size());
695
696    size_t n = buffer->size();
697    CHECK_GT(n, 0u);
698
699    size_t pad = buffer->data()[n - 1];
700
701    CHECK_GT(pad, 0u);
702    CHECK_LE(pad, 16u);
703    CHECK_GE((size_t)n, pad);
704    for (size_t i = 0; i < pad; ++i) {
705        CHECK_EQ((unsigned)buffer->data()[n - 1 - i], pad);
706    }
707
708    n -= pad;
709
710    buffer->setRange(buffer->offset(), n);
711
712    return OK;
713}
714
715void LiveSession::postMonitorQueue(int64_t delayUs) {
716    sp<AMessage> msg = new AMessage(kWhatMonitorQueue, id());
717    msg->setInt32("generation", ++mMonitorQueueGeneration);
718    msg->post(delayUs);
719}
720
721void LiveSession::onSeek(const sp<AMessage> &msg) {
722    int64_t timeUs;
723    CHECK(msg->findInt64("timeUs", &timeUs));
724
725    mSeekTimeUs = timeUs;
726    postMonitorQueue();
727}
728
729status_t LiveSession::getDuration(int64_t *durationUs) {
730    Mutex::Autolock autoLock(mLock);
731    *durationUs = mDurationUs;
732
733    return OK;
734}
735
736bool LiveSession::isSeekable() {
737    int64_t durationUs;
738    return getDuration(&durationUs) == OK && durationUs >= 0;
739}
740
741}  // namespace android
742
743