A first shot at proper support for seeking of rtsp streams.

Change-Id: I9604f2d09feedc0074c0e715be58e719d4483760
related-to-bug: 2556656

7 files changed, 144 insertions, 59 deletions

diff --git a/media/libstagefright/AwesomePlayer.cpp b/media/libstagefright/AwesomePlayer.cpp
index f2653cf..11fdf56 100644
--- a/media/libstagefright/AwesomePlayer.cpp
+++ b/media/libstagefright/AwesomePlayer.cpp
@@ -196,6 +196,7 @@ AwesomePlayer::AwesomePlayer()
       mExtractorFlags(0),
       mLastVideoBuffer(NULL),
       mVideoBuffer(NULL),
+      mRTSPTimeOffset(0),
       mSuspensionState(NULL) {
     CHECK_EQ(mClient.connect(), OK);
 
@@ -393,7 +394,11 @@ void AwesomePlayer::reset_l() {
         mVideoBuffer = NULL;
     }
 
-    mRTSPController.clear();
+    if (mRTSPController != NULL) {
+        mRTSPController->disconnect();
+        mRTSPController.clear();
+    }
+
     mRTPPusher.clear();
     mRTCPPusher.clear();
     mRTPSession.clear();
@@ -738,6 +743,10 @@ status_t AwesomePlayer::getPosition(int64_t *positionUs) {
         *positionUs = 0;
     }
 
+    if (mRTSPController != NULL) {
+        *positionUs += mRTSPTimeOffset;
+    }
+
     return OK;
 }
 
@@ -753,6 +762,17 @@ status_t AwesomePlayer::seekTo(int64_t timeUs) {
 }
 
 status_t AwesomePlayer::seekTo_l(int64_t timeUs) {
+    if (mRTSPController != NULL) {
+        pause_l();
+        mRTSPController->seek(timeUs);
+        play_l();
+
+        notifyListener_l(MEDIA_SEEK_COMPLETE);
+        mSeekNotificationSent = true;
+        mRTSPTimeOffset = timeUs;
+        return OK;
+    }
+
     if (mFlags & CACHE_UNDERRUN) {
         mFlags &= ~CACHE_UNDERRUN;
         play_l();
diff --git a/media/libstagefright/include/ARTSPController.h b/media/libstagefright/include/ARTSPController.h
index 2542e4e..7020564 100644
--- a/media/libstagefright/include/ARTSPController.h
+++ b/media/libstagefright/include/ARTSPController.h
@@ -33,6 +33,8 @@ struct ARTSPController : public MediaExtractor {
     status_t connect(const char *url);
     void disconnect();
 
+    void seek(int64_t timeUs);
+
     virtual size_t countTracks();
     virtual sp<MediaSource> getTrack(size_t index);
 
diff --git a/media/libstagefright/include/AwesomePlayer.h b/media/libstagefright/include/AwesomePlayer.h
index 55e2c36..49b5c78 100644
--- a/media/libstagefright/include/AwesomePlayer.h
+++ b/media/libstagefright/include/AwesomePlayer.h
@@ -180,6 +180,7 @@ private:
 
     sp<ALooper> mLooper;
     sp<ARTSPController> mRTSPController;
+    int64_t mRTSPTimeOffset;
     sp<ARTPSession> mRTPSession;
     sp<UDPPusher> mRTPPusher, mRTCPPusher;
 
diff --git a/media/libstagefright/rtsp/APacketSource.cpp b/media/libstagefright/rtsp/APacketSource.cpp
index 8c56cb7..b930184 100644
--- a/media/libstagefright/rtsp/APacketSource.cpp
+++ b/media/libstagefright/rtsp/APacketSource.cpp
@@ -406,9 +406,7 @@ APacketSource::APacketSource(
         const sp<ASessionDescription> &sessionDesc, size_t index)
     : mInitCheck(NO_INIT),
       mFormat(new MetaData),
-      mEOSResult(OK),
-      mFirstAccessUnit(true),
-      mFirstAccessUnitNTP(0) {
+      mEOSResult(OK) {
     unsigned long PT;
     AString desc;
     AString params;
@@ -550,9 +548,6 @@ status_t APacketSource::initCheck() const {
 }
 
 status_t APacketSource::start(MetaData *params) {
-    mFirstAccessUnit = true;
-    mFirstAccessUnitNTP = 0;
-
     return OK;
 }
 
@@ -600,25 +595,6 @@ void APacketSource::queueAccessUnit(const sp<ABuffer> &buffer) {
         return;
     }
 
-    uint64_t ntpTime;
-    CHECK(buffer->meta()->findInt64(
-                "ntp-time", (int64_t *)&ntpTime));
-
-    if (mFirstAccessUnit) {
-        mFirstAccessUnit = false;
-        mFirstAccessUnitNTP = ntpTime;
-    }
-
-    if (ntpTime > mFirstAccessUnitNTP) {
-        ntpTime -= mFirstAccessUnitNTP;
-    } else {
-        ntpTime = 0;
-    }
-
-    int64_t timeUs = (int64_t)(ntpTime * 1E6 / (1ll << 32));
-
-    buffer->meta()->setInt64("timeUs", timeUs);
-
     Mutex::Autolock autoLock(mLock);
     mBuffers.push_back(buffer);
     mCondition.signal();
@@ -632,31 +608,9 @@ void APacketSource::signalEOS(status_t result) {
     mCondition.signal();
 }
 
-int64_t APacketSource::getQueuedDuration(bool *eos) {
+void APacketSource::flushQueue() {
     Mutex::Autolock autoLock(mLock);
-
-    *eos = (mEOSResult != OK);
-
-    if (mBuffers.empty()) {
-        return 0;
-    }
-
-    sp<ABuffer> buffer = *mBuffers.begin();
-
-    uint64_t ntpTime;
-    CHECK(buffer->meta()->findInt64(
-                "ntp-time", (int64_t *)&ntpTime));
-
-    int64_t firstTimeUs = (int64_t)(ntpTime * 1E6 / (1ll << 32));
-
-    buffer = *--mBuffers.end();
-
-    CHECK(buffer->meta()->findInt64(
-                "ntp-time", (int64_t *)&ntpTime));
-
-    int64_t lastTimeUs = (int64_t)(ntpTime * 1E6 / (1ll << 32));
-
-    return lastTimeUs - firstTimeUs;
+    mBuffers.clear();
 }
 
 }  // namespace android
diff --git a/media/libstagefright/rtsp/APacketSource.h b/media/libstagefright/rtsp/APacketSource.h
index 647da6e..197af3e 100644
--- a/media/libstagefright/rtsp/APacketSource.h
+++ b/media/libstagefright/rtsp/APacketSource.h
@@ -43,7 +43,7 @@ struct APacketSource : public MediaSource {
     void queueAccessUnit(const sp<ABuffer> &buffer);
     void signalEOS(status_t result);
 
-    int64_t getQueuedDuration(bool *eos);
+    void flushQueue();
 
 protected:
     virtual ~APacketSource();
@@ -58,9 +58,6 @@ private:
     List<sp<ABuffer> > mBuffers;
     status_t mEOSResult;
 
-    bool mFirstAccessUnit;
-    uint64_t mFirstAccessUnitNTP;
-
     DISALLOW_EVIL_CONSTRUCTORS(APacketSource);
 };
 
diff --git a/media/libstagefright/rtsp/ARTSPController.cpp b/media/libstagefright/rtsp/ARTSPController.cpp
index ceae3a6..9df17cb 100644
--- a/media/libstagefright/rtsp/ARTSPController.cpp
+++ b/media/libstagefright/rtsp/ARTSPController.cpp
@@ -33,7 +33,7 @@ ARTSPController::ARTSPController(const sp<ALooper> &looper)
 }
 
 ARTSPController::~ARTSPController() {
-    disconnect();
+    CHECK_EQ((int)mState, (int)DISCONNECTED);
     mLooper->unregisterHandler(mReflector->id());
 }
 
@@ -80,6 +80,16 @@ void ARTSPController::disconnect() {
     mHandler.clear();
 }
 
+void ARTSPController::seek(int64_t timeUs) {
+    Mutex::Autolock autoLock(mLock);
+
+    if (mState != CONNECTED) {
+        return;
+    }
+
+    mHandler->seek(timeUs);
+}
+
 size_t ARTSPController::countTracks() {
     if (mHandler == NULL) {
         return 0;
diff --git a/media/libstagefright/rtsp/MyHandler.h b/media/libstagefright/rtsp/MyHandler.h
index 90070c9..e248463 100644
--- a/media/libstagefright/rtsp/MyHandler.h
+++ b/media/libstagefright/rtsp/MyHandler.h
@@ -38,7 +38,10 @@ struct MyHandler : public AHandler {
           mConn(new ARTSPConnection),
           mRTPConn(new ARTPConnection),
           mSessionURL(url),
-          mSetupTracksSuccessful(false) {
+          mSetupTracksSuccessful(false),
+          mSeekPending(false),
+          mFirstAccessUnit(true),
+          mFirstAccessUnitNTP(0) {
 
         mNetLooper->start(false /* runOnCallingThread */,
                           false /* canCallJava */,
@@ -62,6 +65,12 @@ struct MyHandler : public AHandler {
         (new AMessage('abor', id()))->post();
     }
 
+    void seek(int64_t timeUs) {
+        sp<AMessage> msg = new AMessage('seek', id());
+        msg->setInt64("time", timeUs);
+        msg->post();
+    }
+
     virtual void onMessageReceived(const sp<AMessage> &msg) {
         switch (msg->what()) {
             case 'conn':
@@ -88,8 +97,6 @@ struct MyHandler : public AHandler {
 
             case 'disc':
             {
-                LOG(INFO) << "disconnect completed";
-
                 (new AMessage('quit', id()))->post();
                 break;
             }
@@ -337,7 +344,20 @@ struct MyHandler : public AHandler {
                 CHECK(accessUnit->meta()->findInt64(
                             "ntp-time", (int64_t *)&ntpTime));
 
-                accessUnit->meta()->setInt64("ntp-time", ntpTime);
+                if (mFirstAccessUnit) {
+                    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;
@@ -353,6 +373,84 @@ struct MyHandler : public AHandler {
                 break;
             }
 
+            case 'seek':
+            {
+                if (mSeekPending) {
+                    break;
+                }
+
+                int64_t timeUs;
+                CHECK(msg->findInt64("time", &timeUs));
+
+                mSeekPending = true;
+
+                AString request = "PAUSE ";
+                request.append(mSessionURL);
+                request.append(" RTSP/1.0\r\n");
+
+                request.append("Session: ");
+                request.append(mSessionID);
+                request.append("\r\n");
+
+                request.append("\r\n");
+
+                sp<AMessage> reply = new AMessage('see1', id());
+                reply->setInt64("time", timeUs);
+                mConn->sendRequest(request.c_str(), reply);
+                break;
+            }
+
+            case 'see1':
+            {
+                int64_t timeUs;
+                CHECK(msg->findInt64("time", &timeUs));
+
+                AString request = "PLAY ";
+                request.append(mSessionURL);
+                request.append(" RTSP/1.0\r\n");
+
+                request.append("Session: ");
+                request.append(mSessionID);
+                request.append("\r\n");
+
+                request.append(
+                        StringPrintf(
+                            "Range: npt=%lld-\r\n", timeUs / 1000000ll));
+
+                request.append("\r\n");
+
+                sp<AMessage> reply = new AMessage('see2', id());
+                mConn->sendRequest(request.c_str(), reply);
+                break;
+            }
+
+            case 'see2':
+            {
+                CHECK(mSeekPending);
+
+                LOG(INFO) << "seek completed.";
+                mSeekPending = false;
+
+                int32_t result;
+                CHECK(msg->findInt32("result", &result));
+                if (result != OK) {
+                    LOG(ERROR) << "seek FAILED";
+                    break;
+                }
+
+                sp<RefBase> obj;
+                CHECK(msg->findObject("response", &obj));
+                sp<ARTSPResponse> response =
+                    static_cast<ARTSPResponse *>(obj.get());
+
+                CHECK_EQ(response->mStatusCode, 200u);
+
+                for (size_t i = 0; i < mTracks.size(); ++i) {
+                    mTracks.editItemAt(i).mPacketSource->flushQueue();
+                }
+                break;
+            }
+
             default:
                 TRESPASS();
                 break;
@@ -380,6 +478,9 @@ private:
     AString mBaseURL;
     AString mSessionID;
     bool mSetupTracksSuccessful;
+    bool mSeekPending;
+    bool mFirstAccessUnit;
+    uint64_t mFirstAccessUnitNTP;
 
     struct TrackInfo {
         int mRTPSocket;