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/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) {
135 // XXX legacy
136
137 if (extra == NULL) {
138 extra = new AMessage;
139 }
140
141 uint8_t type = buffer[1];
142
143 if (type & 2) {
144 int64_t mediaTimeUs;
145 memcpy(&mediaTimeUs, &buffer[2], sizeof(mediaTimeUs));
146
147 extra->setInt64(kATSParserKeyMediaTimeUs, mediaTimeUs);
148 }
149
150 mTSParser->signalDiscontinuity(
151 ((type & 1) == 0)
152 ? ATSParser::DISCONTINUITY_TIME
153 : ATSParser::DISCONTINUITY_FORMATCHANGE,
154 extra);
155 } else {
156 status_t err = mTSParser->feedTSPacket(buffer, sizeof(buffer));
157
158 if (err != OK) {
159 ALOGE("TS Parser returned error %d", err);
160
161 mTSParser->signalEOS(err);
162 setError(err);
163 break;
164 }
165 }
166 }
167 }
168 }
169
postReadBuffer()170 status_t NuPlayer::StreamingSource::postReadBuffer() {
171 {
172 Mutex::Autolock _l(mBufferingLock);
173 if (mFinalResult != OK) {
174 return mFinalResult;
175 }
176 if (mBuffering) {
177 return OK;
178 }
179 mBuffering = true;
180 }
181
182 (new AMessage(kWhatReadBuffer, this))->post();
183 return OK;
184 }
185
haveSufficientDataOnAllTracks()186 bool NuPlayer::StreamingSource::haveSufficientDataOnAllTracks() {
187 // We're going to buffer at least 2 secs worth data on all tracks before
188 // starting playback (both at startup and after a seek).
189
190 static const int64_t kMinDurationUs = 2000000LL;
191
192 sp<AnotherPacketSource> audioTrack = getSource(true /*audio*/);
193 sp<AnotherPacketSource> videoTrack = getSource(false /*audio*/);
194
195 status_t err;
196 int64_t durationUs;
197 if (audioTrack != NULL
198 && (durationUs = audioTrack->getBufferedDurationUs(&err))
199 < kMinDurationUs
200 && err == OK) {
201 ALOGV("audio track doesn't have enough data yet. (%.2f secs buffered)",
202 durationUs / 1E6);
203 return false;
204 }
205
206 if (videoTrack != NULL
207 && (durationUs = videoTrack->getBufferedDurationUs(&err))
208 < kMinDurationUs
209 && err == OK) {
210 ALOGV("video track doesn't have enough data yet. (%.2f secs buffered)",
211 durationUs / 1E6);
212 return false;
213 }
214
215 return true;
216 }
217
setError(status_t err)218 void NuPlayer::StreamingSource::setError(status_t err) {
219 Mutex::Autolock _l(mBufferingLock);
220 mFinalResult = err;
221 }
222
getSource(bool audio)223 sp<AnotherPacketSource> NuPlayer::StreamingSource::getSource(bool audio) {
224 if (mTSParser == NULL) {
225 return NULL;
226 }
227
228 sp<MediaSource> source = mTSParser->getSource(
229 audio ? ATSParser::AUDIO : ATSParser::VIDEO);
230
231 return static_cast<AnotherPacketSource *>(source.get());
232 }
233
getFormat(bool audio)234 sp<AMessage> NuPlayer::StreamingSource::getFormat(bool audio) {
235 sp<AnotherPacketSource> source = getSource(audio);
236
237 sp<AMessage> format = new AMessage;
238 if (source == NULL) {
239 format->setInt32("err", -EWOULDBLOCK);
240 return format;
241 }
242
243 sp<MetaData> meta = source->getFormat();
244 if (meta == NULL) {
245 format->setInt32("err", -EWOULDBLOCK);
246 return format;
247 }
248 status_t err = convertMetaDataToMessage(meta, &format);
249 if (err != OK) { // format may have been cleared on error
250 return NULL;
251 }
252 return format;
253 }
254
dequeueAccessUnit(bool audio,sp<ABuffer> * accessUnit)255 status_t NuPlayer::StreamingSource::dequeueAccessUnit(
256 bool audio, sp<ABuffer> *accessUnit) {
257 sp<AnotherPacketSource> source = getSource(audio);
258
259 if (source == NULL) {
260 return -EWOULDBLOCK;
261 }
262
263 if (!haveSufficientDataOnAllTracks()) {
264 postReadBuffer();
265 }
266
267 status_t finalResult;
268 if (!source->hasBufferAvailable(&finalResult)) {
269 return finalResult == OK ? -EWOULDBLOCK : finalResult;
270 }
271
272 status_t err = source->dequeueAccessUnit(accessUnit);
273
274 #if !defined(LOG_NDEBUG) || LOG_NDEBUG == 0
275 if (err == OK) {
276 int64_t timeUs;
277 CHECK((*accessUnit)->meta()->findInt64("timeUs", &timeUs));
278 ALOGV("dequeueAccessUnit timeUs=%lld us", timeUs);
279 }
280 #endif
281
282 return err;
283 }
284
isRealTime() const285 bool NuPlayer::StreamingSource::isRealTime() const {
286 return mSource->flags() & IStreamSource::kFlagIsRealTimeData;
287 }
288
onMessageReceived(const sp<AMessage> & msg)289 void NuPlayer::StreamingSource::onMessageReceived(
290 const sp<AMessage> &msg) {
291 switch (msg->what()) {
292 case kWhatReadBuffer:
293 {
294 onReadBuffer();
295
296 {
297 Mutex::Autolock _l(mBufferingLock);
298 mBuffering = false;
299 }
300 break;
301 }
302 default:
303 {
304 TRESPASS();
305 }
306 }
307 }
308
309
310 } // namespace android
311
312