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 "MPEG2TSExtractor"
19
20#include <inttypes.h>
21#include <utils/Log.h>
22
23#include "include/MPEG2TSExtractor.h"
24#include "include/NuCachedSource2.h"
25
26#include <media/stagefright/foundation/ABuffer.h>
27#include <media/stagefright/foundation/ADebug.h>
28#include <media/stagefright/foundation/ALooper.h>
29#include <media/stagefright/DataSource.h>
30#include <media/stagefright/MediaDefs.h>
31#include <media/stagefright/MediaErrors.h>
32#include <media/stagefright/MediaSource.h>
33#include <media/stagefright/MetaData.h>
34#include <media/IStreamSource.h>
35#include <utils/String8.h>
36
37#include "AnotherPacketSource.h"
38#include "ATSParser.h"
39
40namespace android {
41
42static const size_t kTSPacketSize = 188;
43
44struct MPEG2TSSource : public MediaSource {
45    MPEG2TSSource(
46            const sp<MPEG2TSExtractor> &extractor,
47            const sp<AnotherPacketSource> &impl,
48            bool doesSeek);
49
50    virtual status_t start(MetaData *params = NULL);
51    virtual status_t stop();
52    virtual sp<MetaData> getFormat();
53
54    virtual status_t read(
55            MediaBuffer **buffer, const ReadOptions *options = NULL);
56
57private:
58    sp<MPEG2TSExtractor> mExtractor;
59    sp<AnotherPacketSource> mImpl;
60
61    // If there are both audio and video streams, only the video stream
62    // will signal seek on the extractor; otherwise the single stream will seek.
63    bool mDoesSeek;
64
65    DISALLOW_EVIL_CONSTRUCTORS(MPEG2TSSource);
66};
67
68MPEG2TSSource::MPEG2TSSource(
69        const sp<MPEG2TSExtractor> &extractor,
70        const sp<AnotherPacketSource> &impl,
71        bool doesSeek)
72    : mExtractor(extractor),
73      mImpl(impl),
74      mDoesSeek(doesSeek) {
75}
76
77status_t MPEG2TSSource::start(MetaData *params) {
78    return mImpl->start(params);
79}
80
81status_t MPEG2TSSource::stop() {
82    return mImpl->stop();
83}
84
85sp<MetaData> MPEG2TSSource::getFormat() {
86    return mImpl->getFormat();
87}
88
89status_t MPEG2TSSource::read(
90        MediaBuffer **out, const ReadOptions *options) {
91    *out = NULL;
92
93    int64_t seekTimeUs;
94    ReadOptions::SeekMode seekMode;
95    if (mDoesSeek && options && options->getSeekTo(&seekTimeUs, &seekMode)) {
96        // seek is needed
97        status_t err = mExtractor->seek(seekTimeUs, seekMode);
98        if (err != OK) {
99            return err;
100        }
101    }
102
103    if (mExtractor->feedUntilBufferAvailable(mImpl) != OK) {
104        return ERROR_END_OF_STREAM;
105    }
106
107    return mImpl->read(out, options);
108}
109
110////////////////////////////////////////////////////////////////////////////////
111
112MPEG2TSExtractor::MPEG2TSExtractor(const sp<DataSource> &source)
113    : mDataSource(source),
114      mParser(new ATSParser),
115      mLastSyncEvent(0),
116      mOffset(0) {
117    init();
118}
119
120size_t MPEG2TSExtractor::countTracks() {
121    return mSourceImpls.size();
122}
123
124sp<IMediaSource> MPEG2TSExtractor::getTrack(size_t index) {
125    if (index >= mSourceImpls.size()) {
126        return NULL;
127    }
128
129    // The seek reference track (video if present; audio otherwise) performs
130    // seek requests, while other tracks ignore requests.
131    return new MPEG2TSSource(this, mSourceImpls.editItemAt(index),
132            (mSeekSyncPoints == &mSyncPoints.editItemAt(index)));
133}
134
135sp<MetaData> MPEG2TSExtractor::getTrackMetaData(
136        size_t index, uint32_t /* flags */) {
137    return index < mSourceImpls.size()
138        ? mSourceImpls.editItemAt(index)->getFormat() : NULL;
139}
140
141sp<MetaData> MPEG2TSExtractor::getMetaData() {
142    sp<MetaData> meta = new MetaData;
143    meta->setCString(kKeyMIMEType, MEDIA_MIMETYPE_CONTAINER_MPEG2TS);
144
145    return meta;
146}
147
148void MPEG2TSExtractor::init() {
149    bool haveAudio = false;
150    bool haveVideo = false;
151    int64_t startTime = ALooper::GetNowUs();
152
153    while (feedMore(true /* isInit */) == OK) {
154        if (haveAudio && haveVideo) {
155            addSyncPoint_l(mLastSyncEvent);
156            mLastSyncEvent.reset();
157            break;
158        }
159        if (!haveVideo) {
160            sp<AnotherPacketSource> impl =
161                (AnotherPacketSource *)mParser->getSource(
162                        ATSParser::VIDEO).get();
163
164            if (impl != NULL) {
165                haveVideo = true;
166                mSourceImpls.push(impl);
167                mSyncPoints.push();
168                mSeekSyncPoints = &mSyncPoints.editTop();
169            }
170        }
171
172        if (!haveAudio) {
173            sp<AnotherPacketSource> impl =
174                (AnotherPacketSource *)mParser->getSource(
175                        ATSParser::AUDIO).get();
176
177            if (impl != NULL) {
178                haveAudio = true;
179                mSourceImpls.push(impl);
180                mSyncPoints.push();
181                if (!haveVideo) {
182                    mSeekSyncPoints = &mSyncPoints.editTop();
183                }
184            }
185        }
186
187        addSyncPoint_l(mLastSyncEvent);
188        mLastSyncEvent.reset();
189
190        // Wait only for 2 seconds to detect audio/video streams.
191        if (ALooper::GetNowUs() - startTime > 2000000ll) {
192            break;
193        }
194    }
195
196    off64_t size;
197    if (mDataSource->getSize(&size) == OK && (haveAudio || haveVideo)) {
198        sp<AnotherPacketSource> impl = haveVideo
199                ? (AnotherPacketSource *)mParser->getSource(
200                        ATSParser::VIDEO).get()
201                : (AnotherPacketSource *)mParser->getSource(
202                        ATSParser::AUDIO).get();
203        size_t prevSyncSize = 1;
204        int64_t durationUs = -1;
205        List<int64_t> durations;
206        // Estimate duration --- stabilize until you get <500ms deviation.
207        while (feedMore() == OK
208                && ALooper::GetNowUs() - startTime <= 2000000ll) {
209            if (mSeekSyncPoints->size() > prevSyncSize) {
210                prevSyncSize = mSeekSyncPoints->size();
211                int64_t diffUs = mSeekSyncPoints->keyAt(prevSyncSize - 1)
212                        - mSeekSyncPoints->keyAt(0);
213                off64_t diffOffset = mSeekSyncPoints->valueAt(prevSyncSize - 1)
214                        - mSeekSyncPoints->valueAt(0);
215                durationUs = size * diffUs / diffOffset;
216                durations.push_back(durationUs);
217                if (durations.size() > 5) {
218                    durations.erase(durations.begin());
219                    int64_t min = *durations.begin();
220                    int64_t max = *durations.begin();
221                    for (List<int64_t>::iterator i = durations.begin();
222                            i != durations.end();
223                            ++i) {
224                        if (min > *i) {
225                            min = *i;
226                        }
227                        if (max < *i) {
228                            max = *i;
229                        }
230                    }
231                    if (max - min < 500 * 1000) {
232                        break;
233                    }
234                }
235            }
236        }
237        status_t err;
238        int64_t bufferedDurationUs;
239        bufferedDurationUs = impl->getBufferedDurationUs(&err);
240        if (err == ERROR_END_OF_STREAM) {
241            durationUs = bufferedDurationUs;
242        }
243        if (durationUs > 0) {
244            const sp<MetaData> meta = impl->getFormat();
245            meta->setInt64(kKeyDuration, durationUs);
246            impl->setFormat(meta);
247        }
248    }
249
250    ALOGI("haveAudio=%d, haveVideo=%d, elaspedTime=%" PRId64,
251            haveAudio, haveVideo, ALooper::GetNowUs() - startTime);
252}
253
254status_t MPEG2TSExtractor::feedMore(bool isInit) {
255    Mutex::Autolock autoLock(mLock);
256
257    uint8_t packet[kTSPacketSize];
258    ssize_t n = mDataSource->readAt(mOffset, packet, kTSPacketSize);
259
260    if (n < (ssize_t)kTSPacketSize) {
261        if (n >= 0) {
262            mParser->signalEOS(ERROR_END_OF_STREAM);
263        }
264        return (n < 0) ? (status_t)n : ERROR_END_OF_STREAM;
265    }
266
267    ATSParser::SyncEvent event(mOffset);
268    mOffset += n;
269    status_t err = mParser->feedTSPacket(packet, kTSPacketSize, &event);
270    if (event.hasReturnedData()) {
271        if (isInit) {
272            mLastSyncEvent = event;
273        } else {
274            addSyncPoint_l(event);
275        }
276    }
277    return err;
278}
279
280void MPEG2TSExtractor::addSyncPoint_l(const ATSParser::SyncEvent &event) {
281    if (!event.hasReturnedData()) {
282        return;
283    }
284
285    for (size_t i = 0; i < mSourceImpls.size(); ++i) {
286        if (mSourceImpls[i].get() == event.getMediaSource().get()) {
287            KeyedVector<int64_t, off64_t> *syncPoints = &mSyncPoints.editItemAt(i);
288            syncPoints->add(event.getTimeUs(), event.getOffset());
289            // We're keeping the size of the sync points at most 5mb per a track.
290            size_t size = syncPoints->size();
291            if (size >= 327680) {
292                int64_t firstTimeUs = syncPoints->keyAt(0);
293                int64_t lastTimeUs = syncPoints->keyAt(size - 1);
294                if (event.getTimeUs() - firstTimeUs > lastTimeUs - event.getTimeUs()) {
295                    syncPoints->removeItemsAt(0, 4096);
296                } else {
297                    syncPoints->removeItemsAt(size - 4096, 4096);
298                }
299            }
300            break;
301        }
302    }
303}
304
305uint32_t MPEG2TSExtractor::flags() const {
306    return CAN_PAUSE | CAN_SEEK_BACKWARD | CAN_SEEK_FORWARD;
307}
308
309status_t MPEG2TSExtractor::seek(int64_t seekTimeUs,
310        const MediaSource::ReadOptions::SeekMode &seekMode) {
311    if (mSeekSyncPoints == NULL || mSeekSyncPoints->isEmpty()) {
312        ALOGW("No sync point to seek to.");
313        // ... and therefore we have nothing useful to do here.
314        return OK;
315    }
316
317    // Determine whether we're seeking beyond the known area.
318    bool shouldSeekBeyond =
319            (seekTimeUs > mSeekSyncPoints->keyAt(mSeekSyncPoints->size() - 1));
320
321    // Determine the sync point to seek.
322    size_t index = 0;
323    for (; index < mSeekSyncPoints->size(); ++index) {
324        int64_t timeUs = mSeekSyncPoints->keyAt(index);
325        if (timeUs > seekTimeUs) {
326            break;
327        }
328    }
329
330    switch (seekMode) {
331        case MediaSource::ReadOptions::SEEK_NEXT_SYNC:
332            if (index == mSeekSyncPoints->size()) {
333                ALOGW("Next sync not found; starting from the latest sync.");
334                --index;
335            }
336            break;
337        case MediaSource::ReadOptions::SEEK_CLOSEST_SYNC:
338        case MediaSource::ReadOptions::SEEK_CLOSEST:
339            ALOGW("seekMode not supported: %d; falling back to PREVIOUS_SYNC",
340                    seekMode);
341            // fall-through
342        case MediaSource::ReadOptions::SEEK_PREVIOUS_SYNC:
343            if (index == 0) {
344                ALOGW("Previous sync not found; starting from the earliest "
345                        "sync.");
346            } else {
347                --index;
348            }
349            break;
350    }
351    if (!shouldSeekBeyond || mOffset <= mSeekSyncPoints->valueAt(index)) {
352        int64_t actualSeekTimeUs = mSeekSyncPoints->keyAt(index);
353        mOffset = mSeekSyncPoints->valueAt(index);
354        status_t err = queueDiscontinuityForSeek(actualSeekTimeUs);
355        if (err != OK) {
356            return err;
357        }
358    }
359
360    if (shouldSeekBeyond) {
361        status_t err = seekBeyond(seekTimeUs);
362        if (err != OK) {
363            return err;
364        }
365    }
366
367    // Fast-forward to sync frame.
368    for (size_t i = 0; i < mSourceImpls.size(); ++i) {
369        const sp<AnotherPacketSource> &impl = mSourceImpls[i];
370        status_t err;
371        feedUntilBufferAvailable(impl);
372        while (impl->hasBufferAvailable(&err)) {
373            sp<AMessage> meta = impl->getMetaAfterLastDequeued(0);
374            sp<ABuffer> buffer;
375            if (meta == NULL) {
376                return UNKNOWN_ERROR;
377            }
378            int32_t sync;
379            if (meta->findInt32("isSync", &sync) && sync) {
380                break;
381            }
382            err = impl->dequeueAccessUnit(&buffer);
383            if (err != OK) {
384                return err;
385            }
386            feedUntilBufferAvailable(impl);
387        }
388    }
389
390    return OK;
391}
392
393status_t MPEG2TSExtractor::queueDiscontinuityForSeek(int64_t actualSeekTimeUs) {
394    // Signal discontinuity
395    sp<AMessage> extra(new AMessage);
396    extra->setInt64(IStreamListener::kKeyMediaTimeUs, actualSeekTimeUs);
397    mParser->signalDiscontinuity(ATSParser::DISCONTINUITY_TIME, extra);
398
399    // After discontinuity, impl should only have discontinuities
400    // with the last being what we queued. Dequeue them all here.
401    for (size_t i = 0; i < mSourceImpls.size(); ++i) {
402        const sp<AnotherPacketSource> &impl = mSourceImpls.itemAt(i);
403        sp<ABuffer> buffer;
404        status_t err;
405        while (impl->hasBufferAvailable(&err)) {
406            if (err != OK) {
407                return err;
408            }
409            err = impl->dequeueAccessUnit(&buffer);
410            // If the source contains anything but discontinuity, that's
411            // a programming mistake.
412            CHECK(err == INFO_DISCONTINUITY);
413        }
414    }
415
416    // Feed until we have a buffer for each source.
417    for (size_t i = 0; i < mSourceImpls.size(); ++i) {
418        const sp<AnotherPacketSource> &impl = mSourceImpls.itemAt(i);
419        sp<ABuffer> buffer;
420        status_t err = feedUntilBufferAvailable(impl);
421        if (err != OK) {
422            return err;
423        }
424    }
425
426    return OK;
427}
428
429status_t MPEG2TSExtractor::seekBeyond(int64_t seekTimeUs) {
430    // If we're seeking beyond where we know --- read until we reach there.
431    size_t syncPointsSize = mSeekSyncPoints->size();
432
433    while (seekTimeUs > mSeekSyncPoints->keyAt(
434            mSeekSyncPoints->size() - 1)) {
435        status_t err;
436        if (syncPointsSize < mSeekSyncPoints->size()) {
437            syncPointsSize = mSeekSyncPoints->size();
438            int64_t syncTimeUs = mSeekSyncPoints->keyAt(syncPointsSize - 1);
439            // Dequeue buffers before sync point in order to avoid too much
440            // cache building up.
441            sp<ABuffer> buffer;
442            for (size_t i = 0; i < mSourceImpls.size(); ++i) {
443                const sp<AnotherPacketSource> &impl = mSourceImpls[i];
444                int64_t timeUs;
445                while ((err = impl->nextBufferTime(&timeUs)) == OK) {
446                    if (timeUs < syncTimeUs) {
447                        impl->dequeueAccessUnit(&buffer);
448                    } else {
449                        break;
450                    }
451                }
452                if (err != OK && err != -EWOULDBLOCK) {
453                    return err;
454                }
455            }
456        }
457        if (feedMore() != OK) {
458            return ERROR_END_OF_STREAM;
459        }
460    }
461
462    return OK;
463}
464
465status_t MPEG2TSExtractor::feedUntilBufferAvailable(
466        const sp<AnotherPacketSource> &impl) {
467    status_t finalResult;
468    while (!impl->hasBufferAvailable(&finalResult)) {
469        if (finalResult != OK) {
470            return finalResult;
471        }
472
473        status_t err = feedMore();
474        if (err != OK) {
475            impl->signalEOS(err);
476        }
477    }
478    return OK;
479}
480
481////////////////////////////////////////////////////////////////////////////////
482
483bool SniffMPEG2TS(
484        const sp<DataSource> &source, String8 *mimeType, float *confidence,
485        sp<AMessage> *) {
486    for (int i = 0; i < 5; ++i) {
487        char header;
488        if (source->readAt(kTSPacketSize * i, &header, 1) != 1
489                || header != 0x47) {
490            return false;
491        }
492    }
493
494    *confidence = 0.1f;
495    mimeType->setTo(MEDIA_MIMETYPE_CONTAINER_MPEG2TS);
496
497    return true;
498}
499
500}  // namespace android
501