1 files changed, 64 insertions, 17 deletions

diff --git a/media/libstagefright/NuCachedSource2.cpp b/media/libstagefright/NuCachedSource2.cpp
index 06e2d43..bd0a41d 100644
--- a/media/libstagefright/NuCachedSource2.cpp
+++ b/media/libstagefright/NuCachedSource2.cpp
@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+#include <inttypes.h>
+
 //#define LOG_NDEBUG 0
 #define LOG_TAG "NuCachedSource2"
 #include <utils/Log.h>
@@ -135,7 +137,7 @@ size_t PageCache::releaseFromStart(size_t maxBytes) {
 }
 
 void PageCache::copy(size_t from, void *data, size_t size) {
-    ALOGV("copy from %d size %d", from, size);
+    ALOGV("copy from %zu size %zu", from, size);
 
     if (size == 0) {
         return;
@@ -189,6 +191,7 @@ NuCachedSource2::NuCachedSource2(
       mFinalStatus(OK),
       mLastAccessPos(0),
       mFetching(true),
+      mDisconnecting(false),
       mLastFetchTimeUs(-1),
       mNumRetriesLeft(kMaxNumRetries),
       mHighwaterThresholdBytes(kDefaultHighWaterThreshold),
@@ -213,7 +216,14 @@ NuCachedSource2::NuCachedSource2(
 
     mLooper->setName("NuCachedSource2");
     mLooper->registerHandler(mReflector);
-    mLooper->start();
+
+    // Since it may not be obvious why our looper thread needs to be
+    // able to call into java since it doesn't appear to do so at all...
+    // IMediaHTTPConnection may be (and most likely is) implemented in JAVA
+    // and a local JAVA IBinder will call directly into JNI methods.
+    // So whenever we call DataSource::readAt it may end up in a call to
+    // IMediaHTTPConnection::readAt and therefore call back into JAVA.
+    mLooper->start(false /* runOnCallingThread */, true /* canCallJava */);
 
     Mutex::Autolock autoLock(mLock);
     (new AMessage(kWhatFetchMore, mReflector->id()))->post();
@@ -235,6 +245,27 @@ status_t NuCachedSource2::getEstimatedBandwidthKbps(int32_t *kbps) {
     return ERROR_UNSUPPORTED;
 }
 
+void NuCachedSource2::disconnect() {
+    if (mSource->flags() & kIsHTTPBasedSource) {
+        ALOGV("disconnecting HTTPBasedSource");
+
+        {
+            Mutex::Autolock autoLock(mLock);
+            // set mDisconnecting to true, if a fetch returns after
+            // this, the source will be marked as EOS.
+            mDisconnecting = true;
+
+            // explicitly signal mCondition so that the pending readAt()
+            // will immediately return
+            mCondition.signal();
+        }
+
+        // explicitly disconnect from the source, to allow any
+        // pending reads to return more promptly
+        static_cast<HTTPBase *>(mSource.get())->disconnect();
+    }
+}
+
 status_t NuCachedSource2::setCacheStatCollectFreq(int32_t freqMs) {
     if (mSource->flags() & kIsHTTPBasedSource) {
         HTTPBase *source = static_cast<HTTPBase *>(mSource.get());
@@ -298,7 +329,11 @@ void NuCachedSource2::fetchInternal() {
 
         Mutex::Autolock autoLock(mLock);
 
-        if (err == ERROR_UNSUPPORTED || err == -EPIPE) {
+        if (mDisconnecting) {
+            mNumRetriesLeft = 0;
+            mFinalStatus = ERROR_END_OF_STREAM;
+            return;
+        } else if (err == ERROR_UNSUPPORTED || err == -EPIPE) {
             // These are errors that are not likely to go away even if we
             // retry, i.e. the server doesn't support range requests or similar.
             mNumRetriesLeft = 0;
@@ -318,7 +353,14 @@ void NuCachedSource2::fetchInternal() {
 
     Mutex::Autolock autoLock(mLock);
 
-    if (n < 0) {
+    if (n == 0 || mDisconnecting) {
+        ALOGI("ERROR_END_OF_STREAM");
+
+        mNumRetriesLeft = 0;
+        mFinalStatus = ERROR_END_OF_STREAM;
+
+        mCache->releasePage(page);
+    } else if (n < 0) {
         mFinalStatus = n;
         if (n == ERROR_UNSUPPORTED || n == -EPIPE) {
             // These are errors that are not likely to go away even if we
@@ -326,14 +368,7 @@ void NuCachedSource2::fetchInternal() {
             mNumRetriesLeft = 0;
         }
 
-        ALOGE("source returned error %d, %d retries left", n, mNumRetriesLeft);
-        mCache->releasePage(page);
-    } else if (n == 0) {
-        ALOGI("ERROR_END_OF_STREAM");
-
-        mNumRetriesLeft = 0;
-        mFinalStatus = ERROR_END_OF_STREAM;
-
+        ALOGE("source returned error %zd, %d retries left", n, mNumRetriesLeft);
         mCache->releasePage(page);
     } else {
         if (mFinalStatus != OK) {
@@ -421,6 +456,10 @@ void NuCachedSource2::onRead(const sp<AMessage> &msg) {
     }
 
     Mutex::Autolock autoLock(mLock);
+    if (mDisconnecting) {
+        mCondition.signal();
+        return;
+    }
 
     CHECK(mAsyncResult == NULL);
 
@@ -457,16 +496,19 @@ void NuCachedSource2::restartPrefetcherIfNecessary_l(
     size_t actualBytes = mCache->releaseFromStart(maxBytes);
     mCacheOffset += actualBytes;
 
-    ALOGI("restarting prefetcher, totalSize = %d", mCache->totalSize());
+    ALOGI("restarting prefetcher, totalSize = %zu", mCache->totalSize());
     mFetching = true;
 }
 
 ssize_t NuCachedSource2::readAt(off64_t offset, void *data, size_t size) {
     Mutex::Autolock autoSerializer(mSerializer);
 
-    ALOGV("readAt offset %lld, size %d", offset, size);
+    ALOGV("readAt offset %lld, size %zu", offset, size);
 
     Mutex::Autolock autoLock(mLock);
+    if (mDisconnecting) {
+        return ERROR_END_OF_STREAM;
+    }
 
     // If the request can be completely satisfied from the cache, do so.
 
@@ -488,10 +530,15 @@ ssize_t NuCachedSource2::readAt(off64_t offset, void *data, size_t size) {
     CHECK(mAsyncResult == NULL);
     msg->post();
 
-    while (mAsyncResult == NULL) {
+    while (mAsyncResult == NULL && !mDisconnecting) {
         mCondition.wait(mLock);
     }
 
+    if (mDisconnecting) {
+        mAsyncResult.clear();
+        return ERROR_END_OF_STREAM;
+    }
+
     int32_t result;
     CHECK(mAsyncResult->findInt32("result", &result));
 
@@ -532,7 +579,7 @@ size_t NuCachedSource2::approxDataRemaining_l(status_t *finalStatus) const {
 ssize_t NuCachedSource2::readInternal(off64_t offset, void *data, size_t size) {
     CHECK_LE(size, (size_t)mHighwaterThresholdBytes);
 
-    ALOGV("readInternal offset %lld size %d", offset, size);
+    ALOGV("readInternal offset %lld size %zu", offset, size);
 
     Mutex::Autolock autoLock(mLock);
 
@@ -672,7 +719,7 @@ void NuCachedSource2::updateCacheParamsFromString(const char *s) {
         mKeepAliveIntervalUs = kDefaultKeepAliveIntervalUs;
     }
 
-    ALOGV("lowwater = %d bytes, highwater = %d bytes, keepalive = %lld us",
+    ALOGV("lowwater = %zu bytes, highwater = %zu bytes, keepalive = %" PRId64 " us",
          mLowwaterThresholdBytes,
          mHighwaterThresholdBytes,
          mKeepAliveIntervalUs);