mirror of
https://repo.dactyloidae.xyz/Dactyloidae/UXP.git
synced 2026-09-06 15:58:39 +09:00
Issue #1442 - Part 8: Fetch implementation of streams. https://bugzilla.mozilla.org/show_bug.cgi?id=1128959 +worker-friendly pref checking for the DOM API.
This commit is contained in:
Brian Smith
roytam1
parent
dd2c9eb418
commit
bff1f3bc60
13 changed files with 620 additions and 24 deletions
336
dom/fetch/FetchStream.cpp
Normal file
336
dom/fetch/FetchStream.cpp
Normal file
|
|
@ -0,0 +1,336 @@
|
|||
/* -*- Mode: C; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
|
||||
/* 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 "nsNetCID.h"
|
||||
#include "nsITransport.h"
|
||||
#include "nsIStreamTransportService.h"
|
||||
#include "nsProxyRelease.h"
|
||||
|
||||
#include "mozilla/dom/DOMError.h"
|
||||
|
||||
#define FETCH_STREAM_FLAG 0
|
||||
|
||||
static NS_DEFINE_CID(kStreamTransportServiceCID,
|
||||
NS_STREAMTRANSPORTSERVICE_CID);
|
||||
|
||||
namespace mozilla {
|
||||
namespace dom {
|
||||
|
||||
NS_IMPL_ISUPPORTS(FetchStream, nsIInputStreamCallback)
|
||||
|
||||
/* static */ JSObject*
|
||||
FetchStream::Create(JSContext* aCx, nsIGlobalObject* aGlobal,
|
||||
nsIInputStream* aInputStream, ErrorResult& aRv)
|
||||
{
|
||||
MOZ_DIAGNOSTIC_ASSERT(aCx);
|
||||
MOZ_DIAGNOSTIC_ASSERT(aInputStream);
|
||||
|
||||
RefPtr<FetchStream> stream = new FetchStream(aGlobal, aInputStream);
|
||||
|
||||
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 nullptr;
|
||||
}
|
||||
|
||||
stream->mReadableStream = body;
|
||||
|
||||
// JS engine will call the finalize callback.
|
||||
NS_ADDREF(stream.get());
|
||||
return 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);
|
||||
|
||||
MOZ_DIAGNOSTIC_ASSERT(stream->mState == eWaiting ||
|
||||
stream->mState == eChecking);
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
bool nonBlocking = false;
|
||||
nsresult rv = stream->mOriginalInputStream->IsNonBlocking(&nonBlocking);
|
||||
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,
|
||||
/* aStartOffset */ 0,
|
||||
/* aReadLimit */ -1,
|
||||
/* 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;
|
||||
}
|
||||
|
||||
MOZ_DIAGNOSTIC_ASSERT(stream->mInputStream);
|
||||
MOZ_DIAGNOSTIC_ASSERT(!stream->mOriginalInputStream);
|
||||
|
||||
nsresult rv =
|
||||
stream->mInputStream->AsyncWait(stream, 0, 0, nullptr);
|
||||
if (NS_WARN_IF(NS_FAILED(rv))) {
|
||||
stream->ErrorPropagation(aCx, 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);
|
||||
|
||||
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, aStream, rv);
|
||||
return;
|
||||
}
|
||||
|
||||
*aByteWritten = written;
|
||||
|
||||
if (written == 0) {
|
||||
stream->mState = eClosed;
|
||||
JS::ReadableStreamClose(aCx, aStream);
|
||||
return;
|
||||
}
|
||||
|
||||
rv = stream->mInputStream->AsyncWait(stream, 0, 0, nullptr);
|
||||
if (NS_WARN_IF(NS_FAILED(rv))) {
|
||||
stream->ErrorPropagation(aCx, 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);
|
||||
|
||||
RefPtr<FetchStream> stream = static_cast<FetchStream*>(aUnderlyingSource);
|
||||
|
||||
if (stream->mInputStream) {
|
||||
stream->mInputStream->CloseWithStatus(NS_BASE_STREAM_CLOSED);
|
||||
}
|
||||
|
||||
stream->mState = eClosed;
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
void
|
||||
FetchStream::FinalizeCallback(void* aUnderlyingSource, uint8_t aFlags)
|
||||
{
|
||||
MOZ_DIAGNOSTIC_ASSERT(aUnderlyingSource);
|
||||
MOZ_DIAGNOSTIC_ASSERT(aFlags == FETCH_STREAM_FLAG);
|
||||
|
||||
RefPtr<FetchStream> stream =
|
||||
dont_AddRef(static_cast<FetchStream*>(aUnderlyingSource));
|
||||
|
||||
stream->mState = eClosed;
|
||||
stream->mReadableStream = nullptr;
|
||||
}
|
||||
|
||||
FetchStream::FetchStream(nsIGlobalObject* aGlobal,
|
||||
nsIInputStream* aInputStream)
|
||||
: mState(eWaiting)
|
||||
, mGlobal(aGlobal)
|
||||
, mOriginalInputStream(aInputStream)
|
||||
, mOwningEventTarget(nullptr)
|
||||
, mReadableStream(nullptr)
|
||||
{
|
||||
MOZ_DIAGNOSTIC_ASSERT(aInputStream);
|
||||
}
|
||||
|
||||
FetchStream::~FetchStream()
|
||||
{
|
||||
NS_ProxyRelease(mOwningEventTarget, mGlobal.forget());
|
||||
}
|
||||
|
||||
void
|
||||
FetchStream::ErrorPropagation(JSContext* aCx, JS::HandleObject aStream,
|
||||
nsresult aError)
|
||||
{
|
||||
// Nothing to do.
|
||||
if (mState == eClosed) {
|
||||
return;
|
||||
}
|
||||
|
||||
// We cannot continue with any other operation.
|
||||
mState = eClosed;
|
||||
|
||||
// Let's close the stream.
|
||||
if (aError == NS_BASE_STREAM_CLOSED) {
|
||||
JS::ReadableStreamClose(aCx, aStream);
|
||||
return;
|
||||
}
|
||||
|
||||
nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(mGlobal);
|
||||
|
||||
// Let's use a generic error.
|
||||
RefPtr<DOMError> error = new DOMError(window, NS_ERROR_DOM_TYPE_ERR);
|
||||
|
||||
JS::Rooted<JS::Value> errorValue(aCx);
|
||||
if (ToJSValue(aCx, error, &errorValue)) {
|
||||
JS::ReadableStreamError(aCx, aStream, errorValue);
|
||||
}
|
||||
}
|
||||
|
||||
NS_IMETHODIMP
|
||||
FetchStream::OnInputStreamReady(nsIAsyncInputStream* aStream)
|
||||
{
|
||||
MOZ_DIAGNOSTIC_ASSERT(aStream);
|
||||
|
||||
// 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, mReadableStream);
|
||||
|
||||
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, 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;
|
||||
JS::ReadableStreamUpdateDataAvailableFromSource(cx, stream, size);
|
||||
|
||||
// The WriteInto callback changes mState to eChecking.
|
||||
MOZ_DIAGNOSTIC_ASSERT(mState == eChecking);
|
||||
|
||||
return NS_OK;
|
||||
}
|
||||
|
||||
} // dom namespace
|
||||
} // mozilla namespace
|
||||
Loading…
Add table
Add a link
Reference in a new issue