#include "include/ThrottledSource.h"
#include "include/MPEG2TSExtractor.h"
-#include "ARTPSession.h"
-#include "APacketSource.h"
-#include "ASessionDescription.h"
-#include "UDPPusher.h"
-
#include <binder/IPCThreadState.h>
#include <media/stagefright/foundation/hexdump.h>
#include <media/stagefright/foundation/ADebug.h>
static int64_t kLowWaterMarkUs = 2000000ll; // 2secs
static int64_t kHighWaterMarkUs = 10000000ll; // 10secs
+static int64_t kHighWaterMarkRTSPUs = 4000000ll; // 4secs
static const size_t kLowWaterMarkBytes = 40000;
static const size_t kHighWaterMarkBytes = 200000;
mLiveSession.clear();
}
- mRTPPusher.clear();
- mRTCPPusher.clear();
- mRTPSession.clear();
-
if (mVideoSource != NULL) {
mVideoSource->stop();
LOGV("cachedDurationUs = %.2f secs, eos=%d",
cachedDurationUs / 1E6, eos);
+ int64_t highWaterMarkUs =
+ (mRTSPController != NULL) ? kHighWaterMarkRTSPUs : kHighWaterMarkUs;
+
if ((mFlags & PLAYING) && !eos
&& (cachedDurationUs < kLowWaterMarkUs)) {
LOGI("cache is running low (%.2f secs) , pausing.",
pause_l();
ensureCacheIsFetching_l();
notifyListener_l(MEDIA_INFO, MEDIA_INFO_BUFFERING_START);
- } else if (eos || cachedDurationUs > kHighWaterMarkUs) {
+ } else if (eos || cachedDurationUs > highWaterMarkUs) {
if (mFlags & CACHE_UNDERRUN) {
LOGI("cache has filled up (%.2f secs), resuming.",
cachedDurationUs / 1E6);
mTimeSourceDeltaUs = realTimeUs - mediaTimeUs;
}
- if (!wasSeeking && mRTPSession == NULL) {
+ if (!wasSeeking) {
// Let's display the first frame after seeking right away.
- // We'll completely ignore timestamps for gtalk videochat
- // and we'll play incoming video as fast as we get it.
int64_t nowUs = ts->getRealTimeUs() - mTimeSourceDeltaUs;
->setLiveSession(mLiveSession);
return setDataSource_l(extractor);
- } else if (!strncmp("rtsp://gtalk/", mUri.string(), 13)) {
- if (mLooper == NULL) {
- mLooper = new ALooper;
- mLooper->setName("gtalk rtp");
- mLooper->start(
- false /* runOnCallingThread */,
- false /* canCallJava */,
- PRIORITY_HIGHEST);
- }
-
- const char *startOfCodecString = &mUri.string()[13];
- const char *startOfSlash1 = strchr(startOfCodecString, '/');
- if (startOfSlash1 == NULL) {
- return BAD_VALUE;
- }
- const char *startOfWidthString = &startOfSlash1[1];
- const char *startOfSlash2 = strchr(startOfWidthString, '/');
- if (startOfSlash2 == NULL) {
- return BAD_VALUE;
- }
- const char *startOfHeightString = &startOfSlash2[1];
-
- String8 codecString(startOfCodecString, startOfSlash1 - startOfCodecString);
- String8 widthString(startOfWidthString, startOfSlash2 - startOfWidthString);
- String8 heightString(startOfHeightString);
-
-#if 0
- mRTPPusher = new UDPPusher("/data/misc/rtpout.bin", 5434);
- mLooper->registerHandler(mRTPPusher);
-
- mRTCPPusher = new UDPPusher("/data/misc/rtcpout.bin", 5435);
- mLooper->registerHandler(mRTCPPusher);
-#endif
-
- mRTPSession = new ARTPSession;
- mLooper->registerHandler(mRTPSession);
-
-#if 0
- // My AMR SDP
- static const char *raw =
- "v=0\r\n"
- "o=- 64 233572944 IN IP4 127.0.0.0\r\n"
- "s=QuickTime\r\n"
- "t=0 0\r\n"
- "a=range:npt=0-315\r\n"
- "a=isma-compliance:2,2.0,2\r\n"
- "m=audio 5434 RTP/AVP 97\r\n"
- "c=IN IP4 127.0.0.1\r\n"
- "b=AS:30\r\n"
- "a=rtpmap:97 AMR/8000/1\r\n"
- "a=fmtp:97 octet-align\r\n";
-#elif 1
- String8 sdp;
- sdp.appendFormat(
- "v=0\r\n"
- "o=- 64 233572944 IN IP4 127.0.0.0\r\n"
- "s=QuickTime\r\n"
- "t=0 0\r\n"
- "a=range:npt=0-315\r\n"
- "a=isma-compliance:2,2.0,2\r\n"
- "m=video 5434 RTP/AVP 97\r\n"
- "c=IN IP4 127.0.0.1\r\n"
- "b=AS:30\r\n"
- "a=rtpmap:97 %s/90000\r\n"
- "a=cliprect:0,0,%s,%s\r\n"
- "a=framesize:97 %s-%s\r\n",
-
- codecString.string(),
- heightString.string(), widthString.string(),
- widthString.string(), heightString.string()
- );
- const char *raw = sdp.string();
-
-#endif
-
- sp<ASessionDescription> desc = new ASessionDescription;
- CHECK(desc->setTo(raw, strlen(raw)));
-
- CHECK_EQ(mRTPSession->setup(desc), (status_t)OK);
-
- if (mRTPPusher != NULL) {
- mRTPPusher->start();
- }
-
- if (mRTCPPusher != NULL) {
- mRTCPPusher->start();
- }
-
- CHECK_EQ(mRTPSession->countTracks(), 1u);
- sp<MediaSource> source = mRTPSession->trackAt(0);
-
-#if 0
- bool eos;
- while (((APacketSource *)source.get())
- ->getQueuedDuration(&eos) < 5000000ll && !eos) {
- usleep(100000ll);
- }
-#endif
-
- const char *mime;
- CHECK(source->getFormat()->findCString(kKeyMIMEType, &mime));
-
- if (!strncasecmp("video/", mime, 6)) {
- setVideoSource(source);
- } else {
- CHECK(!strncasecmp("audio/", mime, 6));
- setAudioSource(source);
- }
-
- mExtractorFlags = MediaExtractor::CAN_PAUSE;
-
- return OK;
} else if (!strncasecmp("rtsp://", mUri.string(), 7)) {
if (mLooper == NULL) {
mLooper = new ALooper;
struct ALooper;
struct ARTSPController;
-struct ARTPSession;
-struct UDPPusher;
class DrmManagerClinet;
class DecryptHandle;
sp<ALooper> mLooper;
sp<ARTSPController> mRTSPController;
- sp<ARTPSession> mRTPSession;
- sp<UDPPusher> mRTPPusher, mRTCPPusher;
sp<LiveSession> mLiveSession;
// static
void ARTPAssembler::CopyTimes(const sp<ABuffer> &to, const sp<ABuffer> &from) {
- uint64_t ntpTime;
- CHECK(from->meta()->findInt64("ntp-time", (int64_t *)&ntpTime));
-
uint32_t rtpTime;
CHECK(from->meta()->findInt32("rtp-time", (int32_t *)&rtpTime));
- to->meta()->setInt64("ntp-time", ntpTime);
to->meta()->setInt32("rtp-time", rtpTime);
// Copy the seq number.
break;
}
- case kWhatFakeTimestamps:
- {
- onFakeTimestamps();
- break;
- }
-
default:
{
TRESPASS();
buffer->setInt32Data(u16at(&data[2]));
buffer->setRange(payloadOffset, size - payloadOffset);
- if ((mFlags & kFakeTimestamps) && !source->timeEstablished()) {
- source->timeUpdate(rtpTime, 0);
- source->timeUpdate(rtpTime + 90000, 0x100000000ll);
- CHECK(source->timeEstablished());
- }
-
source->processRTPPacket(buffer);
return OK;
sp<ARTPSource> source = findSource(s, id);
- if ((mFlags & kFakeTimestamps) == 0) {
- source->timeUpdate(rtpTime, ntpTime);
- }
+ source->timeUpdate(rtpTime, ntpTime);
return 0;
}
}
}
-void ARTPConnection::fakeTimestamps() {
- (new AMessage(kWhatFakeTimestamps, id()))->post();
-}
-
-void ARTPConnection::onFakeTimestamps() {
- List<StreamInfo>::iterator it = mStreams.begin();
- while (it != mStreams.end()) {
- StreamInfo &info = *it++;
-
- for (size_t j = 0; j < info.mSources.size(); ++j) {
- sp<ARTPSource> source = info.mSources.valueAt(j);
-
- if (!source->timeEstablished()) {
- source->timeUpdate(0, 0);
- source->timeUpdate(0 + 90000, 0x100000000ll);
-
- mFlags |= kFakeTimestamps;
- }
- }
- }
-}
-
} // namespace android
struct ARTPConnection : public AHandler {
enum Flags {
- kFakeTimestamps = 1,
kRegularlyRequestFIR = 2,
};
static void MakePortPair(
int *rtpSocket, int *rtcpSocket, unsigned *rtpPort);
- void fakeTimestamps();
-
protected:
virtual ~ARTPConnection();
virtual void onMessageReceived(const sp<AMessage> &msg);
kWhatRemoveStream,
kWhatPollStreams,
kWhatInjectPacket,
- kWhatFakeTimestamps,
};
static const int64_t kSelectTimeoutUs;
void onPollStreams();
void onInjectPacket(const sp<AMessage> &msg);
void onSendReceiverReports();
- void onFakeTimestamps();
status_t receive(StreamInfo *info, bool receiveRTP);
mDesc = desc;
- mRTPConn = new ARTPConnection(
- ARTPConnection::kFakeTimestamps
- | ARTPConnection::kRegularlyRequestFIR);
+ mRTPConn = new ARTPConnection(ARTPConnection::kRegularlyRequestFIR);
looper()->registerHandler(mRTPConn);
: mID(id),
mHighestSeqNumber(0),
mNumBuffersReceived(0),
- mNumTimes(0),
mLastNTPTime(0),
mLastNTPTimeUpdateUs(0),
mIssueFIRRequests(false),
mLastFIRRequestUs(-1),
- mNextFIRSeqNo((rand() * 256.0) / RAND_MAX) {
+ mNextFIRSeqNo((rand() * 256.0) / RAND_MAX),
+ mNotify(notify) {
unsigned long PT;
AString desc;
AString params;
}
void ARTPSource::processRTPPacket(const sp<ABuffer> &buffer) {
- if (queuePacket(buffer)
- && mNumTimes == 2
- && mAssembler != NULL) {
+ if (queuePacket(buffer) && mAssembler != NULL) {
mAssembler->onPacketReceived(this);
}
}
void ARTPSource::timeUpdate(uint32_t rtpTime, uint64_t ntpTime) {
- LOGV("timeUpdate");
-
mLastNTPTime = ntpTime;
mLastNTPTimeUpdateUs = ALooper::GetNowUs();
- if (mNumTimes == 2) {
- mNTPTime[0] = mNTPTime[1];
- mRTPTime[0] = mRTPTime[1];
- mNumTimes = 1;
- }
- mNTPTime[mNumTimes] = ntpTime;
- mRTPTime[mNumTimes++] = rtpTime;
-
- if (timeEstablished()) {
- for (List<sp<ABuffer> >::iterator it = mQueue.begin();
- it != mQueue.end(); ++it) {
- sp<AMessage> meta = (*it)->meta();
-
- uint32_t rtpTime;
- CHECK(meta->findInt32("rtp-time", (int32_t *)&rtpTime));
-
- meta->setInt64("ntp-time", RTP2NTP(rtpTime));
- }
- }
+ sp<AMessage> notify = mNotify->dup();
+ notify->setInt32("time-update", true);
+ notify->setInt32("rtp-time", rtpTime);
+ notify->setInt64("ntp-time", ntpTime);
+ notify->post();
}
bool ARTPSource::queuePacket(const sp<ABuffer> &buffer) {
uint32_t seqNum = (uint32_t)buffer->int32Data();
- if (mNumTimes == 2) {
- sp<AMessage> meta = buffer->meta();
-
- uint32_t rtpTime;
- CHECK(meta->findInt32("rtp-time", (int32_t *)&rtpTime));
-
- meta->setInt64("ntp-time", RTP2NTP(rtpTime));
- }
-
if (mNumBuffersReceived++ == 0) {
mHighestSeqNumber = seqNum;
mQueue.push_back(buffer);
return true;
}
-uint64_t ARTPSource::RTP2NTP(uint32_t rtpTime) const {
- CHECK_EQ(mNumTimes, 2u);
-
- return mNTPTime[0] + (double)(mNTPTime[1] - mNTPTime[0])
- * ((double)rtpTime - (double)mRTPTime[0])
- / (double)(mRTPTime[1] - mRTPTime[0]);
-}
-
void ARTPSource::byeReceived() {
mAssembler->onByeReceived();
}
void addReceiverReport(const sp<ABuffer> &buffer);
void addFIR(const sp<ABuffer> &buffer);
- bool timeEstablished() const {
- return mNumTimes == 2;
- }
-
private:
uint32_t mID;
uint32_t mHighestSeqNumber;
List<sp<ABuffer> > mQueue;
sp<ARTPAssembler> mAssembler;
- size_t mNumTimes;
- uint64_t mNTPTime[2];
- uint32_t mRTPTime[2];
-
uint64_t mLastNTPTime;
int64_t mLastNTPTimeUpdateUs;
int64_t mLastFIRRequestUs;
uint8_t mNextFIRSeqNo;
- uint64_t RTP2NTP(uint32_t rtpTime) const;
+ sp<AMessage> mNotify;
bool queuePacket(const sp<ABuffer> &buffer);
APacketSource.cpp \
ARTPAssembler.cpp \
ARTPConnection.cpp \
- ARTPSession.cpp \
ARTPSource.cpp \
ARTPWriter.cpp \
ARTSPConnection.cpp \
ARTSPController.cpp \
ASessionDescription.cpp \
- UDPPusher.cpp \
LOCAL_C_INCLUDES:= \
$(JNI_H_INCLUDE) \
#include <arpa/inet.h>
#include <sys/socket.h>
-// If no access units are received within 3 secs, assume that the rtp
+// If no access units are received within 5 secs, assume that the rtp
// stream has ended and signal end of stream.
-static int64_t kAccessUnitTimeoutUs = 3000000ll;
+static int64_t kAccessUnitTimeoutUs = 5000000ll;
// If no access units arrive for the first 10 secs after starting the
// stream, assume none ever will and signal EOS or switch transports.
mSetupTracksSuccessful(false),
mSeekPending(false),
mFirstAccessUnit(true),
- mFirstAccessUnitNTP(0),
+ mNTPAnchorUs(-1),
+ mMediaAnchorUs(-1),
+ mLastMediaTimeUs(0),
mNumAccessUnitsReceived(0),
mCheckPending(false),
mCheckGeneration(0),
mSetupTracksSuccessful = false;
mSeekPending = false;
mFirstAccessUnit = true;
- mFirstAccessUnitNTP = 0;
+ mNTPAnchorUs = -1;
+ mMediaAnchorUs = -1;
mNumAccessUnitsReceived = 0;
mReceivedFirstRTCPPacket = false;
mReceivedFirstRTPPacket = false;
case 'accu':
{
+ int32_t timeUpdate;
+ if (msg->findInt32("time-update", &timeUpdate) && timeUpdate) {
+ size_t trackIndex;
+ CHECK(msg->findSize("track-index", &trackIndex));
+
+ uint32_t rtpTime;
+ uint64_t ntpTime;
+ CHECK(msg->findInt32("rtp-time", (int32_t *)&rtpTime));
+ CHECK(msg->findInt64("ntp-time", (int64_t *)&ntpTime));
+
+ onTimeUpdate(trackIndex, rtpTime, ntpTime);
+ break;
+ }
+
int32_t first;
if (msg->findInt32("first-rtcp", &first)) {
mReceivedFirstRTCPPacket = true;
break;
}
- uint64_t ntpTime;
- CHECK(accessUnit->meta()->findInt64(
- "ntp-time", (int64_t *)&ntpTime));
-
- uint32_t rtpTime;
- CHECK(accessUnit->meta()->findInt32(
- "rtp-time", (int32_t *)&rtpTime));
-
if (track->mNewSegment) {
track->mNewSegment = false;
-
- LOGV("first segment unit ntpTime=0x%016llx rtpTime=%u seq=%d",
- ntpTime, rtpTime, seqNum);
}
- if (mFirstAccessUnit) {
- mDoneMsg->setInt32("result", OK);
- mDoneMsg->post();
- mDoneMsg = NULL;
-
- mFirstAccessUnit = false;
- mFirstAccessUnitNTP = ntpTime;
- }
-
- if (ntpTime >= mFirstAccessUnitNTP) {
- ntpTime -= mFirstAccessUnitNTP;
- } else {
- ntpTime = 0;
- }
-
- int64_t timeUs = (int64_t)(ntpTime * 1E6 / (1ll << 32));
-
- accessUnit->meta()->setInt64("timeUs", timeUs);
-
-#if 0
- int32_t damaged;
- if (accessUnit->meta()->findInt32("damaged", &damaged)
- && damaged != 0) {
- LOGI("ignoring damaged AU");
- } else
-#endif
- {
- TrackInfo *track = &mTracks.editItemAt(trackIndex);
- track->mPacketSource->queueAccessUnit(accessUnit);
- }
+ onAccessUnitComplete(trackIndex, accessUnit);
break;
}
{
// Session is paused now.
for (size_t i = 0; i < mTracks.size(); ++i) {
- mTracks.editItemAt(i).mPacketSource->flushQueue();
+ TrackInfo *info = &mTracks.editItemAt(i);
+
+ info->mPacketSource->flushQueue();
+ info->mRTPAnchor = 0;
+ info->mNTPAnchorUs = -1;
}
+ mNTPAnchorUs = -1;
+
int64_t timeUs;
CHECK(msg->findInt64("time", &timeUs));
} else {
parsePlayResponse(response);
+ ssize_t i = response->mHeaders.indexOfKey("rtp-info");
+ CHECK_GE(i, 0);
+
+ LOGV("rtp-info: %s", response->mHeaders.valueAt(i).c_str());
+
LOGI("seek completed.");
}
}
mTryFakeRTCP = true;
mReceivedFirstRTCPPacket = true;
- mRTPConn->fakeTimestamps();
} else {
LOGW("Never received any data, switching transports.");
uint32_t rtpTime = strtoul(val.c_str(), &end, 10);
- LOGV("track #%d: rtpTime=%u <=> ntp=%.2f", n, rtpTime, npt1);
+ LOGV("track #%d: rtpTime=%u <=> npt=%.2f", n, rtpTime, npt1);
info->mPacketSource->setNormalPlayTimeMapping(
rtpTime, (int64_t)(npt1 * 1E6));
}
private:
+ struct TrackInfo {
+ AString mURL;
+ int mRTPSocket;
+ int mRTCPSocket;
+ bool mUsingInterleavedTCP;
+ uint32_t mFirstSeqNumInSegment;
+ bool mNewSegment;
+
+ uint32_t mRTPAnchor;
+ int64_t mNTPAnchorUs;
+ int32_t mTimeScale;
+
+ sp<APacketSource> mPacketSource;
+
+ // Stores packets temporarily while no notion of time
+ // has been established yet.
+ List<sp<ABuffer> > mPackets;
+ };
+
sp<ALooper> mLooper;
sp<ALooper> mNetLooper;
sp<ARTSPConnection> mConn;
bool mSetupTracksSuccessful;
bool mSeekPending;
bool mFirstAccessUnit;
- uint64_t mFirstAccessUnitNTP;
+
+ int64_t mNTPAnchorUs;
+ int64_t mMediaAnchorUs;
+ int64_t mLastMediaTimeUs;
+
int64_t mNumAccessUnitsReceived;
bool mCheckPending;
int32_t mCheckGeneration;
bool mReceivedFirstRTPPacket;
bool mSeekable;
- struct TrackInfo {
- AString mURL;
- int mRTPSocket;
- int mRTCPSocket;
- bool mUsingInterleavedTCP;
- uint32_t mFirstSeqNumInSegment;
- bool mNewSegment;
-
- sp<APacketSource> mPacketSource;
- };
Vector<TrackInfo> mTracks;
sp<AMessage> mDoneMsg;
info->mUsingInterleavedTCP = false;
info->mFirstSeqNumInSegment = 0;
info->mNewSegment = true;
+ info->mRTPAnchor = 0;
+ info->mNTPAnchorUs = -1;
+
+ unsigned long PT;
+ AString formatDesc;
+ AString formatParams;
+ mSessionDesc->getFormatType(index, &PT, &formatDesc, &formatParams);
+
+ int32_t timescale;
+ int32_t numChannels;
+ ASessionDescription::ParseFormatDesc(
+ formatDesc.c_str(), ×cale, &numChannels);
+
+ info->mTimeScale = timescale;
LOGV("track #%d URL=%s", mTracks.size(), trackURL.c_str());
return true;
}
+ void onTimeUpdate(int32_t trackIndex, uint32_t rtpTime, uint64_t ntpTime) {
+ LOGV("onTimeUpdate track %d, rtpTime = 0x%08x, ntpTime = 0x%016llx",
+ trackIndex, rtpTime, ntpTime);
+
+ int64_t ntpTimeUs = (int64_t)(ntpTime * 1E6 / (1ll << 32));
+
+ TrackInfo *track = &mTracks.editItemAt(trackIndex);
+
+ track->mRTPAnchor = rtpTime;
+ track->mNTPAnchorUs = ntpTimeUs;
+
+ if (mNTPAnchorUs < 0) {
+ mNTPAnchorUs = ntpTimeUs;
+ mMediaAnchorUs = mLastMediaTimeUs;
+ }
+ }
+
+ void onAccessUnitComplete(
+ int32_t trackIndex, const sp<ABuffer> &accessUnit) {
+ LOGV("onAccessUnitComplete track %d", trackIndex);
+
+ if (mFirstAccessUnit) {
+ mDoneMsg->setInt32("result", OK);
+ mDoneMsg->post();
+ mDoneMsg = NULL;
+
+ mFirstAccessUnit = false;
+ }
+
+ TrackInfo *track = &mTracks.editItemAt(trackIndex);
+
+ if (mNTPAnchorUs < 0 || mMediaAnchorUs < 0 || track->mNTPAnchorUs < 0) {
+ LOGV("storing accessUnit, no time established yet");
+ track->mPackets.push_back(accessUnit);
+ return;
+ }
+
+ while (!track->mPackets.empty()) {
+ sp<ABuffer> accessUnit = *track->mPackets.begin();
+ track->mPackets.erase(track->mPackets.begin());
+
+ if (addMediaTimestamp(trackIndex, track, accessUnit)) {
+ track->mPacketSource->queueAccessUnit(accessUnit);
+ }
+ }
+
+ if (addMediaTimestamp(trackIndex, track, accessUnit)) {
+ track->mPacketSource->queueAccessUnit(accessUnit);
+ }
+ }
+
+ bool addMediaTimestamp(
+ int32_t trackIndex, const TrackInfo *track,
+ const sp<ABuffer> &accessUnit) {
+ uint32_t rtpTime;
+ CHECK(accessUnit->meta()->findInt32(
+ "rtp-time", (int32_t *)&rtpTime));
+
+ int64_t relRtpTimeUs =
+ (((int64_t)rtpTime - (int64_t)track->mRTPAnchor) * 1000000ll)
+ / track->mTimeScale;
+
+ int64_t ntpTimeUs = track->mNTPAnchorUs + relRtpTimeUs;
+
+ int64_t mediaTimeUs = mMediaAnchorUs + ntpTimeUs - mNTPAnchorUs;
+
+ if (mediaTimeUs > mLastMediaTimeUs) {
+ mLastMediaTimeUs = mediaTimeUs;
+ }
+
+ if (mediaTimeUs < 0) {
+ LOGV("dropping early accessUnit.");
+ return false;
+ }
+
+ LOGV("track %d rtpTime=%d mediaTimeUs = %lld us (%.2f secs)",
+ trackIndex, rtpTime, mediaTimeUs, mediaTimeUs / 1E6);
+
+ accessUnit->meta()->setInt64("timeUs", mediaTimeUs);
+
+ return true;
+ }
+
+
DISALLOW_EVIL_CONSTRUCTORS(MyHandler);
};
OSDN Git Service
