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 "StreamingSource"
19 #include <utils/Log.h>
20
21 #include "StreamingSource.h"
22
23 #include "ATSParser.h"
24 #include "AnotherPacketSource.h"
25 #include "NuPlayerStreamListener.h"
26
27 #include <media/stagefright/MediaSource.h>
28 #include <media/stagefright/foundation/ABuffer.h>
29 #include <media/stagefright/foundation/ADebug.h>
30 #include <media/stagefright/foundation/AMessage.h>
31 #include <media/stagefright/foundation/MediaKeys.h>
32 #include <media/stagefright/MetaData.h>
33 #include <media/stagefright/Utils.h>
34
35 namespace android {
36
37 const int32_t kNumListenerQueuePackets = 80;
38
StreamingSource(const sp<AMessage> & notify,const sp<IStreamSource> & source)39 NuPlayer::StreamingSource::StreamingSource(
40 const sp<AMessage> ¬ify,
41 const sp<IStreamSource> &source)
42 : Source(notify),
43 mSource(source),
44 mFinalResult(OK),
45 mBuffering(false) {
46 }
47
~StreamingSource()48 NuPlayer::StreamingSource::~StreamingSource() {
49 if (mLooper != NULL) {
50 mLooper->unregisterHandler(id());
51 mLooper->stop();
52 }
53 }
54
getBufferingSettings(BufferingSettings * buffering)55 status_t NuPlayer::StreamingSource::getBufferingSettings(
56 BufferingSettings *buffering /* nonnull */) {
57 *buffering = BufferingSettings();
58 return OK;
59 }
60
setBufferingSettings(const BufferingSettings &)61 status_t NuPlayer::StreamingSource::setBufferingSettings(
62 const BufferingSettings & /* buffering */) {
63 return OK;
64 }
65
prepareAsync()66 void NuPlayer::StreamingSource::prepareAsync() {
67 if (mLooper == NULL) {
68 mLooper = new ALooper;
69 mLooper->setName("streaming");
70 mLooper->start();
71
72 mLooper->registerHandler(this);
73 }
74
75 notifyVideoSizeChanged();
76 notifyFlagsChanged(0);
77 notifyPrepared();
78 }
79
start()80 void NuPlayer::StreamingSource::start() {
81 mStreamListener = new NuPlayerStreamListener(mSource, NULL);
82 mSource->setListener(mStreamListener);
83
84 uint32_t sourceFlags = mSource->flags();
85
86 uint32_t parserFlags = ATSParser::TS_TIMESTAMPS_ARE_ABSOLUTE;
87 if (sourceFlags & IStreamSource::kFlagAlignedVideoData) {
88 parserFlags |= ATSParser::ALIGNED_VIDEO_DATA;
89 }
90
91 mTSParser = new ATSParser(parserFlags);
92
93 mStreamListener->start();
94
95 postReadBuffer();
96 }
97
feedMoreTSData()98 status_t NuPlayer::StreamingSource::feedMoreTSData() {
99 return postReadBuffer();
100 }
101
onReadBuffer()102 void NuPlayer::StreamingSource::onReadBuffer() {
103 for (int32_t i = 0; i < kNumListenerQueuePackets; ++i) {
104 char buffer[188];
105 sp<AMessage> extra;
106 ssize_t n = mStreamListener->read(buffer, sizeof(buffer), &extra);
107
108 if (n == 0) {
109 ALOGI("input data EOS reached.");
110 mTSParser->signalEOS(ERROR_END_OF_STREAM);
111 setError(ERROR_END_OF_STREAM);
112 break;
113 } else if (n == INFO_DISCONTINUITY) {
114 int32_t type = ATSParser::DISCONTINUITY_TIME;
115
116 int32_t mask;
117 if (extra != NULL
118 && extra->findInt32(
119 kIStreamListenerKeyDiscontinuityMask, &mask)) {
120 if (mask == 0) {
121 ALOGE("Client specified an illegal discontinuity type.");
122 setError(ERROR_UNSUPPORTED);
123 break;
124 }
125
126 type = mask;
127 }
128
129 mTSParser->signalDiscontinuity(
130 (ATSParser::DiscontinuityType)type, extra);
131 } else if (n < 0) {
132 break;
133 } else {
134 if (buffer[0] == 0x00) { // OK to access buffer[0] since n must be > 0 here
135 // XXX legacy
136
137 if (extra == NULL) {
138 extra = new AMessage;
139 }
140
141 uint8_t type = 0;
142 if (n > 1) {
143 type = buffer[1];
144
145 if ((type & 2) && (n >= 2 + sizeof(int64_t))) {
146 int64_t mediaTimeUs;
147 memcpy(&mediaTimeUs, &buffer[2], sizeof(mediaTimeUs));
148
149 extra->setInt64(kATSParserKeyMediaTimeUs, mediaTimeUs);
150 }
151 }
152
153 mTSParser->signalDiscontinuity(
154 ((type & 1) == 0)
155 ? ATSParser::DISCONTINUITY_TIME
156 : ATSParser::DISCONTINUITY_FORMATCHANGE,
157 extra);
158 } else {
159 status_t err = mTSParser->feedTSPacket(buffer, n);
160
161 if (err != OK) {
162 ALOGE("TS Parser returned error %d", err);
163
164 mTSParser->signalEOS(err);
165 setError(err);
166 break;
167 }
168 }
169 }
170 }
171 }
172
postReadBuffer()173 status_t NuPlayer::StreamingSource::postReadBuffer() {
174 {
175 Mutex::Autolock _l(mBufferingLock);
176 if (mFinalResult != OK) {
177 return mFinalResult;
178 }
179 if (mBuffering) {
180 return OK;
181 }
182 mBuffering = true;
183 }
184
185 (new AMessage(kWhatReadBuffer, this))->post();
186 return OK;
187 }
188
haveSufficientDataOnAllTracks()189 bool NuPlayer::StreamingSource::haveSufficientDataOnAllTracks() {
190 // We're going to buffer at least 2 secs worth data on all tracks before
191 // starting playback (both at startup and after a seek).
192
193 static const int64_t kMinDurationUs = 2000000LL;
194
195 sp<AnotherPacketSource> audioTrack = getSource(true /*audio*/);
196 sp<AnotherPacketSource> videoTrack = getSource(false /*audio*/);
197
198 status_t err;
199 int64_t durationUs;
200 if (audioTrack != NULL
201 && (durationUs = audioTrack->getBufferedDurationUs(&err))
202 < kMinDurationUs
203 && err == OK) {
204 ALOGV("audio track doesn't have enough data yet. (%.2f secs buffered)",
205 durationUs / 1E6);
206 return false;
207 }
208
209 if (videoTrack != NULL
210 && (durationUs = videoTrack->getBufferedDurationUs(&err))
211 < kMinDurationUs
212 && err == OK) {
213 ALOGV("video track doesn't have enough data yet. (%.2f secs buffered)",
214 durationUs / 1E6);
215 return false;
216 }
217
218 return true;
219 }
220
setError(status_t err)221 void NuPlayer::StreamingSource::setError(status_t err) {
222 Mutex::Autolock _l(mBufferingLock);
223 mFinalResult = err;
224 }
225
getSource(bool audio)226 sp<AnotherPacketSource> NuPlayer::StreamingSource::getSource(bool audio) {
227 if (mTSParser == NULL) {
228 return NULL;
229 }
230
231 sp<MediaSource> source = mTSParser->getSource(
232 audio ? ATSParser::AUDIO : ATSParser::VIDEO);
233
234 return static_cast<AnotherPacketSource *>(source.get());
235 }
236
getFormat(bool audio)237 sp<AMessage> NuPlayer::StreamingSource::getFormat(bool audio) {
238 sp<AnotherPacketSource> source = getSource(audio);
239
240 sp<AMessage> format = new AMessage;
241 if (source == NULL) {
242 format->setInt32("err", -EWOULDBLOCK);
243 return format;
244 }
245
246 sp<MetaData> meta = source->getFormat();
247 if (meta == NULL) {
248 format->setInt32("err", -EWOULDBLOCK);
249 return format;
250 }
251 status_t err = convertMetaDataToMessage(meta, &format);
252 if (err != OK) { // format may have been cleared on error
253 return NULL;
254 }
255 return format;
256 }
257
dequeueAccessUnit(bool audio,sp<ABuffer> * accessUnit)258 status_t NuPlayer::StreamingSource::dequeueAccessUnit(
259 bool audio, sp<ABuffer> *accessUnit) {
260 sp<AnotherPacketSource> source = getSource(audio);
261
262 if (source == NULL) {
263 return -EWOULDBLOCK;
264 }
265
266 if (!haveSufficientDataOnAllTracks()) {
267 postReadBuffer();
268 }
269
270 status_t finalResult;
271 if (!source->hasBufferAvailable(&finalResult)) {
272 return finalResult == OK ? -EWOULDBLOCK : finalResult;
273 }
274
275 status_t err = source->dequeueAccessUnit(accessUnit);
276
277 #if !defined(LOG_NDEBUG) || LOG_NDEBUG == 0
278 if (err == OK) {
279 int64_t timeUs;
280 CHECK((*accessUnit)->meta()->findInt64("timeUs", &timeUs));
281 ALOGV("dequeueAccessUnit timeUs=%lld us", timeUs);
282 }
283 #endif
284
285 return err;
286 }
287
isRealTime() const288 bool NuPlayer::StreamingSource::isRealTime() const {
289 return mSource->flags() & IStreamSource::kFlagIsRealTimeData;
290 }
291
onMessageReceived(const sp<AMessage> & msg)292 void NuPlayer::StreamingSource::onMessageReceived(
293 const sp<AMessage> &msg) {
294 switch (msg->what()) {
295 case kWhatReadBuffer:
296 {
297 onReadBuffer();
298
299 {
300 Mutex::Autolock _l(mBufferingLock);
301 mBuffering = false;
302 }
303 break;
304 }
305 default:
306 {
307 TRESPASS();
308 }
309 }
310 }
311
312
313 } // namespace android
314
315