RTSPSource.cpp revision 4ad74b2b4507c8b2714b7b7584578d2d43ae2633 
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 "RTSPSource"
19#include <utils/Log.h>
20
21#include "RTSPSource.h"
22
23#include "AnotherPacketSource.h"
24#include "MyHandler.h"
25#include "SDPLoader.h"
26
27#include <media/IMediaHTTPService.h>
28#include <media/stagefright/MediaDefs.h>
29#include <media/stagefright/MetaData.h>
30
31namespace android {
32
33const int64_t kNearEOSTimeoutUs = 2000000ll; // 2 secs
34
35NuPlayer::RTSPSource::RTSPSource(
36        const sp<AMessage> &notify,
37        const sp<IMediaHTTPService> &httpService,
38        const char *url,
39        const KeyedVector<String8, String8> *headers,
40        bool uidValid,
41        uid_t uid,
42        bool isSDP)
43    : Source(notify),
44      mHTTPService(httpService),
45      mURL(url),
46      mUIDValid(uidValid),
47      mUID(uid),
48      mFlags(0),
49      mIsSDP(isSDP),
50      mState(DISCONNECTED),
51      mFinalResult(OK),
52      mDisconnectReplyID(0),
53      mBuffering(false),
54      mSeekGeneration(0),
55      mEOSTimeoutAudio(0),
56      mEOSTimeoutVideo(0) {
57    if (headers) {
58        mExtraHeaders = *headers;
59
60        ssize_t index =
61            mExtraHeaders.indexOfKey(String8("x-hide-urls-from-log"));
62
63        if (index >= 0) {
64            mFlags |= kFlagIncognito;
65
66            mExtraHeaders.removeItemsAt(index);
67        }
68    }
69}
70
71NuPlayer::RTSPSource::~RTSPSource() {
72    if (mLooper != NULL) {
73        mLooper->unregisterHandler(id());
74        mLooper->stop();
75    }
76}
77
78void NuPlayer::RTSPSource::prepareAsync() {
79    if (mIsSDP && mHTTPService == NULL) {
80        notifyPrepared(BAD_VALUE);
81        return;
82    }
83
84    if (mLooper == NULL) {
85        mLooper = new ALooper;
86        mLooper->setName("rtsp");
87        mLooper->start();
88
89        mLooper->registerHandler(this);
90    }
91
92    CHECK(mHandler == NULL);
93    CHECK(mSDPLoader == NULL);
94
95    sp<AMessage> notify = new AMessage(kWhatNotify, this);
96
97    CHECK_EQ(mState, (int)DISCONNECTED);
98    mState = CONNECTING;
99
100    if (mIsSDP) {
101        mSDPLoader = new SDPLoader(notify,
102                (mFlags & kFlagIncognito) ? SDPLoader::kFlagIncognito : 0,
103                mHTTPService);
104
105        mSDPLoader->load(
106                mURL.c_str(), mExtraHeaders.isEmpty() ? NULL : &mExtraHeaders);
107    } else {
108        mHandler = new MyHandler(mURL.c_str(), notify, mUIDValid, mUID);
109        mLooper->registerHandler(mHandler);
110
111        mHandler->connect();
112    }
113
114    startBufferingIfNecessary();
115}
116
117void NuPlayer::RTSPSource::start() {
118}
119
120void NuPlayer::RTSPSource::stop() {
121    if (mLooper == NULL) {
122        return;
123    }
124    sp<AMessage> msg = new AMessage(kWhatDisconnect, this);
125
126    sp<AMessage> dummy;
127    msg->postAndAwaitResponse(&dummy);
128}
129
130void NuPlayer::RTSPSource::pause() {
131    int64_t mediaDurationUs = 0;
132    getDuration(&mediaDurationUs);
133    for (size_t index = 0; index < mTracks.size(); index++) {
134        TrackInfo *info = &mTracks.editItemAt(index);
135        sp<AnotherPacketSource> source = info->mSource;
136
137        // Check if EOS or ERROR is received
138        if (source != NULL && source->isFinished(mediaDurationUs)) {
139            return;
140        }
141    }
142    if (mHandler != NULL) {
143        mHandler->pause();
144    }
145}
146
147void NuPlayer::RTSPSource::resume() {
148    if (mHandler != NULL) {
149        mHandler->resume();
150    }
151}
152
153status_t NuPlayer::RTSPSource::feedMoreTSData() {
154    Mutex::Autolock _l(mBufferingLock);
155    return mFinalResult;
156}
157
158sp<MetaData> NuPlayer::RTSPSource::getFormatMeta(bool audio) {
159    sp<AnotherPacketSource> source = getSource(audio);
160
161    if (source == NULL) {
162        return NULL;
163    }
164
165    return source->getFormat();
166}
167
168bool NuPlayer::RTSPSource::haveSufficientDataOnAllTracks() {
169    // We're going to buffer at least 2 secs worth data on all tracks before
170    // starting playback (both at startup and after a seek).
171
172    static const int64_t kMinDurationUs = 2000000ll;
173
174    int64_t mediaDurationUs = 0;
175    getDuration(&mediaDurationUs);
176    if ((mAudioTrack != NULL && mAudioTrack->isFinished(mediaDurationUs))
177            || (mVideoTrack != NULL && mVideoTrack->isFinished(mediaDurationUs))) {
178        return true;
179    }
180
181    status_t err;
182    int64_t durationUs;
183    if (mAudioTrack != NULL
184            && (durationUs = mAudioTrack->getBufferedDurationUs(&err))
185                    < kMinDurationUs
186            && err == OK) {
187        ALOGV("audio track doesn't have enough data yet. (%.2f secs buffered)",
188              durationUs / 1E6);
189        return false;
190    }
191
192    if (mVideoTrack != NULL
193            && (durationUs = mVideoTrack->getBufferedDurationUs(&err))
194                    < kMinDurationUs
195            && err == OK) {
196        ALOGV("video track doesn't have enough data yet. (%.2f secs buffered)",
197              durationUs / 1E6);
198        return false;
199    }
200
201    return true;
202}
203
204status_t NuPlayer::RTSPSource::dequeueAccessUnit(
205        bool audio, sp<ABuffer> *accessUnit) {
206    if (!stopBufferingIfNecessary()) {
207        return -EWOULDBLOCK;
208    }
209
210    sp<AnotherPacketSource> source = getSource(audio);
211
212    if (source == NULL) {
213        return -EWOULDBLOCK;
214    }
215
216    status_t finalResult;
217    if (!source->hasBufferAvailable(&finalResult)) {
218        if (finalResult == OK) {
219            int64_t mediaDurationUs = 0;
220            getDuration(&mediaDurationUs);
221            sp<AnotherPacketSource> otherSource = getSource(!audio);
222            status_t otherFinalResult;
223
224            // If other source already signaled EOS, this source should also signal EOS
225            if (otherSource != NULL &&
226                    !otherSource->hasBufferAvailable(&otherFinalResult) &&
227                    otherFinalResult == ERROR_END_OF_STREAM) {
228                source->signalEOS(ERROR_END_OF_STREAM);
229                return ERROR_END_OF_STREAM;
230            }
231
232            // If this source has detected near end, give it some time to retrieve more
233            // data before signaling EOS
234            if (source->isFinished(mediaDurationUs)) {
235                int64_t eosTimeout = audio ? mEOSTimeoutAudio : mEOSTimeoutVideo;
236                if (eosTimeout == 0) {
237                    setEOSTimeout(audio, ALooper::GetNowUs());
238                } else if ((ALooper::GetNowUs() - eosTimeout) > kNearEOSTimeoutUs) {
239                    setEOSTimeout(audio, 0);
240                    source->signalEOS(ERROR_END_OF_STREAM);
241                    return ERROR_END_OF_STREAM;
242                }
243                return -EWOULDBLOCK;
244            }
245
246            if (!(otherSource != NULL && otherSource->isFinished(mediaDurationUs))) {
247                // We should not enter buffering mode
248                // if any of the sources already have detected EOS.
249                startBufferingIfNecessary();
250            }
251
252            return -EWOULDBLOCK;
253        }
254        return finalResult;
255    }
256
257    setEOSTimeout(audio, 0);
258
259    return source->dequeueAccessUnit(accessUnit);
260}
261
262sp<AnotherPacketSource> NuPlayer::RTSPSource::getSource(bool audio) {
263    if (mTSParser != NULL) {
264        sp<MediaSource> source = mTSParser->getSource(
265                audio ? ATSParser::AUDIO : ATSParser::VIDEO);
266
267        return static_cast<AnotherPacketSource *>(source.get());
268    }
269
270    return audio ? mAudioTrack : mVideoTrack;
271}
272
273void NuPlayer::RTSPSource::setEOSTimeout(bool audio, int64_t timeout) {
274    if (audio) {
275        mEOSTimeoutAudio = timeout;
276    } else {
277        mEOSTimeoutVideo = timeout;
278    }
279}
280
281status_t NuPlayer::RTSPSource::getDuration(int64_t *durationUs) {
282    *durationUs = 0ll;
283
284    int64_t audioDurationUs;
285    if (mAudioTrack != NULL
286            && mAudioTrack->getFormat()->findInt64(
287                kKeyDuration, &audioDurationUs)
288            && audioDurationUs > *durationUs) {
289        *durationUs = audioDurationUs;
290    }
291
292    int64_t videoDurationUs;
293    if (mVideoTrack != NULL
294            && mVideoTrack->getFormat()->findInt64(
295                kKeyDuration, &videoDurationUs)
296            && videoDurationUs > *durationUs) {
297        *durationUs = videoDurationUs;
298    }
299
300    return OK;
301}
302
303status_t NuPlayer::RTSPSource::seekTo(int64_t seekTimeUs) {
304    sp<AMessage> msg = new AMessage(kWhatPerformSeek, this);
305    msg->setInt32("generation", ++mSeekGeneration);
306    msg->setInt64("timeUs", seekTimeUs);
307
308    sp<AMessage> response;
309    status_t err = msg->postAndAwaitResponse(&response);
310    if (err == OK && response != NULL) {
311        CHECK(response->findInt32("err", &err));
312    }
313
314    return err;
315}
316
317void NuPlayer::RTSPSource::performSeek(int64_t seekTimeUs) {
318    if (mState != CONNECTED) {
319        finishSeek(INVALID_OPERATION);
320        return;
321    }
322
323    mState = SEEKING;
324    mHandler->seek(seekTimeUs);
325}
326
327void NuPlayer::RTSPSource::onMessageReceived(const sp<AMessage> &msg) {
328    if (msg->what() == kWhatDisconnect) {
329        sp<AReplyToken> replyID;
330        CHECK(msg->senderAwaitsResponse(&replyID));
331
332        mDisconnectReplyID = replyID;
333        finishDisconnectIfPossible();
334        return;
335    } else if (msg->what() == kWhatPerformSeek) {
336        int32_t generation;
337        CHECK(msg->findInt32("generation", &generation));
338        CHECK(msg->senderAwaitsResponse(&mSeekReplyID));
339
340        if (generation != mSeekGeneration) {
341            // obsolete.
342            finishSeek(OK);
343            return;
344        }
345
346        int64_t seekTimeUs;
347        CHECK(msg->findInt64("timeUs", &seekTimeUs));
348
349        performSeek(seekTimeUs);
350        return;
351    }
352
353    CHECK_EQ(msg->what(), (int)kWhatNotify);
354
355    int32_t what;
356    CHECK(msg->findInt32("what", &what));
357
358    switch (what) {
359        case MyHandler::kWhatConnected:
360        {
361            onConnected();
362
363            notifyVideoSizeChanged();
364
365            uint32_t flags = 0;
366
367            if (mHandler->isSeekable()) {
368                flags = FLAG_CAN_PAUSE
369                        | FLAG_CAN_SEEK
370                        | FLAG_CAN_SEEK_BACKWARD
371                        | FLAG_CAN_SEEK_FORWARD;
372            }
373
374            notifyFlagsChanged(flags);
375            notifyPrepared();
376            break;
377        }
378
379        case MyHandler::kWhatDisconnected:
380        {
381            onDisconnected(msg);
382            break;
383        }
384
385        case MyHandler::kWhatSeekDone:
386        {
387            mState = CONNECTED;
388            // Unblock seekTo here in case we attempted to seek in a live stream
389            finishSeek(OK);
390            break;
391        }
392
393        case MyHandler::kWhatSeekPaused:
394        {
395            sp<AnotherPacketSource> source = getSource(true /* audio */);
396            if (source != NULL) {
397                source->queueDiscontinuity(ATSParser::DISCONTINUITY_NONE,
398                        /* extra */ NULL,
399                        /* discard */ true);
400            }
401            source = getSource(false /* video */);
402            if (source != NULL) {
403                source->queueDiscontinuity(ATSParser::DISCONTINUITY_NONE,
404                        /* extra */ NULL,
405                        /* discard */ true);
406            };
407
408            status_t err = OK;
409            msg->findInt32("err", &err);
410
411            if (err == OK) {
412                int64_t timeUs;
413                CHECK(msg->findInt64("time", &timeUs));
414                mHandler->continueSeekAfterPause(timeUs);
415            } else {
416                finishSeek(err);
417            }
418            break;
419        }
420
421        case MyHandler::kWhatAccessUnit:
422        {
423            size_t trackIndex;
424            CHECK(msg->findSize("trackIndex", &trackIndex));
425
426            if (mTSParser == NULL) {
427                CHECK_LT(trackIndex, mTracks.size());
428            } else {
429                CHECK_EQ(trackIndex, 0u);
430            }
431
432            sp<ABuffer> accessUnit;
433            CHECK(msg->findBuffer("accessUnit", &accessUnit));
434
435            int32_t damaged;
436            if (accessUnit->meta()->findInt32("damaged", &damaged)
437                    && damaged) {
438                ALOGI("dropping damaged access unit.");
439                break;
440            }
441
442            if (mTSParser != NULL) {
443                size_t offset = 0;
444                status_t err = OK;
445                while (offset + 188 <= accessUnit->size()) {
446                    err = mTSParser->feedTSPacket(
447                            accessUnit->data() + offset, 188);
448                    if (err != OK) {
449                        break;
450                    }
451
452                    offset += 188;
453                }
454
455                if (offset < accessUnit->size()) {
456                    err = ERROR_MALFORMED;
457                }
458
459                if (err != OK) {
460                    sp<AnotherPacketSource> source = getSource(false /* audio */);
461                    if (source != NULL) {
462                        source->signalEOS(err);
463                    }
464
465                    source = getSource(true /* audio */);
466                    if (source != NULL) {
467                        source->signalEOS(err);
468                    }
469                }
470                break;
471            }
472
473            TrackInfo *info = &mTracks.editItemAt(trackIndex);
474
475            sp<AnotherPacketSource> source = info->mSource;
476            if (source != NULL) {
477                uint32_t rtpTime;
478                CHECK(accessUnit->meta()->findInt32("rtp-time", (int32_t *)&rtpTime));
479
480                if (!info->mNPTMappingValid) {
481                    // This is a live stream, we didn't receive any normal
482                    // playtime mapping. We won't map to npt time.
483                    source->queueAccessUnit(accessUnit);
484                    break;
485                }
486
487                int64_t nptUs =
488                    ((double)rtpTime - (double)info->mRTPTime)
489                        / info->mTimeScale
490                        * 1000000ll
491                        + info->mNormalPlaytimeUs;
492
493                accessUnit->meta()->setInt64("timeUs", nptUs);
494
495                source->queueAccessUnit(accessUnit);
496            }
497            break;
498        }
499
500        case MyHandler::kWhatEOS:
501        {
502            int32_t finalResult;
503            CHECK(msg->findInt32("finalResult", &finalResult));
504            CHECK_NE(finalResult, (status_t)OK);
505
506            if (mTSParser != NULL) {
507                sp<AnotherPacketSource> source = getSource(false /* audio */);
508                if (source != NULL) {
509                    source->signalEOS(finalResult);
510                }
511
512                source = getSource(true /* audio */);
513                if (source != NULL) {
514                    source->signalEOS(finalResult);
515                }
516
517                return;
518            }
519
520            size_t trackIndex;
521            CHECK(msg->findSize("trackIndex", &trackIndex));
522            CHECK_LT(trackIndex, mTracks.size());
523
524            TrackInfo *info = &mTracks.editItemAt(trackIndex);
525            sp<AnotherPacketSource> source = info->mSource;
526            if (source != NULL) {
527                source->signalEOS(finalResult);
528            }
529
530            break;
531        }
532
533        case MyHandler::kWhatSeekDiscontinuity:
534        {
535            size_t trackIndex;
536            CHECK(msg->findSize("trackIndex", &trackIndex));
537            CHECK_LT(trackIndex, mTracks.size());
538
539            TrackInfo *info = &mTracks.editItemAt(trackIndex);
540            sp<AnotherPacketSource> source = info->mSource;
541            if (source != NULL) {
542                source->queueDiscontinuity(
543                        ATSParser::DISCONTINUITY_TIME,
544                        NULL,
545                        true /* discard */);
546            }
547
548            break;
549        }
550
551        case MyHandler::kWhatNormalPlayTimeMapping:
552        {
553            size_t trackIndex;
554            CHECK(msg->findSize("trackIndex", &trackIndex));
555            CHECK_LT(trackIndex, mTracks.size());
556
557            uint32_t rtpTime;
558            CHECK(msg->findInt32("rtpTime", (int32_t *)&rtpTime));
559
560            int64_t nptUs;
561            CHECK(msg->findInt64("nptUs", &nptUs));
562
563            TrackInfo *info = &mTracks.editItemAt(trackIndex);
564            info->mRTPTime = rtpTime;
565            info->mNormalPlaytimeUs = nptUs;
566            info->mNPTMappingValid = true;
567            break;
568        }
569
570        case SDPLoader::kWhatSDPLoaded:
571        {
572            onSDPLoaded(msg);
573            break;
574        }
575
576        default:
577            TRESPASS();
578    }
579}
580
581void NuPlayer::RTSPSource::onConnected() {
582    CHECK(mAudioTrack == NULL);
583    CHECK(mVideoTrack == NULL);
584
585    size_t numTracks = mHandler->countTracks();
586    for (size_t i = 0; i < numTracks; ++i) {
587        int32_t timeScale;
588        sp<MetaData> format = mHandler->getTrackFormat(i, &timeScale);
589
590        const char *mime;
591        CHECK(format->findCString(kKeyMIMEType, &mime));
592
593        if (!strcasecmp(mime, MEDIA_MIMETYPE_CONTAINER_MPEG2TS)) {
594            // Very special case for MPEG2 Transport Streams.
595            CHECK_EQ(numTracks, 1u);
596
597            mTSParser = new ATSParser;
598            return;
599        }
600
601        bool isAudio = !strncasecmp(mime, "audio/", 6);
602        bool isVideo = !strncasecmp(mime, "video/", 6);
603
604        TrackInfo info;
605        info.mTimeScale = timeScale;
606        info.mRTPTime = 0;
607        info.mNormalPlaytimeUs = 0ll;
608        info.mNPTMappingValid = false;
609
610        if ((isAudio && mAudioTrack == NULL)
611                || (isVideo && mVideoTrack == NULL)) {
612            sp<AnotherPacketSource> source = new AnotherPacketSource(format);
613
614            if (isAudio) {
615                mAudioTrack = source;
616            } else {
617                mVideoTrack = source;
618            }
619
620            info.mSource = source;
621        }
622
623        mTracks.push(info);
624    }
625
626    mState = CONNECTED;
627}
628
629void NuPlayer::RTSPSource::onSDPLoaded(const sp<AMessage> &msg) {
630    status_t err;
631    CHECK(msg->findInt32("result", &err));
632
633    mSDPLoader.clear();
634
635    if (mDisconnectReplyID != 0) {
636        err = UNKNOWN_ERROR;
637    }
638
639    if (err == OK) {
640        sp<ASessionDescription> desc;
641        sp<RefBase> obj;
642        CHECK(msg->findObject("description", &obj));
643        desc = static_cast<ASessionDescription *>(obj.get());
644
645        AString rtspUri;
646        if (!desc->findAttribute(0, "a=control", &rtspUri)) {
647            ALOGE("Unable to find url in SDP");
648            err = UNKNOWN_ERROR;
649        } else {
650            sp<AMessage> notify = new AMessage(kWhatNotify, this);
651
652            mHandler = new MyHandler(rtspUri.c_str(), notify, mUIDValid, mUID);
653            mLooper->registerHandler(mHandler);
654
655            mHandler->loadSDP(desc);
656        }
657    }
658
659    if (err != OK) {
660        if (mState == CONNECTING) {
661            // We're still in the preparation phase, signal that it
662            // failed.
663            notifyPrepared(err);
664        }
665
666        mState = DISCONNECTED;
667        setError(err);
668
669        if (mDisconnectReplyID != 0) {
670            finishDisconnectIfPossible();
671        }
672    }
673}
674
675void NuPlayer::RTSPSource::onDisconnected(const sp<AMessage> &msg) {
676    if (mState == DISCONNECTED) {
677        return;
678    }
679
680    status_t err;
681    CHECK(msg->findInt32("result", &err));
682    CHECK_NE(err, (status_t)OK);
683
684    mLooper->unregisterHandler(mHandler->id());
685    mHandler.clear();
686
687    if (mState == CONNECTING) {
688        // We're still in the preparation phase, signal that it
689        // failed.
690        notifyPrepared(err);
691    }
692
693    mState = DISCONNECTED;
694    setError(err);
695
696    if (mDisconnectReplyID != 0) {
697        finishDisconnectIfPossible();
698    }
699}
700
701void NuPlayer::RTSPSource::finishDisconnectIfPossible() {
702    if (mState != DISCONNECTED) {
703        if (mHandler != NULL) {
704            mHandler->disconnect();
705        } else if (mSDPLoader != NULL) {
706            mSDPLoader->cancel();
707        }
708        return;
709    }
710
711    (new AMessage)->postReply(mDisconnectReplyID);
712    mDisconnectReplyID = 0;
713}
714
715void NuPlayer::RTSPSource::setError(status_t err) {
716    Mutex::Autolock _l(mBufferingLock);
717    mFinalResult = err;
718}
719
720void NuPlayer::RTSPSource::startBufferingIfNecessary() {
721    Mutex::Autolock _l(mBufferingLock);
722
723    if (!mBuffering) {
724        mBuffering = true;
725
726        sp<AMessage> notify = dupNotify();
727        notify->setInt32("what", kWhatBufferingStart);
728        notify->post();
729    }
730}
731
732bool NuPlayer::RTSPSource::stopBufferingIfNecessary() {
733    Mutex::Autolock _l(mBufferingLock);
734
735    if (mBuffering) {
736        if (!haveSufficientDataOnAllTracks()) {
737            return false;
738        }
739
740        mBuffering = false;
741
742        sp<AMessage> notify = dupNotify();
743        notify->setInt32("what", kWhatBufferingEnd);
744        notify->post();
745    }
746
747    return true;
748}
749
750void NuPlayer::RTSPSource::finishSeek(status_t err) {
751    if (mSeekReplyID == NULL) {
752        return;
753    }
754    sp<AMessage> seekReply = new AMessage;
755    seekReply->setInt32("err", err);
756    seekReply->postReply(mSeekReplyID);
757    mSeekReplyID = NULL;
758}
759
760}  // namespace android
761