gecko-dev/dom/fetch/FetchStream.cpp

584 строки
16 KiB
C++

/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=2 et sw=2 tw=80: */
/* 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 "FetchStream.h"
#include "mozilla/dom/DOMException.h"
#include "mozilla/dom/WorkerCommon.h"
#include "mozilla/dom/WorkerPrivate.h"
#include "nsProxyRelease.h"
#include "nsStreamUtils.h"
#define FETCH_STREAM_FLAG 0
static NS_DEFINE_CID(kStreamTransportServiceCID,
NS_STREAMTRANSPORTSERVICE_CID);
namespace mozilla {
namespace dom {
class FetchStream::WorkerShutdown final : public WorkerControlRunnable
{
public:
WorkerShutdown(WorkerPrivate* aWorkerPrivate, RefPtr<FetchStream> aStream)
: WorkerControlRunnable(aWorkerPrivate, WorkerThreadUnchangedBusyCount)
, mStream(aStream)
{}
bool
WorkerRun(JSContext* aCx, WorkerPrivate* aWorkerPrivate) override
{
mStream->ReleaseObjects();
return true;
}
// This runnable starts from a JS Thread. We need to disable a couple of
// assertions overring the following methods.
bool
PreDispatch(WorkerPrivate* aWorkerPrivate) override
{
return true;
}
void
PostDispatch(WorkerPrivate* aWorkerPrivate, bool aDispatchResult) override
{}
private:
RefPtr<FetchStream> mStream;
};
NS_IMPL_ISUPPORTS(FetchStream, nsIInputStreamCallback, nsIObserver,
nsISupportsWeakReference)
/* static */ void
FetchStream::Create(JSContext* aCx, FetchStreamHolder* aStreamHolder,
nsIGlobalObject* aGlobal, nsIInputStream* aInputStream,
JS::MutableHandle<JSObject*> aStream, ErrorResult& aRv)
{
MOZ_DIAGNOSTIC_ASSERT(aCx);
MOZ_DIAGNOSTIC_ASSERT(aStreamHolder);
MOZ_DIAGNOSTIC_ASSERT(aInputStream);
RefPtr<FetchStream> stream =
new FetchStream(aGlobal, aStreamHolder, aInputStream);
if (NS_IsMainThread()) {
nsCOMPtr<nsIObserverService> os = mozilla::services::GetObserverService();
if (NS_WARN_IF(!os)) {
aRv.Throw(NS_ERROR_FAILURE);
return;
}
aRv = os->AddObserver(stream, DOM_WINDOW_DESTROYED_TOPIC, true);
if (NS_WARN_IF(aRv.Failed())) {
return;
}
} else {
WorkerPrivate* workerPrivate = GetWorkerPrivateFromContext(aCx);
MOZ_ASSERT(workerPrivate);
RefPtr<WeakWorkerRef> workerRef =
WeakWorkerRef::Create(workerPrivate, [stream]() {
stream->Close();
});
if (NS_WARN_IF(!workerRef)) {
aRv.Throw(NS_ERROR_DOM_INVALID_STATE_ERR);
return;
}
// Note, this will create a ref-cycle between the holder and the stream.
// The cycle is broken when the stream is closed or the worker begins
// shutting down.
stream->mWorkerRef = workerRef.forget();
}
if (!JS::HasReadableStreamCallbacks(aCx)) {
JS::SetReadableStreamCallbacks(aCx,
&FetchStream::RequestDataCallback,
&FetchStream::WriteIntoReadRequestCallback,
&FetchStream::CancelCallback,
&FetchStream::ClosedCallback,
&FetchStream::ErroredCallback,
&FetchStream::FinalizeCallback);
}
JS::Rooted<JSObject*> body(aCx,
JS::NewReadableExternalSourceStreamObject(aCx, stream, FETCH_STREAM_FLAG));
if (!body) {
aRv.StealExceptionFromJSContext(aCx);
return;
}
// This will be released in FetchStream::FinalizeCallback(). We are
// guaranteed the jsapi will call FinalizeCallback when ReadableStream
// js object is finalized.
NS_ADDREF(stream.get());
aStream.set(body);
}
/* static */ void
FetchStream::RequestDataCallback(JSContext* aCx,
JS::HandleObject aStream,
void* aUnderlyingSource,
uint8_t aFlags,
size_t aDesiredSize)
{
MOZ_DIAGNOSTIC_ASSERT(aUnderlyingSource);
MOZ_DIAGNOSTIC_ASSERT(aFlags == FETCH_STREAM_FLAG);
MOZ_DIAGNOSTIC_ASSERT(JS::ReadableStreamIsDisturbed(aStream));
RefPtr<FetchStream> stream = static_cast<FetchStream*>(aUnderlyingSource);
stream->AssertIsOnOwningThread();
MutexAutoLock lock(stream->mMutex);
MOZ_DIAGNOSTIC_ASSERT(stream->mState == eInitializing ||
stream->mState == eWaiting ||
stream->mState == eChecking ||
stream->mState == eReading);
if (stream->mState == eReading) {
// We are already reading data.
return;
}
if (stream->mState == eChecking) {
// If we are looking for more data, there is nothing else we should do:
// let's move this checking operation in a reading.
MOZ_ASSERT(stream->mInputStream);
stream->mState = eReading;
return;
}
if (stream->mState == eInitializing) {
// The stream has been used for the first time.
stream->mStreamHolder->MarkAsRead();
}
stream->mState = eReading;
if (!stream->mInputStream) {
// This is the first use of the stream. Let's convert the
// mOriginalInputStream into an nsIAsyncInputStream.
MOZ_ASSERT(stream->mOriginalInputStream);
nsCOMPtr<nsIAsyncInputStream> asyncStream;
nsresult rv =
NS_MakeAsyncNonBlockingInputStream(stream->mOriginalInputStream.forget(),
getter_AddRefs(asyncStream));
if (NS_WARN_IF(NS_FAILED(rv))) {
stream->ErrorPropagation(aCx, lock, aStream, rv);
return;
}
stream->mInputStream = asyncStream;
stream->mOriginalInputStream = nullptr;
}
MOZ_DIAGNOSTIC_ASSERT(stream->mInputStream);
MOZ_DIAGNOSTIC_ASSERT(!stream->mOriginalInputStream);
nsresult rv =
stream->mInputStream->AsyncWait(stream, 0, 0,
stream->mOwningEventTarget);
if (NS_WARN_IF(NS_FAILED(rv))) {
stream->ErrorPropagation(aCx, lock, aStream, rv);
return;
}
// All good.
}
/* static */ void
FetchStream::WriteIntoReadRequestCallback(JSContext* aCx,
JS::HandleObject aStream,
void* aUnderlyingSource,
uint8_t aFlags, void* aBuffer,
size_t aLength, size_t* aByteWritten)
{
MOZ_DIAGNOSTIC_ASSERT(aUnderlyingSource);
MOZ_DIAGNOSTIC_ASSERT(aFlags == FETCH_STREAM_FLAG);
MOZ_DIAGNOSTIC_ASSERT(aBuffer);
MOZ_DIAGNOSTIC_ASSERT(aByteWritten);
RefPtr<FetchStream> stream = static_cast<FetchStream*>(aUnderlyingSource);
stream->AssertIsOnOwningThread();
MutexAutoLock lock(stream->mMutex);
MOZ_DIAGNOSTIC_ASSERT(stream->mInputStream);
MOZ_DIAGNOSTIC_ASSERT(stream->mState == eWriting);
stream->mState = eChecking;
uint32_t written;
nsresult rv =
stream->mInputStream->Read(static_cast<char*>(aBuffer), aLength, &written);
if (NS_WARN_IF(NS_FAILED(rv))) {
stream->ErrorPropagation(aCx, lock, aStream, rv);
return;
}
*aByteWritten = written;
if (written == 0) {
stream->CloseAndReleaseObjects(aCx, lock, aStream);
return;
}
rv = stream->mInputStream->AsyncWait(stream, 0, 0,
stream->mOwningEventTarget);
if (NS_WARN_IF(NS_FAILED(rv))) {
stream->ErrorPropagation(aCx, lock, aStream, rv);
return;
}
// All good.
}
/* static */ JS::Value
FetchStream::CancelCallback(JSContext* aCx, JS::HandleObject aStream,
void* aUnderlyingSource, uint8_t aFlags,
JS::HandleValue aReason)
{
MOZ_DIAGNOSTIC_ASSERT(aUnderlyingSource);
MOZ_DIAGNOSTIC_ASSERT(aFlags == FETCH_STREAM_FLAG);
// This is safe because we created an extra reference in FetchStream::Create()
// that won't be released until FetchStream::FinalizeCallback() is called.
// We are guaranteed that won't happen until the js ReadableStream object
// is finalized.
FetchStream* stream = static_cast<FetchStream*>(aUnderlyingSource);
stream->AssertIsOnOwningThread();
if (stream->mState == eInitializing) {
// The stream has been used for the first time.
stream->mStreamHolder->MarkAsRead();
}
if (stream->mInputStream) {
stream->mInputStream->CloseWithStatus(NS_BASE_STREAM_CLOSED);
}
// It could be that we don't have mInputStream yet, but we still have the
// original stream. We need to close that too.
if (stream->mOriginalInputStream) {
MOZ_ASSERT(!stream->mInputStream);
stream->mOriginalInputStream->Close();
}
stream->ReleaseObjects();
return JS::UndefinedValue();
}
/* static */ void
FetchStream::ClosedCallback(JSContext* aCx, JS::HandleObject aStream,
void* aUnderlyingSource, uint8_t aFlags)
{
MOZ_DIAGNOSTIC_ASSERT(aUnderlyingSource);
MOZ_DIAGNOSTIC_ASSERT(aFlags == FETCH_STREAM_FLAG);
}
/* static */ void
FetchStream::ErroredCallback(JSContext* aCx, JS::HandleObject aStream,
void* aUnderlyingSource, uint8_t aFlags,
JS::HandleValue aReason)
{
MOZ_DIAGNOSTIC_ASSERT(aUnderlyingSource);
MOZ_DIAGNOSTIC_ASSERT(aFlags == FETCH_STREAM_FLAG);
// This is safe because we created an extra reference in FetchStream::Create()
// that won't be released until FetchStream::FinalizeCallback() is called.
// We are guaranteed that won't happen until the js ReadableStream object
// is finalized.
FetchStream* stream = static_cast<FetchStream*>(aUnderlyingSource);
stream->AssertIsOnOwningThread();
if (stream->mState == eInitializing) {
// The stream has been used for the first time.
stream->mStreamHolder->MarkAsRead();
}
if (stream->mInputStream) {
stream->mInputStream->CloseWithStatus(NS_BASE_STREAM_CLOSED);
}
stream->ReleaseObjects();
}
void
FetchStream::FinalizeCallback(void* aUnderlyingSource, uint8_t aFlags)
{
MOZ_DIAGNOSTIC_ASSERT(aUnderlyingSource);
MOZ_DIAGNOSTIC_ASSERT(aFlags == FETCH_STREAM_FLAG);
// This can be called in any thread.
// This takes ownership of the ref created in FetchStream::Create().
RefPtr<FetchStream> stream =
dont_AddRef(static_cast<FetchStream*>(aUnderlyingSource));
stream->ReleaseObjects();
}
FetchStream::FetchStream(nsIGlobalObject* aGlobal,
FetchStreamHolder* aStreamHolder,
nsIInputStream* aInputStream)
: mMutex("FetchStream::mMutex")
, mState(eInitializing)
, mGlobal(aGlobal)
, mStreamHolder(aStreamHolder)
, mOwningEventTarget(aGlobal->EventTargetFor(TaskCategory::Other))
, mOriginalInputStream(aInputStream)
{
MOZ_DIAGNOSTIC_ASSERT(aInputStream);
MOZ_DIAGNOSTIC_ASSERT(aStreamHolder);
}
FetchStream::~FetchStream()
{
}
void
FetchStream::ErrorPropagation(JSContext* aCx,
const MutexAutoLock& aProofOfLock,
JS::HandleObject aStream,
nsresult aError)
{
AssertIsOnOwningThread();
// Nothing to do.
if (mState == eClosed) {
return;
}
// Let's close the stream.
if (aError == NS_BASE_STREAM_CLOSED) {
CloseAndReleaseObjects(aCx, aProofOfLock, aStream);
return;
}
// Let's use a generic error.
RefPtr<DOMException> error = DOMException::Create(NS_ERROR_DOM_TYPE_ERR);
JS::Rooted<JS::Value> errorValue(aCx);
if (ToJSValue(aCx, error, &errorValue)) {
MutexAutoUnlock unlock(mMutex);
JS::ReadableStreamError(aCx, aStream, errorValue);
}
ReleaseObjects(aProofOfLock);
}
NS_IMETHODIMP
FetchStream::OnInputStreamReady(nsIAsyncInputStream* aStream)
{
AssertIsOnOwningThread();
MOZ_DIAGNOSTIC_ASSERT(aStream);
MutexAutoLock lock(mMutex);
// Already closed. We have nothing else to do here.
if (mState == eClosed) {
return NS_OK;
}
MOZ_DIAGNOSTIC_ASSERT(mInputStream);
MOZ_DIAGNOSTIC_ASSERT(mState == eReading || mState == eChecking);
AutoJSAPI jsapi;
if (NS_WARN_IF(!jsapi.Init(mGlobal))) {
// Without JSContext we are not able to close the stream or to propagate the
// error.
return NS_ERROR_FAILURE;
}
JSContext* cx = jsapi.cx();
JS::Rooted<JSObject*> stream(cx, mStreamHolder->ReadableStreamBody());
uint64_t size = 0;
nsresult rv = mInputStream->Available(&size);
if (NS_SUCCEEDED(rv) && size == 0) {
// In theory this should not happen. If size is 0, the stream should be
// considered closed.
rv = NS_BASE_STREAM_CLOSED;
}
// No warning for stream closed.
if (rv == NS_BASE_STREAM_CLOSED || NS_WARN_IF(NS_FAILED(rv))) {
ErrorPropagation(cx, lock, stream, rv);
return NS_OK;
}
// This extra checking is completed. Let's wait for the next read request.
if (mState == eChecking) {
mState = eWaiting;
return NS_OK;
}
mState = eWriting;
{
MutexAutoUnlock unlock(mMutex);
JS::ReadableStreamUpdateDataAvailableFromSource(cx, stream, size);
}
// The WriteInto callback changes mState to eChecking.
MOZ_DIAGNOSTIC_ASSERT(mState == eChecking);
return NS_OK;
}
/* static */ nsresult
FetchStream::RetrieveInputStream(void* aUnderlyingReadableStreamSource,
nsIInputStream** aInputStream)
{
MOZ_ASSERT(aUnderlyingReadableStreamSource);
MOZ_ASSERT(aInputStream);
RefPtr<FetchStream> stream =
static_cast<FetchStream*>(aUnderlyingReadableStreamSource);
stream->AssertIsOnOwningThread();
// if mOriginalInputStream is null, the reading already started. We don't want
// to expose the internal inputStream.
if (NS_WARN_IF(!stream->mOriginalInputStream)) {
return NS_ERROR_DOM_INVALID_STATE_ERR;
}
nsCOMPtr<nsIInputStream> inputStream = stream->mOriginalInputStream;
inputStream.forget(aInputStream);
return NS_OK;
}
void
FetchStream::Close()
{
AssertIsOnOwningThread();
MutexAutoLock lock(mMutex);
if (mState == eClosed) {
return;
}
AutoJSAPI jsapi;
if (NS_WARN_IF(!jsapi.Init(mGlobal))) {
ReleaseObjects(lock);
return;
}
JSContext* cx = jsapi.cx();
JS::Rooted<JSObject*> stream(cx, mStreamHolder->ReadableStreamBody());
CloseAndReleaseObjects(cx, lock, stream);
}
void
FetchStream::CloseAndReleaseObjects(JSContext* aCx,
const MutexAutoLock& aProofOfLock,
JS::HandleObject aStream)
{
AssertIsOnOwningThread();
MOZ_DIAGNOSTIC_ASSERT(mState != eClosed);
ReleaseObjects(aProofOfLock);
MutexAutoUnlock unlock(mMutex);
if (JS::ReadableStreamIsReadable(aStream)) {
JS::ReadableStreamClose(aCx, aStream);
}
}
void
FetchStream::ReleaseObjects()
{
MutexAutoLock lock(mMutex);
ReleaseObjects(lock);
}
void
FetchStream::ReleaseObjects(const MutexAutoLock& aProofOfLock)
{
// This method can be called on 2 possible threads: the owning one and a JS
// thread used to release resources. If we are on the JS thread, we need to
// dispatch a runnable to go back to the owning thread in order to release
// resources correctly.
if (mState == eClosed) {
// Already gone. Nothing to do.
return;
}
mState = eClosed;
if (!NS_IsMainThread() && !IsCurrentThreadRunningWorker()) {
// Let's dispatch a WorkerControlRunnable if the owning thread is a worker.
if (mWorkerRef) {
RefPtr<WorkerShutdown> r =
new WorkerShutdown(mWorkerRef->GetUnsafePrivate(), this);
Unused << NS_WARN_IF(!r->Dispatch());
return;
}
// A normal runnable of the owning thread is the main-thread.
RefPtr<FetchStream> self = this;
RefPtr<Runnable> r =
NS_NewRunnableFunction("FetchStream::ReleaseObjects",
[self] () { self->ReleaseObjects(); });
mOwningEventTarget->Dispatch(r.forget());
return;
}
AssertIsOnOwningThread();
if (NS_IsMainThread()) {
nsCOMPtr<nsIObserverService> obs = mozilla::services::GetObserverService();
if (obs) {
obs->RemoveObserver(this, DOM_WINDOW_DESTROYED_TOPIC);
}
}
mWorkerRef = nullptr;
mGlobal = nullptr;
mStreamHolder->NullifyStream();
mStreamHolder = nullptr;
}
#ifdef DEBUG
void
FetchStream::AssertIsOnOwningThread()
{
NS_ASSERT_OWNINGTHREAD(FetchStream);
}
#endif
// nsIObserver
// -----------
NS_IMETHODIMP
FetchStream::Observe(nsISupports* aSubject, const char* aTopic,
const char16_t* aData)
{
AssertIsOnMainThread();
AssertIsOnOwningThread();
MOZ_ASSERT(strcmp(aTopic, DOM_WINDOW_DESTROYED_TOPIC) == 0);
nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(mGlobal);
if (SameCOMIdentity(aSubject, window)) {
Close();
}
return NS_OK;
}
} // dom namespace
} // mozilla namespace