summaryrefslogtreecommitdiffstats
path: root/netwerk/base/nsAsyncStreamCopier.cpp
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--netwerk/base/nsAsyncStreamCopier.cpp405
1 files changed, 405 insertions, 0 deletions
diff --git a/netwerk/base/nsAsyncStreamCopier.cpp b/netwerk/base/nsAsyncStreamCopier.cpp
new file mode 100644
index 0000000000..de8363d17d
--- /dev/null
+++ b/netwerk/base/nsAsyncStreamCopier.cpp
@@ -0,0 +1,405 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+#include "nsAsyncStreamCopier.h"
+#include "nsComponentManagerUtils.h"
+#include "nsIOService.h"
+#include "nsIEventTarget.h"
+#include "nsStreamUtils.h"
+#include "nsThreadUtils.h"
+#include "nsNetUtil.h"
+#include "nsNetCID.h"
+#include "nsIBufferedStreams.h"
+#include "nsIRequestObserver.h"
+#include "mozilla/Logging.h"
+
+using namespace mozilla;
+using namespace mozilla::net;
+
+#undef LOG
+//
+// MOZ_LOG=nsStreamCopier:5
+//
+static LazyLogModule gStreamCopierLog("nsStreamCopier");
+#define LOG(args) MOZ_LOG(gStreamCopierLog, mozilla::LogLevel::Debug, args)
+
+/**
+ * An event used to perform initialization off the main thread.
+ */
+class AsyncApplyBufferingPolicyEvent final : public Runnable {
+ public:
+ /**
+ * @param aCopier
+ * The nsAsyncStreamCopier requesting the information.
+ */
+ explicit AsyncApplyBufferingPolicyEvent(nsAsyncStreamCopier* aCopier)
+ : mozilla::Runnable("AsyncApplyBufferingPolicyEvent"),
+ mCopier(aCopier),
+ mTarget(GetCurrentSerialEventTarget()) {}
+
+ NS_IMETHOD Run() override {
+ nsresult rv = mCopier->ApplyBufferingPolicy();
+ if (NS_FAILED(rv)) {
+ mCopier->Cancel(rv);
+ return NS_OK;
+ }
+
+ rv = mTarget->Dispatch(
+ NewRunnableMethod("nsAsyncStreamCopier::AsyncCopyInternal", mCopier,
+ &nsAsyncStreamCopier::AsyncCopyInternal),
+ NS_DISPATCH_NORMAL);
+ MOZ_ASSERT(NS_SUCCEEDED(rv));
+
+ if (NS_FAILED(rv)) {
+ mCopier->Cancel(rv);
+ }
+ return NS_OK;
+ }
+
+ private:
+ RefPtr<nsAsyncStreamCopier> mCopier;
+ nsCOMPtr<nsIEventTarget> mTarget;
+};
+
+//-----------------------------------------------------------------------------
+
+nsAsyncStreamCopier::nsAsyncStreamCopier()
+ : mChunkSize(nsIOService::gDefaultSegmentSize) {
+ LOG(("Creating nsAsyncStreamCopier @%p\n", this));
+}
+
+nsAsyncStreamCopier::~nsAsyncStreamCopier() {
+ LOG(("Destroying nsAsyncStreamCopier @%p\n", this));
+}
+
+bool nsAsyncStreamCopier::IsComplete(nsresult* status) {
+ MutexAutoLock lock(mLock);
+ if (status) *status = mStatus;
+ return !mIsPending;
+}
+
+nsIRequest* nsAsyncStreamCopier::AsRequest() {
+ return static_cast<nsIRequest*>(static_cast<nsIAsyncStreamCopier*>(this));
+}
+
+void nsAsyncStreamCopier::Complete(nsresult status) {
+ LOG(("nsAsyncStreamCopier::Complete [this=%p status=%" PRIx32 "]\n", this,
+ static_cast<uint32_t>(status)));
+
+ nsCOMPtr<nsIRequestObserver> observer;
+ nsCOMPtr<nsISupports> ctx;
+ {
+ MutexAutoLock lock(mLock);
+ mCopierCtx = nullptr;
+
+ if (mIsPending) {
+ mIsPending = false;
+ mStatus = status;
+
+ // setup OnStopRequest callback and release references...
+ observer = mObserver;
+ mObserver = nullptr;
+ }
+ }
+
+ if (observer) {
+ LOG((" calling OnStopRequest [status=%" PRIx32 "]\n",
+ static_cast<uint32_t>(status)));
+ observer->OnStopRequest(AsRequest(), status);
+ }
+}
+
+void nsAsyncStreamCopier::OnAsyncCopyComplete(void* closure, nsresult status) {
+ // AddRef'd in AsyncCopy. Will be released at the end of the method.
+ RefPtr<nsAsyncStreamCopier> self = dont_AddRef((nsAsyncStreamCopier*)closure);
+ self->Complete(status);
+}
+
+//-----------------------------------------------------------------------------
+// nsISupports
+
+// We cannot use simply NS_IMPL_ISUPPORTSx as both
+// nsIAsyncStreamCopier and nsIAsyncStreamCopier2 implement nsIRequest
+
+NS_IMPL_ADDREF(nsAsyncStreamCopier)
+NS_IMPL_RELEASE(nsAsyncStreamCopier)
+NS_INTERFACE_TABLE_HEAD(nsAsyncStreamCopier)
+ NS_INTERFACE_TABLE_BEGIN
+ NS_INTERFACE_TABLE_ENTRY(nsAsyncStreamCopier, nsIAsyncStreamCopier)
+ NS_INTERFACE_TABLE_ENTRY(nsAsyncStreamCopier, nsIAsyncStreamCopier2)
+ NS_INTERFACE_TABLE_ENTRY_AMBIGUOUS(nsAsyncStreamCopier, nsIRequest,
+ nsIAsyncStreamCopier)
+ NS_INTERFACE_TABLE_ENTRY_AMBIGUOUS(nsAsyncStreamCopier, nsISupports,
+ nsIAsyncStreamCopier)
+ NS_INTERFACE_TABLE_END
+NS_INTERFACE_TABLE_TAIL
+
+//-----------------------------------------------------------------------------
+// nsIRequest
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::GetName(nsACString& name) {
+ name.Truncate();
+ return NS_OK;
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::IsPending(bool* result) {
+ *result = !IsComplete();
+ return NS_OK;
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::GetStatus(nsresult* status) {
+ IsComplete(status);
+ return NS_OK;
+}
+
+NS_IMETHODIMP nsAsyncStreamCopier::SetCanceledReason(
+ const nsACString& aReason) {
+ return nsIAsyncStreamCopier::SetCanceledReasonImpl(aReason);
+}
+
+NS_IMETHODIMP nsAsyncStreamCopier::GetCanceledReason(nsACString& aReason) {
+ return nsIAsyncStreamCopier::GetCanceledReasonImpl(aReason);
+}
+
+NS_IMETHODIMP nsAsyncStreamCopier::CancelWithReason(nsresult aStatus,
+ const nsACString& aReason) {
+ return nsIAsyncStreamCopier::CancelWithReasonImpl(aStatus, aReason);
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::Cancel(nsresult status) {
+ nsCOMPtr<nsISupports> copierCtx;
+ {
+ MutexAutoLock lock(mLock);
+ if (!mIsPending) {
+ return NS_OK;
+ }
+ copierCtx.swap(mCopierCtx);
+ }
+
+ if (NS_SUCCEEDED(status)) {
+ NS_WARNING("cancel with non-failure status code");
+ status = NS_BASE_STREAM_CLOSED;
+ }
+
+ if (copierCtx) NS_CancelAsyncCopy(copierCtx, status);
+
+ return NS_OK;
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::Suspend() {
+ MOZ_ASSERT_UNREACHABLE("nsAsyncStreamCopier::Suspend");
+ return NS_ERROR_NOT_IMPLEMENTED;
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::Resume() {
+ MOZ_ASSERT_UNREACHABLE("nsAsyncStreamCopier::Resume");
+ return NS_ERROR_NOT_IMPLEMENTED;
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::GetLoadFlags(nsLoadFlags* aLoadFlags) {
+ *aLoadFlags = LOAD_NORMAL;
+ return NS_OK;
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::SetLoadFlags(nsLoadFlags aLoadFlags) { return NS_OK; }
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::GetTRRMode(nsIRequest::TRRMode* aTRRMode) {
+ return nsIAsyncStreamCopier::GetTRRModeImpl(aTRRMode);
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::SetTRRMode(nsIRequest::TRRMode aTRRMode) {
+ return nsIAsyncStreamCopier::SetTRRModeImpl(aTRRMode);
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::GetLoadGroup(nsILoadGroup** aLoadGroup) {
+ *aLoadGroup = nullptr;
+ return NS_OK;
+}
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::SetLoadGroup(nsILoadGroup* aLoadGroup) { return NS_OK; }
+
+// Can't be accessed by multiple threads yet
+nsresult nsAsyncStreamCopier::InitInternal(
+ nsIInputStream* source, nsIOutputStream* sink, nsIEventTarget* target,
+ uint32_t chunkSize, bool closeSource,
+ bool closeSink) MOZ_NO_THREAD_SAFETY_ANALYSIS {
+ NS_ASSERTION(!mSource && !mSink, "Init() called more than once");
+ if (chunkSize == 0) {
+ chunkSize = nsIOService::gDefaultSegmentSize;
+ }
+ mChunkSize = chunkSize;
+
+ mSource = source;
+ mSink = sink;
+ mCloseSource = closeSource;
+ mCloseSink = closeSink;
+
+ if (target) {
+ mTarget = target;
+ } else {
+ nsresult rv;
+ mTarget = do_GetService(NS_STREAMTRANSPORTSERVICE_CONTRACTID, &rv);
+ if (NS_FAILED(rv)) {
+ return rv;
+ }
+ }
+
+ return NS_OK;
+}
+
+//-----------------------------------------------------------------------------
+// nsIAsyncStreamCopier
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::Init(nsIInputStream* source, nsIOutputStream* sink,
+ nsIEventTarget* target, bool sourceBuffered,
+ bool sinkBuffered, uint32_t chunkSize,
+ bool closeSource, bool closeSink) {
+ NS_ASSERTION(sourceBuffered || sinkBuffered,
+ "at least one stream must be buffered");
+ mMode = sourceBuffered ? NS_ASYNCCOPY_VIA_READSEGMENTS
+ : NS_ASYNCCOPY_VIA_WRITESEGMENTS;
+
+ return InitInternal(source, sink, target, chunkSize, closeSource, closeSink);
+}
+
+//-----------------------------------------------------------------------------
+// nsIAsyncStreamCopier2
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::Init(nsIInputStream* source, nsIOutputStream* sink,
+ nsIEventTarget* target, uint32_t chunkSize,
+ bool closeSource, bool closeSink) {
+ mShouldSniffBuffering = true;
+
+ return InitInternal(source, sink, target, chunkSize, closeSource, closeSink);
+}
+
+/**
+ * Detect whether the input or the output stream is buffered,
+ * bufferize one of them if neither is buffered.
+ */
+nsresult nsAsyncStreamCopier::ApplyBufferingPolicy() {
+ // This function causes I/O, it must not be executed on the main
+ // thread.
+ MOZ_ASSERT(!NS_IsMainThread());
+
+ if (NS_OutputStreamIsBuffered(mSink)) {
+ // Sink is buffered, no need to perform additional buffering
+ mMode = NS_ASYNCCOPY_VIA_WRITESEGMENTS;
+ return NS_OK;
+ }
+ if (NS_InputStreamIsBuffered(mSource)) {
+ // Source is buffered, no need to perform additional buffering
+ mMode = NS_ASYNCCOPY_VIA_READSEGMENTS;
+ return NS_OK;
+ }
+
+ // No buffering, let's buffer the sink
+ nsresult rv;
+ nsCOMPtr<nsIBufferedOutputStream> sink =
+ do_CreateInstance(NS_BUFFEREDOUTPUTSTREAM_CONTRACTID, &rv);
+ if (NS_FAILED(rv)) {
+ return rv;
+ }
+
+ rv = sink->Init(mSink, mChunkSize);
+ if (NS_FAILED(rv)) {
+ return rv;
+ }
+
+ mMode = NS_ASYNCCOPY_VIA_WRITESEGMENTS;
+ mSink = sink;
+ return NS_OK;
+}
+
+//-----------------------------------------------------------------------------
+// Both nsIAsyncStreamCopier and nsIAsyncStreamCopier2
+
+NS_IMETHODIMP
+nsAsyncStreamCopier::AsyncCopy(nsIRequestObserver* observer, nsISupports* ctx) {
+ LOG(("nsAsyncStreamCopier::AsyncCopy [this=%p observer=%p]\n", this,
+ observer));
+
+ NS_ASSERTION(mSource && mSink, "not initialized");
+ nsresult rv;
+
+ if (observer) {
+ // build proxy for observer events
+ rv = NS_NewRequestObserverProxy(getter_AddRefs(mObserver), observer, ctx);
+ if (NS_FAILED(rv)) return rv;
+ }
+
+ // from this point forward, AsyncCopy is going to return NS_OK. any errors
+ // will be reported via OnStopRequest.
+ {
+ MutexAutoLock lock(mLock);
+ mIsPending = true;
+ }
+
+ if (mObserver) {
+ rv = mObserver->OnStartRequest(AsRequest());
+ if (NS_FAILED(rv)) Cancel(rv);
+ }
+
+ if (!mShouldSniffBuffering) {
+ // No buffer sniffing required, let's proceed
+ AsyncCopyInternal();
+ return NS_OK;
+ }
+
+ if (NS_IsMainThread()) {
+ // Don't perform buffer sniffing on the main thread
+ nsCOMPtr<nsIRunnable> event = new AsyncApplyBufferingPolicyEvent(this);
+ rv = mTarget->Dispatch(event, NS_DISPATCH_NORMAL);
+ if (NS_FAILED(rv)) {
+ Cancel(rv);
+ }
+ return NS_OK;
+ }
+
+ // We're not going to block the main thread, so let's sniff here
+ rv = ApplyBufferingPolicy();
+ if (NS_FAILED(rv)) {
+ Cancel(rv);
+ }
+ AsyncCopyInternal();
+ return NS_OK;
+}
+
+// Launch async copy.
+// All errors are reported through the observer.
+void nsAsyncStreamCopier::AsyncCopyInternal() {
+ MOZ_ASSERT(mMode == NS_ASYNCCOPY_VIA_READSEGMENTS ||
+ mMode == NS_ASYNCCOPY_VIA_WRITESEGMENTS);
+
+ nsresult rv;
+ // We want to receive progress notifications; release happens in
+ // OnAsyncCopyComplete.
+ RefPtr<nsAsyncStreamCopier> self = this;
+ {
+ MutexAutoLock lock(mLock);
+ rv = NS_AsyncCopy(mSource, mSink, mTarget, mMode, mChunkSize,
+ OnAsyncCopyComplete, this, mCloseSource, mCloseSink,
+ getter_AddRefs(mCopierCtx));
+ }
+ if (NS_FAILED(rv)) {
+ Cancel(rv);
+ return; // release self
+ }
+
+ Unused << self.forget(); // Will be released in OnAsyncCopyComplete
+}