Bug 1371699 - Use of NonBlockingAsyncInputStream in our code base, r=froydnj

dom/fetch/FetchStream.cpp

@@ -8,9 +8,8 @@
 #include "mozilla/dom/DOMException.h"
 #include "mozilla/dom/WorkerCommon.h"
 #include "mozilla/dom/WorkerPrivate.h"
-#include "nsITransport.h"
-#include "nsIStreamTransportService.h"
 #include "nsProxyRelease.h"
+#include "nsStreamUtils.h"
 
 #define FETCH_STREAM_FLAG 0
 
@@ -206,45 +205,15 @@ FetchStream::RequestDataCallback(JSContext* aCx,
     // mOriginalInputStream into an nsIAsyncInputStream.
     MOZ_ASSERT(stream->mOriginalInputStream);
 
-    bool nonBlocking = false;
-    nsresult rv = stream->mOriginalInputStream->IsNonBlocking(&nonBlocking);
+    nsCOMPtr<nsIAsyncInputStream> asyncStream;
+    nsresult rv =
+      NS_MakeAsyncNonBlockingInputStream(stream->mOriginalInputStream,
+                                         getter_AddRefs(asyncStream));
     if (NS_WARN_IF(NS_FAILED(rv))) {
       stream->ErrorPropagation(aCx, aStream, rv);
       return;
     }
 
-    nsCOMPtr<nsIAsyncInputStream> asyncStream =
-      do_QueryInterface(stream->mOriginalInputStream);
-    if (!nonBlocking || !asyncStream) {
-      nsCOMPtr<nsIStreamTransportService> sts =
-        do_GetService(kStreamTransportServiceCID, &rv);
-      if (NS_WARN_IF(NS_FAILED(rv))) {
-        stream->ErrorPropagation(aCx, aStream, rv);
-        return;
-      }
-
-      nsCOMPtr<nsITransport> transport;
-      rv = sts->CreateInputTransport(stream->mOriginalInputStream,
-                                     /* aCloseWhenDone */ true,
-                                     getter_AddRefs(transport));
-      if (NS_WARN_IF(NS_FAILED(rv))) {
-        stream->ErrorPropagation(aCx, aStream, rv);
-        return;
-      }
-
-      nsCOMPtr<nsIInputStream> wrapper;
-      rv = transport->OpenInputStream(/* aFlags */ 0,
-                                       /* aSegmentSize */ 0,
-                                       /* aSegmentCount */ 0,
-                                       getter_AddRefs(wrapper));
-      if (NS_WARN_IF(NS_FAILED(rv))) {
-        stream->ErrorPropagation(aCx, aStream, rv);
-        return;
-      }
-
-      asyncStream = do_QueryInterface(wrapper);
-    }
-
     stream->mInputStream = asyncStream;
     stream->mOriginalInputStream = nullptr;
   }

dom/file/FileReader.cpp

@@ -9,8 +9,6 @@
 #include "nsIEventTarget.h"
 #include "nsIGlobalObject.h"
 #include "nsITimer.h"
-#include "nsITransport.h"
-#include "nsIStreamTransportService.h"
 
 #include "mozilla/Base64.h"
 #include "mozilla/CheckedInt.h"
@@ -25,7 +23,6 @@
 #include "nsCycleCollectionParticipant.h"
 #include "nsDOMJSUtils.h"
 #include "nsError.h"
-#include "nsNetCID.h"
 #include "nsNetUtil.h"
 #include "xpcpublic.h"
 
@@ -41,8 +38,6 @@ namespace dom {
 
 const uint64_t kUnknownSize = uint64_t(-1);
 
-static NS_DEFINE_CID(kStreamTransportServiceCID, NS_STREAMTRANSPORTSERVICE_CID);
-
 NS_IMPL_CYCLE_COLLECTION_CLASS(FileReader)
 
 NS_IMPL_CYCLE_COLLECTION_TRAVERSE_BEGIN_INHERITED(FileReader,
@@ -416,44 +411,12 @@ FileReader::ReadFileContent(Blob& aBlob,
     return;
   }
 
-  bool nonBlocking = false;
-  aRv = stream->IsNonBlocking(&nonBlocking);
+  aRv = NS_MakeAsyncNonBlockingInputStream(stream,
+                                           getter_AddRefs(mAsyncStream));
   if (NS_WARN_IF(aRv.Failed())) {
     return;
   }
 
-  mAsyncStream = do_QueryInterface(stream);
-
-  // We want to have a non-blocking nsIAsyncInputStream.
-  if (!mAsyncStream || !nonBlocking) {
-    nsresult rv;
-    nsCOMPtr<nsIStreamTransportService> sts =
-      do_GetService(kStreamTransportServiceCID, &rv);
-    if (NS_WARN_IF(NS_FAILED(rv))) {
-      aRv.Throw(rv);
-      return;
-    }
-
-    nsCOMPtr<nsITransport> transport;
-    aRv = sts->CreateInputTransport(stream,
-                                    /* aCloseWhenDone */ true,
-                                    getter_AddRefs(transport));
-    if (NS_WARN_IF(aRv.Failed())) {
-      return;
-    }
-
-    nsCOMPtr<nsIInputStream> wrapper;
-    aRv = transport->OpenInputStream(/* aFlags */ 0,
-                                     /* aSegmentSize */ 0,
-                                     /* aSegmentCount */ 0,
-                                     getter_AddRefs(wrapper));
-    if (NS_WARN_IF(aRv.Failed())) {
-      return;
-    }
-
-    mAsyncStream = do_QueryInterface(wrapper);
-  }
-
   MOZ_ASSERT(mAsyncStream);
 
   mTotal = mBlob->GetSize(aRv);

dom/file/ipc/IPCBlobInputStream.cpp

@@ -9,6 +9,7 @@
 #include "IPCBlobInputStreamStorage.h"
 #include "mozilla/ipc/InputStreamParams.h"
 #include "mozilla/SlicedInputStream.h"
+#include "mozilla/NonBlockingAsyncInputStream.h"
 #include "IPCBlobInputStreamThread.h"
 #include "nsIAsyncInputStream.h"
 #include "nsIAsyncOutputStream.h"
@@ -645,7 +646,17 @@ IPCBlobInputStream::EnsureAsyncRemoteStream()
   }
 
   nsCOMPtr<nsIAsyncInputStream> asyncStream = do_QueryInterface(mRemoteStream);
-  if (!asyncStream || !nonBlocking) {
+
+  // If non-blocking and non-async, let's use NonBlockingAsyncInputStream.
+  if (nonBlocking && !asyncStream) {
+    rv = NonBlockingAsyncInputStream::Create(mRemoteStream,
+                                             getter_AddRefs(asyncStream));
+    if (NS_WARN_IF(NS_FAILED(rv))) {
+      return rv;
+    }
+  }
+
+  if (!asyncStream) {
     // Let's make the stream async using the DOMFile thread.
     nsCOMPtr<nsIAsyncInputStream> pipeIn;
     nsCOMPtr<nsIAsyncOutputStream> pipeOut;

netwerk/base/nsIStreamTransportService.idl

@@ -12,6 +12,9 @@ interface nsIInputAvailableCallback;
 /**
  * This service read/writes a stream on a background thread.
  *
+ * Note: instead of using this interface, probably you want to use
+ * NS_MakeAsyncNonBlockingInputStream.
+ *
  * Use this service to transform any blocking stream (e.g., file stream)
  * into a fully asynchronous stream that can be read/written without 
  * blocking the main thread.

netwerk/base/nsInputStreamPump.cpp

@@ -13,6 +13,8 @@
 #include "nsThreadUtils.h"
 #include "nsCOMPtr.h"
 #include "mozilla/Logging.h"
+#include "mozilla/NonBlockingAsyncInputStream.h"
+#include "mozilla/SlicedInputStream.h"
 #include "GeckoProfiler.h"
 #include "nsIStreamListener.h"
 #include "nsILoadGroup.h"
@@ -332,6 +334,11 @@ nsInputStreamPump::AsyncRead(nsIStreamListener *listener, nsISupports *ctxt)
 
     if (nonBlocking) {
         mAsyncStream = do_QueryInterface(mStream);
+        if (!mAsyncStream) {
+            rv = NonBlockingAsyncInputStream::Create(mStream,
+                                                     getter_AddRefs(mAsyncStream));
+            if (NS_WARN_IF(NS_FAILED(rv))) return rv;
+        }
     }
 
     if (!mAsyncStream) {

xpcom/io/nsStreamUtils.cpp

@@ -21,9 +21,14 @@
 #include "nsNetCID.h"
 #include "nsServiceManagerUtils.h"
 #include "nsThreadUtils.h"
+#include "nsITransport.h"
+#include "nsIStreamTransportService.h"
+#include "NonBlockingAsyncInputStream.h"
 
 using namespace mozilla;
 
+static NS_DEFINE_CID(kStreamTransportServiceCID, NS_STREAMTRANSPORTSERVICE_CID);
+
 //-----------------------------------------------------------------------------
 
 // This is a nsICancelableRunnable because we can dispatch it to Workers and
@@ -962,3 +967,60 @@ NS_CloneInputStream(nsIInputStream* aSource, nsIInputStream** aCloneOut,
 
   return NS_OK;
 }
+
+nsresult
+NS_MakeAsyncNonBlockingInputStream(nsIInputStream* aSource,
+                                   nsIAsyncInputStream** aAsyncInputStream)
+{
+  if (NS_WARN_IF(!aSource || !aAsyncInputStream)) {
+    return NS_ERROR_FAILURE;
+  }
+
+  bool nonBlocking = false;
+  nsresult rv = aSource->IsNonBlocking(&nonBlocking);
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    return rv;
+  }
+
+  nsCOMPtr<nsIAsyncInputStream> asyncStream = do_QueryInterface(aSource);
+
+  if (nonBlocking && asyncStream) {
+    // This stream is perfect!
+    asyncStream.forget(aAsyncInputStream);
+    return NS_OK;
+  }
+
+  if (nonBlocking) {
+    // If the stream is non-blocking but not async, we wrap it.
+    return NonBlockingAsyncInputStream::Create(aSource, aAsyncInputStream);
+  }
+
+  nsCOMPtr<nsIStreamTransportService> sts =
+    do_GetService(kStreamTransportServiceCID, &rv);
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    return rv;
+  }
+
+  nsCOMPtr<nsITransport> transport;
+  rv = sts->CreateInputTransport(aSource,
+                                 /* aCloseWhenDone */ true,
+                                 getter_AddRefs(transport));
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    return rv;
+  }
+
+  nsCOMPtr<nsIInputStream> wrapper;
+  rv = transport->OpenInputStream(/* aFlags */ 0,
+                                  /* aSegmentSize */ 0,
+                                  /* aSegmentCount */ 0,
+                                  getter_AddRefs(wrapper));
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    return rv;
+  }
+
+  asyncStream = do_QueryInterface(wrapper);
+  MOZ_ASSERT(asyncStream);
+
+  asyncStream.forget(aAsyncInputStream);
+  return NS_OK;
+}

xpcom/io/nsStreamUtils.h

@@ -12,6 +12,7 @@
 #include "nsIInputStream.h"
 #include "nsTArray.h"
 
+class nsIAsyncInputStream;
 class nsIOutputStream;
 class nsIInputStreamCallback;
 class nsIOutputStreamCallback;
@@ -293,4 +294,23 @@ extern nsresult
 NS_CloneInputStream(nsIInputStream* aSource, nsIInputStream** aCloneOut,
                     nsIInputStream** aReplacementOut = nullptr);
 
+/*
+ * This function returns a non-blocking nsIAsyncInputStream. Internally,
+ * different approaches are used based on what |aSource| is and what it
+ * implements.
+ *
+ * If the |aSource| is already a non-blocking and async stream,
+ * |aAsyncInputStream| will be equal to |aSource|.
+ *
+ * Otherwise, if |aSource| is just non-blocking, NonBlockingAsyncInputStream
+ * class is used in order to make it async.
+ *
+ * The last step is to use nsIStreamTransportService and create a pipe in order
+ * to expose a non-blocking async inputStream and read |aSource| data from
+ * a separate thread.
+ */
+extern nsresult
+NS_MakeAsyncNonBlockingInputStream(nsIInputStream* aSource,
+                                   nsIAsyncInputStream** aAsyncInputStream);
+
 #endif // !nsStreamUtils_h__