Merge "Various changes to improve rtsp networking, reduce packet loss and adapt to ALooper API changes." into gingerbread

cmds/stagefright/stagefright.cpp

@@ -25,6 +25,8 @@
 #include <binder/IServiceManager.h>
 #include <binder/ProcessState.h>
 #include <media/IMediaPlayerService.h>
+#include <media/stagefright/foundation/ALooper.h>
+#include "include/ARTSPController.h"
 #include <media/stagefright/AudioPlayer.h>
 #include <media/stagefright/DataSource.h>
 #include <media/stagefright/JPEGSource.h>
@@ -365,6 +367,9 @@ int main(int argc, char **argv) {
     gPlaybackAudio = false;
     gWriteMP4 = false;
 
+    sp<ALooper> looper;
+    sp<ARTSPController> rtspController;
+
     int res;
     while ((res = getopt(argc, argv, "han:lm:b:ptsow:k")) >= 0) {
         switch (res) {
@@ -576,7 +581,8 @@ int main(int argc, char **argv) {
 
         sp<DataSource> dataSource = DataSource::CreateFromURI(filename);
 
-        if (strncasecmp(filename, "sine:", 5) && dataSource == NULL) {
+        if ((strncasecmp(filename, "sine:", 5)
+                && strncasecmp(filename, "rtsp://", 7)) && dataSource == NULL) {
             fprintf(stderr, "Unable to create data source.\n");
             return 1;
         }
@@ -601,11 +607,29 @@ int main(int argc, char **argv) {
             }
             mediaSource = new SineSource(sampleRate, 1);
         } else {
-            sp<MediaExtractor> extractor = MediaExtractor::Create(dataSource);
+            sp<MediaExtractor> extractor;
+
+            if (!strncasecmp("rtsp://", filename, 7)) {
+                if (looper == NULL) {
+                    looper = new ALooper;
+                    looper->start();
+                }
+
+                rtspController = new ARTSPController(looper);
+                status_t err = rtspController->connect(filename);
+                if (err != OK) {
+                    fprintf(stderr, "could not connect to rtsp server.\n");
+                    return -1;
+                }
+
+                extractor = rtspController.get();
+            } else {
+                extractor = MediaExtractor::Create(dataSource);
                 if (extractor == NULL) {
                     fprintf(stderr, "could not create extractor.\n");
                     return -1;
                 }
+            }
 
             size_t numTracks = extractor->countTracks();
 
@@ -654,6 +678,13 @@ int main(int argc, char **argv) {
         } else {
             playSource(&client, mediaSource);
         }
+
+        if (rtspController != NULL) {
+            rtspController->disconnect();
+            rtspController.clear();
+
+            sleep(3);
+        }
     }
 
     client.disconnect();

include/media/stagefright/foundation/ALooper.h

@@ -41,7 +41,9 @@ struct ALooper : public RefBase {
 
     status_t start(
             bool runOnCallingThread = false,
-            bool canCallJava = false);
+            bool canCallJava = false,
+            int32_t priority = PRIORITY_DEFAULT
+            );
 
     status_t stop();
 

media/libstagefright/foundation/ALooper.cpp

@@ -73,7 +73,8 @@ void ALooper::unregisterHandler(handler_id handlerID) {
     gLooperRoster.unregisterHandler(handlerID);
 }
 
-status_t ALooper::start(bool runOnCallingThread, bool canCallJava) {
+status_t ALooper::start(
+        bool runOnCallingThread, bool canCallJava, int32_t priority) {
     if (runOnCallingThread) {
         {
             Mutex::Autolock autoLock(mLock);
@@ -99,7 +100,7 @@ status_t ALooper::start(bool runOnCallingThread, bool canCallJava) {
 
     mThread = new LooperThread(this, canCallJava);
 
-    status_t err = mThread->run("ALooper");
+    status_t err = mThread->run("ALooper", priority);
     if (err != OK) {
         mThread.clear();
     }

media/libstagefright/rtsp/AMPEG4AudioAssembler.cpp

@@ -144,6 +144,11 @@ void AMPEG4AudioAssembler::submitAccessUnit() {
 
     accessUnit->meta()->setInt64("ntp-time", ntpTime);
 
+#if 0
+    printf(mAccessUnitDamaged ? "X" : ".");
+    fflush(stdout);
+#endif
+
     if (mAccessUnitDamaged) {
         accessUnit->meta()->setInt32("damaged", true);
     }

media/libstagefright/rtsp/ARTSPController.cpp

@@ -38,6 +38,8 @@ status_t ARTSPController::connect(const char *url) {
     }
 
     mHandler = new MyHandler(url, mLooper);
+    mHandler->connect();
+
     sleep(10);
 
     return OK;
@@ -48,6 +50,7 @@ void ARTSPController::disconnect() {
         return;
     }
 
+    mHandler->disconnect();
     mHandler.clear();
 }
 

media/libstagefright/rtsp/MyHandler.h

@@ -34,19 +34,33 @@ namespace android {
 struct MyHandler : public AHandler {
     MyHandler(const char *url, const sp<ALooper> &looper)
         : mLooper(looper),
+          mNetLooper(new ALooper),
           mConn(new ARTSPConnection),
           mRTPConn(new ARTPConnection),
           mSessionURL(url),
           mSetupTracksSuccessful(false),
           mFirstAccessUnit(true),
           mFirstAccessUnitNTP(-1) {
+
+        mNetLooper->start(false /* runOnCallingThread */,
+                          false /* canCallJava */,
+                          PRIORITY_HIGHEST);
+    }
+
+    void connect() {
         mLooper->registerHandler(this);
         mLooper->registerHandler(mConn);
-        mLooper->registerHandler(mRTPConn);
+        (1 ? mNetLooper : mLooper)->registerHandler(mRTPConn);
         sp<AMessage> reply = new AMessage('conn', id());
+
         mConn->connect(mSessionURL.c_str(), reply);
     }
 
+    void disconnect() {
+        sp<AMessage> reply = new AMessage('disc', id());
+        mConn->disconnect(reply);
+    }
+
     virtual void onMessageReceived(const sp<AMessage> &msg) {
         switch (msg->what()) {
             case 'conn':
@@ -290,7 +304,6 @@ struct MyHandler : public AHandler {
 
             case 'quit':
             {
-                mLooper->stop();
                 break;
             }
 
@@ -320,8 +333,17 @@ struct MyHandler : public AHandler {
 
                 accessUnit->meta()->setInt64("ntp-time", ntpTime);
 
+#if 0
+                int32_t damaged;
+                if (accessUnit->meta()->findInt32("damaged", &damaged)
+                        && damaged != 0) {
+                    LOG(INFO) << "ignoring damaged AU";
+                } else
+#endif
+                {
                     TrackInfo *track = &mTracks.editItemAt(trackIndex);
                     track->mPacketSource->queueAccessUnit(accessUnit);
+                }
                 break;
             }
 
@@ -344,6 +366,7 @@ struct MyHandler : public AHandler {
 
 private:
     sp<ALooper> mLooper;
+    sp<ALooper> mNetLooper;
     sp<ARTSPConnection> mConn;
     sp<ARTPConnection> mRTPConn;
     sp<ASessionDescription> mSessionDesc;