diff options
Diffstat (limited to 'xpcom/io/nsMultiplexInputStream.cpp')
| -rw-r--r-- | xpcom/io/nsMultiplexInputStream.cpp | 1557 |
1 files changed, 1557 insertions, 0 deletions
diff --git a/xpcom/io/nsMultiplexInputStream.cpp b/xpcom/io/nsMultiplexInputStream.cpp new file mode 100644 index 0000000000..bc8a67ed23 --- /dev/null +++ b/xpcom/io/nsMultiplexInputStream.cpp @@ -0,0 +1,1557 @@ +/* -*- 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/. */ + +/** + * The multiplex stream concatenates a list of input streams into a single + * stream. + */ + +#include "mozilla/Attributes.h" +#include "mozilla/CheckedInt.h" +#include "mozilla/MathAlgorithms.h" +#include "mozilla/Mutex.h" + +#include "base/basictypes.h" + +#include "nsMultiplexInputStream.h" +#include "nsIBufferedStreams.h" +#include "nsICloneableInputStream.h" +#include "nsIMultiplexInputStream.h" +#include "nsISeekableStream.h" +#include "nsCOMPtr.h" +#include "nsCOMArray.h" +#include "nsIClassInfoImpl.h" +#include "nsIIPCSerializableInputStream.h" +#include "mozilla/ipc/InputStreamUtils.h" +#include "nsIAsyncInputStream.h" +#include "nsIInputStreamLength.h" +#include "nsNetUtil.h" +#include "nsStreamUtils.h" + +using namespace mozilla; +using namespace mozilla::ipc; + +using mozilla::DeprecatedAbs; + +class nsMultiplexInputStream final : public nsIMultiplexInputStream, + public nsISeekableStream, + public nsIIPCSerializableInputStream, + public nsICloneableInputStream, + public nsIAsyncInputStream, + public nsIInputStreamCallback, + public nsIInputStreamLength, + public nsIAsyncInputStreamLength { + public: + nsMultiplexInputStream(); + + NS_DECL_THREADSAFE_ISUPPORTS + NS_DECL_NSIINPUTSTREAM + NS_DECL_NSIMULTIPLEXINPUTSTREAM + NS_DECL_NSISEEKABLESTREAM + NS_DECL_NSITELLABLESTREAM + NS_DECL_NSIIPCSERIALIZABLEINPUTSTREAM + NS_DECL_NSICLONEABLEINPUTSTREAM + NS_DECL_NSIASYNCINPUTSTREAM + NS_DECL_NSIINPUTSTREAMCALLBACK + NS_DECL_NSIINPUTSTREAMLENGTH + NS_DECL_NSIASYNCINPUTSTREAMLENGTH + + // This is used for nsIAsyncInputStream::AsyncWait + void AsyncWaitCompleted(); + + // This is used for nsIAsyncInputStreamLength::AsyncLengthWait + void AsyncWaitCompleted(int64_t aLength, const MutexAutoLock& aProofOfLock) + MOZ_REQUIRES(mLock); + + struct StreamData { + nsresult Initialize(nsIInputStream* aOriginalStream) { + mCurrentPos = 0; + + mOriginalStream = aOriginalStream; + + mBufferedStream = aOriginalStream; + if (!NS_InputStreamIsBuffered(mBufferedStream)) { + nsCOMPtr<nsIInputStream> bufferedStream; + nsresult rv = NS_NewBufferedInputStream(getter_AddRefs(bufferedStream), + mBufferedStream.forget(), 4096); + NS_ENSURE_SUCCESS(rv, rv); + mBufferedStream = bufferedStream; + } + + mAsyncStream = do_QueryInterface(mBufferedStream); + mSeekableStream = do_QueryInterface(mBufferedStream); + + return NS_OK; + } + + nsCOMPtr<nsIInputStream> mOriginalStream; + + // Equal to mOriginalStream or a wrap around the original stream to make it + // buffered. + nsCOMPtr<nsIInputStream> mBufferedStream; + + // This can be null. + nsCOMPtr<nsIAsyncInputStream> mAsyncStream; + // This can be null. + nsCOMPtr<nsISeekableStream> mSeekableStream; + + uint64_t mCurrentPos; + }; + + Mutex& GetLock() MOZ_RETURN_CAPABILITY(mLock) { return mLock; } + + private: + ~nsMultiplexInputStream() = default; + + void NextStream() MOZ_REQUIRES(mLock) { + ++mCurrentStream; + mStartedReadingCurrent = false; + } + + nsresult AsyncWaitInternal(); + + // This method updates mSeekableStreams, mTellableStreams, + // mIPCSerializableStreams and mCloneableStreams values. + void UpdateQIMap(StreamData& aStream) MOZ_REQUIRES(mLock); + + struct MOZ_STACK_CLASS ReadSegmentsState { + nsCOMPtr<nsIInputStream> mThisStream; + uint32_t mOffset; + nsWriteSegmentFun mWriter; + void* mClosure; + bool mDone; + }; + + void SerializedComplexityInternal(uint32_t aMaxSize, uint32_t* aSizeUsed, + uint32_t* aPipes, uint32_t* aTransferables, + bool* aSerializeAsPipe); + + static nsresult ReadSegCb(nsIInputStream* aIn, void* aClosure, + const char* aFromRawSegment, uint32_t aToOffset, + uint32_t aCount, uint32_t* aWriteCount); + + bool IsSeekable() const; + bool IsIPCSerializable() const; + bool IsCloneable() const; + bool IsAsyncInputStream() const; + bool IsInputStreamLength() const; + bool IsAsyncInputStreamLength() const; + + Mutex mLock; // Protects access to all data members. + + nsTArray<StreamData> mStreams MOZ_GUARDED_BY(mLock); + + uint32_t mCurrentStream MOZ_GUARDED_BY(mLock); + bool mStartedReadingCurrent MOZ_GUARDED_BY(mLock); + nsresult mStatus MOZ_GUARDED_BY(mLock); + nsCOMPtr<nsIInputStreamCallback> mAsyncWaitCallback MOZ_GUARDED_BY(mLock); + uint32_t mAsyncWaitFlags MOZ_GUARDED_BY(mLock); + uint32_t mAsyncWaitRequestedCount MOZ_GUARDED_BY(mLock); + nsCOMPtr<nsIEventTarget> mAsyncWaitEventTarget MOZ_GUARDED_BY(mLock); + nsCOMPtr<nsIInputStreamLengthCallback> mAsyncWaitLengthCallback + MOZ_GUARDED_BY(mLock); + + class AsyncWaitLengthHelper; + RefPtr<AsyncWaitLengthHelper> mAsyncWaitLengthHelper MOZ_GUARDED_BY(mLock); + + uint32_t mSeekableStreams MOZ_GUARDED_BY(mLock); + uint32_t mIPCSerializableStreams MOZ_GUARDED_BY(mLock); + uint32_t mCloneableStreams MOZ_GUARDED_BY(mLock); + + // These are Atomics so that we can check them in QueryInterface without + // taking a lock (to look at mStreams.Length() and the numbers above) + // With no streams added yet, all of these are possible + Atomic<bool, Relaxed> mIsSeekableStream{true}; + Atomic<bool, Relaxed> mIsIPCSerializableStream{true}; + Atomic<bool, Relaxed> mIsCloneableStream{true}; + + Atomic<bool, Relaxed> mIsAsyncInputStream{false}; + Atomic<bool, Relaxed> mIsInputStreamLength{false}; + Atomic<bool, Relaxed> mIsAsyncInputStreamLength{false}; +}; + +NS_IMPL_ADDREF(nsMultiplexInputStream) +NS_IMPL_RELEASE(nsMultiplexInputStream) + +NS_IMPL_CLASSINFO(nsMultiplexInputStream, nullptr, nsIClassInfo::THREADSAFE, + NS_MULTIPLEXINPUTSTREAM_CID) + +NS_INTERFACE_MAP_BEGIN(nsMultiplexInputStream) + NS_INTERFACE_MAP_ENTRY(nsIMultiplexInputStream) + NS_INTERFACE_MAP_ENTRY(nsIInputStream) + NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsISeekableStream, IsSeekable()) + NS_INTERFACE_MAP_ENTRY(nsITellableStream) + NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsIIPCSerializableInputStream, + IsIPCSerializable()) + NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsICloneableInputStream, IsCloneable()) + NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsIAsyncInputStream, IsAsyncInputStream()) + NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsIInputStreamCallback, + IsAsyncInputStream()) + NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsIInputStreamLength, + IsInputStreamLength()) + NS_INTERFACE_MAP_ENTRY_CONDITIONAL(nsIAsyncInputStreamLength, + IsAsyncInputStreamLength()) + NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsIMultiplexInputStream) + NS_IMPL_QUERY_CLASSINFO(nsMultiplexInputStream) +NS_INTERFACE_MAP_END + +NS_IMPL_CI_INTERFACE_GETTER(nsMultiplexInputStream, nsIMultiplexInputStream, + nsIInputStream, nsISeekableStream, + nsITellableStream) + +static nsresult AvailableMaybeSeek(nsMultiplexInputStream::StreamData& aStream, + uint64_t* aResult) { + nsresult rv = aStream.mBufferedStream->Available(aResult); + if (rv == NS_BASE_STREAM_CLOSED) { + // Blindly seek to the current position if Available() returns + // NS_BASE_STREAM_CLOSED. + // If nsIFileInputStream is closed in Read() due to CLOSE_ON_EOF flag, + // Seek() could reopen the file if REOPEN_ON_REWIND flag is set. + if (aStream.mSeekableStream) { + nsresult rvSeek = + aStream.mSeekableStream->Seek(nsISeekableStream::NS_SEEK_CUR, 0); + if (NS_SUCCEEDED(rvSeek)) { + rv = aStream.mBufferedStream->Available(aResult); + } + } + } + return rv; +} + +nsMultiplexInputStream::nsMultiplexInputStream() + : mLock("nsMultiplexInputStream lock"), + mCurrentStream(0), + mStartedReadingCurrent(false), + mStatus(NS_OK), + mAsyncWaitFlags(0), + mAsyncWaitRequestedCount(0), + mSeekableStreams(0), + mIPCSerializableStreams(0), + mCloneableStreams(0) {} + +NS_IMETHODIMP +nsMultiplexInputStream::GetCount(uint32_t* aCount) { + MutexAutoLock lock(mLock); + *aCount = mStreams.Length(); + return NS_OK; +} + +NS_IMETHODIMP +nsMultiplexInputStream::AppendStream(nsIInputStream* aStream) { + MutexAutoLock lock(mLock); + + StreamData* streamData = mStreams.AppendElement(fallible); + if (NS_WARN_IF(!streamData)) { + return NS_ERROR_OUT_OF_MEMORY; + } + + nsresult rv = streamData->Initialize(aStream); + NS_ENSURE_SUCCESS(rv, rv); + + UpdateQIMap(*streamData); + + if (mStatus == NS_BASE_STREAM_CLOSED) { + // We were closed, but now we have more data to read. + mStatus = NS_OK; + } + + return NS_OK; +} + +NS_IMETHODIMP +nsMultiplexInputStream::GetStream(uint32_t aIndex, nsIInputStream** aResult) { + MutexAutoLock lock(mLock); + + if (aIndex >= mStreams.Length()) { + return NS_ERROR_NOT_AVAILABLE; + } + + StreamData& streamData = mStreams.ElementAt(aIndex); + nsCOMPtr<nsIInputStream> stream = streamData.mOriginalStream; + stream.forget(aResult); + return NS_OK; +} + +NS_IMETHODIMP +nsMultiplexInputStream::Close() { + nsTArray<nsCOMPtr<nsIInputStream>> streams; + + // Let's take a copy of the streams becuase, calling close() it could trigger + // a nsIInputStreamCallback immediately and we don't want to create a deadlock + // with mutex. + { + MutexAutoLock lock(mLock); + uint32_t len = mStreams.Length(); + for (uint32_t i = 0; i < len; ++i) { + if (NS_WARN_IF( + !streams.AppendElement(mStreams[i].mBufferedStream, fallible))) { + mStatus = NS_BASE_STREAM_CLOSED; + return NS_ERROR_OUT_OF_MEMORY; + } + } + mStatus = NS_BASE_STREAM_CLOSED; + } + + nsresult rv = NS_OK; + + uint32_t len = streams.Length(); + for (uint32_t i = 0; i < len; ++i) { + nsresult rv2 = streams[i]->Close(); + // We still want to close all streams, but we should return an error + if (NS_FAILED(rv2)) { + rv = rv2; + } + } + + return rv; +} + +NS_IMETHODIMP +nsMultiplexInputStream::Available(uint64_t* aResult) { + *aResult = 0; + + MutexAutoLock lock(mLock); + if (NS_FAILED(mStatus)) { + return mStatus; + } + + uint64_t avail = 0; + nsresult rv = NS_BASE_STREAM_CLOSED; + + uint32_t len = mStreams.Length(); + for (uint32_t i = mCurrentStream; i < len; i++) { + uint64_t streamAvail; + rv = AvailableMaybeSeek(mStreams[i], &streamAvail); + if (rv == NS_BASE_STREAM_CLOSED) { + // If a stream is closed, we continue with the next one. + // If this is the current stream we move to the following stream. + if (mCurrentStream == i) { + NextStream(); + } + + // If this is the last stream, we want to return this error code. + continue; + } + + if (NS_WARN_IF(NS_FAILED(rv))) { + mStatus = rv; + return mStatus; + } + + // If the current stream is async, we have to return what we have so far + // without processing the following streams. This is needed because + // ::Available should return only what is currently available. In case of an + // nsIAsyncInputStream, we have to call AsyncWait() in order to read more. + if (mStreams[i].mAsyncStream) { + avail += streamAvail; + break; + } + + if (streamAvail == 0) { + // Nothing to read for this stream. Let's move to the next one. + continue; + } + + avail += streamAvail; + } + + // We still have something to read. We don't want to return an error code yet. + if (avail) { + *aResult = avail; + return NS_OK; + } + + // Let's propagate the last error message. + mStatus = rv; + return rv; +} + +NS_IMETHODIMP +nsMultiplexInputStream::StreamStatus() { + MutexAutoLock lock(mLock); + return mStatus; +} + +NS_IMETHODIMP +nsMultiplexInputStream::Read(char* aBuf, uint32_t aCount, uint32_t* aResult) { + MutexAutoLock lock(mLock); + // It is tempting to implement this method in terms of ReadSegments, but + // that would prevent this class from being used with streams that only + // implement Read (e.g., file streams). + + *aResult = 0; + + if (mStatus == NS_BASE_STREAM_CLOSED) { + return NS_OK; + } + if (NS_FAILED(mStatus)) { + return mStatus; + } + + nsresult rv = NS_OK; + + uint32_t len = mStreams.Length(); + while (mCurrentStream < len && aCount) { + uint32_t read; + rv = mStreams[mCurrentStream].mBufferedStream->Read(aBuf, aCount, &read); + + // XXX some streams return NS_BASE_STREAM_CLOSED to indicate EOF. + // (This is a bug in those stream implementations) + if (rv == NS_BASE_STREAM_CLOSED) { + MOZ_ASSERT_UNREACHABLE( + "Input stream's Read method returned " + "NS_BASE_STREAM_CLOSED"); + rv = NS_OK; + read = 0; + } else if (NS_FAILED(rv)) { + break; + } + + if (read == 0) { + NextStream(); + } else { + NS_ASSERTION(aCount >= read, "Read more than requested"); + *aResult += read; + aCount -= read; + aBuf += read; + mStartedReadingCurrent = true; + + mStreams[mCurrentStream].mCurrentPos += read; + } + } + return *aResult ? NS_OK : rv; +} + +NS_IMETHODIMP +nsMultiplexInputStream::ReadSegments(nsWriteSegmentFun aWriter, void* aClosure, + uint32_t aCount, uint32_t* aResult) { + MutexAutoLock lock(mLock); + + if (mStatus == NS_BASE_STREAM_CLOSED) { + *aResult = 0; + return NS_OK; + } + if (NS_FAILED(mStatus)) { + return mStatus; + } + + NS_ASSERTION(aWriter, "missing aWriter"); + + nsresult rv = NS_OK; + ReadSegmentsState state; + state.mThisStream = this; + state.mOffset = 0; + state.mWriter = aWriter; + state.mClosure = aClosure; + state.mDone = false; + + uint32_t len = mStreams.Length(); + while (mCurrentStream < len && aCount) { + uint32_t read; + rv = mStreams[mCurrentStream].mBufferedStream->ReadSegments( + ReadSegCb, &state, aCount, &read); + + // XXX some streams return NS_BASE_STREAM_CLOSED to indicate EOF. + // (This is a bug in those stream implementations) + if (rv == NS_BASE_STREAM_CLOSED) { + MOZ_ASSERT_UNREACHABLE( + "Input stream's Read method returned " + "NS_BASE_STREAM_CLOSED"); + rv = NS_OK; + read = 0; + } + + // if |aWriter| decided to stop reading segments... + if (state.mDone || NS_FAILED(rv)) { + break; + } + + // if stream is empty, then advance to the next stream. + if (read == 0) { + NextStream(); + } else { + NS_ASSERTION(aCount >= read, "Read more than requested"); + state.mOffset += read; + aCount -= read; + mStartedReadingCurrent = true; + + mStreams[mCurrentStream].mCurrentPos += read; + } + } + + // if we successfully read some data, then this call succeeded. + *aResult = state.mOffset; + return state.mOffset ? NS_OK : rv; +} + +nsresult nsMultiplexInputStream::ReadSegCb(nsIInputStream* aIn, void* aClosure, + const char* aFromRawSegment, + uint32_t aToOffset, uint32_t aCount, + uint32_t* aWriteCount) { + nsresult rv; + ReadSegmentsState* state = (ReadSegmentsState*)aClosure; + rv = (state->mWriter)(state->mThisStream, state->mClosure, aFromRawSegment, + aToOffset + state->mOffset, aCount, aWriteCount); + if (NS_FAILED(rv)) { + state->mDone = true; + } + return rv; +} + +NS_IMETHODIMP +nsMultiplexInputStream::IsNonBlocking(bool* aNonBlocking) { + MutexAutoLock lock(mLock); + + uint32_t len = mStreams.Length(); + if (len == 0) { + // Claim to be non-blocking, since we won't block the caller. + *aNonBlocking = true; + return NS_OK; + } + + for (uint32_t i = 0; i < len; ++i) { + nsresult rv = mStreams[i].mBufferedStream->IsNonBlocking(aNonBlocking); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + // If one is blocking the entire stream becomes blocking. + if (!*aNonBlocking) { + return NS_OK; + } + } + + return NS_OK; +} + +NS_IMETHODIMP +nsMultiplexInputStream::Seek(int32_t aWhence, int64_t aOffset) { + MutexAutoLock lock(mLock); + + if (NS_FAILED(mStatus)) { + return mStatus; + } + + nsresult rv; + + uint32_t oldCurrentStream = mCurrentStream; + bool oldStartedReadingCurrent = mStartedReadingCurrent; + + if (aWhence == NS_SEEK_SET) { + int64_t remaining = aOffset; + if (aOffset == 0) { + mCurrentStream = 0; + } + for (uint32_t i = 0; i < mStreams.Length(); ++i) { + nsCOMPtr<nsISeekableStream> stream = mStreams[i].mSeekableStream; + if (!stream) { + return NS_ERROR_FAILURE; + } + + // See if all remaining streams should be rewound + if (remaining == 0) { + if (i < oldCurrentStream || + (i == oldCurrentStream && oldStartedReadingCurrent)) { + rv = stream->Seek(NS_SEEK_SET, 0); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + mStreams[i].mCurrentPos = 0; + continue; + } else { + break; + } + } + + // Get position in the current stream + int64_t streamPos; + if (i > oldCurrentStream || + (i == oldCurrentStream && !oldStartedReadingCurrent)) { + streamPos = 0; + } else { + streamPos = mStreams[i].mCurrentPos; + } + + // See if we need to seek the current stream forward or backward + if (remaining < streamPos) { + rv = stream->Seek(NS_SEEK_SET, remaining); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + mStreams[i].mCurrentPos = remaining; + mCurrentStream = i; + mStartedReadingCurrent = remaining != 0; + + remaining = 0; + } else if (remaining > streamPos) { + if (i < oldCurrentStream) { + // We're already at end so no need to seek this stream + remaining -= streamPos; + NS_ASSERTION(remaining >= 0, "Remaining invalid"); + } else { + uint64_t avail; + rv = AvailableMaybeSeek(mStreams[i], &avail); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + int64_t newPos = XPCOM_MIN(remaining, streamPos + (int64_t)avail); + + rv = stream->Seek(NS_SEEK_SET, newPos); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + mStreams[i].mCurrentPos = newPos; + mCurrentStream = i; + mStartedReadingCurrent = true; + + remaining -= newPos; + NS_ASSERTION(remaining >= 0, "Remaining invalid"); + } + } else { + NS_ASSERTION(remaining == streamPos, "Huh?"); + MOZ_ASSERT(remaining != 0, "Zero remaining should be handled earlier"); + remaining = 0; + mCurrentStream = i; + mStartedReadingCurrent = true; + } + } + + return NS_OK; + } + + if (aWhence == NS_SEEK_CUR && aOffset > 0) { + int64_t remaining = aOffset; + for (uint32_t i = mCurrentStream; remaining && i < mStreams.Length(); ++i) { + uint64_t avail; + rv = AvailableMaybeSeek(mStreams[i], &avail); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + int64_t seek = XPCOM_MIN((int64_t)avail, remaining); + + rv = mStreams[i].mSeekableStream->Seek(NS_SEEK_CUR, seek); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + mStreams[i].mCurrentPos += seek; + mCurrentStream = i; + mStartedReadingCurrent = true; + + remaining -= seek; + } + + return NS_OK; + } + + if (aWhence == NS_SEEK_CUR && aOffset < 0) { + int64_t remaining = -aOffset; + for (uint32_t i = mCurrentStream; remaining && i != (uint32_t)-1; --i) { + int64_t pos = mStreams[i].mCurrentPos; + + int64_t seek = XPCOM_MIN(pos, remaining); + + rv = mStreams[i].mSeekableStream->Seek(NS_SEEK_CUR, -seek); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + mStreams[i].mCurrentPos -= seek; + mCurrentStream = i; + mStartedReadingCurrent = seek != -pos; + + remaining -= seek; + } + + return NS_OK; + } + + if (aWhence == NS_SEEK_CUR) { + NS_ASSERTION(aOffset == 0, "Should have handled all non-zero values"); + + return NS_OK; + } + + if (aWhence == NS_SEEK_END) { + if (aOffset > 0) { + return NS_ERROR_INVALID_ARG; + } + + int64_t remaining = aOffset; + int32_t i; + for (i = mStreams.Length() - 1; i >= 0; --i) { + nsCOMPtr<nsISeekableStream> stream = mStreams[i].mSeekableStream; + + uint64_t avail; + rv = AvailableMaybeSeek(mStreams[i], &avail); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + int64_t streamLength = avail + mStreams[i].mCurrentPos; + + // The seek(END) can be completed in the current stream. + if (streamLength >= DeprecatedAbs(remaining)) { + rv = stream->Seek(NS_SEEK_END, remaining); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + mStreams[i].mCurrentPos = streamLength + remaining; + mCurrentStream = i; + mStartedReadingCurrent = true; + break; + } + + // We are at the beginning of this stream. + rv = stream->Seek(NS_SEEK_SET, 0); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + remaining += streamLength; + mStreams[i].mCurrentPos = 0; + } + + // Any other stream must be set to the end. + for (--i; i >= 0; --i) { + nsCOMPtr<nsISeekableStream> stream = mStreams[i].mSeekableStream; + + uint64_t avail; + rv = AvailableMaybeSeek(mStreams[i], &avail); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + int64_t streamLength = avail + mStreams[i].mCurrentPos; + + rv = stream->Seek(NS_SEEK_END, 0); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + mStreams[i].mCurrentPos = streamLength; + } + + return NS_OK; + } + + // other Seeks not implemented yet + return NS_ERROR_NOT_IMPLEMENTED; +} + +NS_IMETHODIMP +nsMultiplexInputStream::Tell(int64_t* aResult) { + MutexAutoLock lock(mLock); + + if (NS_FAILED(mStatus)) { + return mStatus; + } + + int64_t ret64 = 0; +#ifdef DEBUG + bool zeroFound = false; +#endif + + for (uint32_t i = 0; i < mStreams.Length(); ++i) { + ret64 += mStreams[i].mCurrentPos; + +#ifdef DEBUG + // When we see 1 stream with currentPos = 0, all the remaining streams must + // be set to 0 as well. + MOZ_ASSERT_IF(zeroFound, mStreams[i].mCurrentPos == 0); + if (mStreams[i].mCurrentPos == 0) { + zeroFound = true; + } +#endif + } + *aResult = ret64; + + return NS_OK; +} + +NS_IMETHODIMP +nsMultiplexInputStream::SetEOF() { return NS_ERROR_NOT_IMPLEMENTED; } + +NS_IMETHODIMP +nsMultiplexInputStream::CloseWithStatus(nsresult aStatus) { return Close(); } + +// This class is used to inform nsMultiplexInputStream that it's time to execute +// the asyncWait callback. +class AsyncWaitRunnable final : public DiscardableRunnable { + RefPtr<nsMultiplexInputStream> mStream; + + public: + static void Create(nsMultiplexInputStream* aStream, + nsIEventTarget* aEventTarget) { + RefPtr<AsyncWaitRunnable> runnable = new AsyncWaitRunnable(aStream); + if (aEventTarget) { + aEventTarget->Dispatch(runnable.forget(), NS_DISPATCH_NORMAL); + } else { + runnable->Run(); + } + } + + NS_IMETHOD + Run() override { + mStream->AsyncWaitCompleted(); + return NS_OK; + } + + private: + explicit AsyncWaitRunnable(nsMultiplexInputStream* aStream) + : DiscardableRunnable("AsyncWaitRunnable"), mStream(aStream) { + MOZ_ASSERT(aStream); + } +}; + +NS_IMETHODIMP +nsMultiplexInputStream::AsyncWait(nsIInputStreamCallback* aCallback, + uint32_t aFlags, uint32_t aRequestedCount, + nsIEventTarget* aEventTarget) { + { + MutexAutoLock lock(mLock); + + // We must execute the callback also when the stream is closed. + if (NS_FAILED(mStatus) && mStatus != NS_BASE_STREAM_CLOSED) { + return mStatus; + } + + if (NS_WARN_IF(mAsyncWaitCallback && aCallback && + mAsyncWaitCallback != aCallback)) { + return NS_ERROR_FAILURE; + } + + mAsyncWaitCallback = aCallback; + mAsyncWaitFlags = aFlags; + mAsyncWaitRequestedCount = aRequestedCount; + mAsyncWaitEventTarget = aEventTarget; + } + + return AsyncWaitInternal(); +} + +nsresult nsMultiplexInputStream::AsyncWaitInternal() { + nsCOMPtr<nsIAsyncInputStream> stream; + nsIInputStreamCallback* asyncWaitCallback = nullptr; + uint32_t asyncWaitFlags = 0; + uint32_t asyncWaitRequestedCount = 0; + nsCOMPtr<nsIEventTarget> asyncWaitEventTarget; + + { + MutexAutoLock lock(mLock); + + // Let's take the first async stream if we are not already closed, and if + // it has data to read or if it async. + if (mStatus != NS_BASE_STREAM_CLOSED) { + for (; mCurrentStream < mStreams.Length(); NextStream()) { + stream = mStreams[mCurrentStream].mAsyncStream; + if (stream) { + break; + } + + uint64_t avail = 0; + nsresult rv = AvailableMaybeSeek(mStreams[mCurrentStream], &avail); + if (rv == NS_BASE_STREAM_CLOSED || (NS_SUCCEEDED(rv) && avail == 0)) { + // Nothing to read here. Let's move on. + continue; + } + + if (NS_FAILED(rv)) { + return rv; + } + + break; + } + } + + asyncWaitCallback = mAsyncWaitCallback ? this : nullptr; + asyncWaitFlags = mAsyncWaitFlags; + asyncWaitRequestedCount = mAsyncWaitRequestedCount; + asyncWaitEventTarget = mAsyncWaitEventTarget; + + MOZ_ASSERT_IF(stream, NS_SUCCEEDED(mStatus)); + } + + // If we are here it's because we are already closed, or if the current stream + // is not async. In both case we have to execute the callback. + if (!stream) { + if (asyncWaitCallback) { + AsyncWaitRunnable::Create(this, asyncWaitEventTarget); + } + return NS_OK; + } + + return stream->AsyncWait(asyncWaitCallback, asyncWaitFlags, + asyncWaitRequestedCount, asyncWaitEventTarget); +} + +NS_IMETHODIMP +nsMultiplexInputStream::OnInputStreamReady(nsIAsyncInputStream* aStream) { + nsCOMPtr<nsIInputStreamCallback> callback; + + // When OnInputStreamReady is called, we could be in 2 scenarios: + // a. there is something to read; + // b. the stream is closed. + // But if the stream is closed and it was not the last one, we must proceed + // with the following stream in order to have something to read by the callee. + + { + MutexAutoLock lock(mLock); + + // The callback has been nullified in the meantime. + if (!mAsyncWaitCallback) { + return NS_OK; + } + + if (NS_SUCCEEDED(mStatus)) { + uint64_t avail = 0; + nsresult rv = NS_OK; + // Only check `Available()` if `aStream` is actually the current stream, + // otherwise we'll always want to re-poll, as we got the callback for the + // wrong stream. + if (mCurrentStream < mStreams.Length() && + aStream == mStreams[mCurrentStream].mAsyncStream) { + rv = aStream->Available(&avail); + } + if (rv == NS_BASE_STREAM_CLOSED || (NS_SUCCEEDED(rv) && avail == 0)) { + // This stream is either closed, has no data available, or is not the + // current stream. If it is closed and current, move to the next stream, + // otherwise re-wait on the current stream until it has data available + // or becomes closed. + // Unlike streams not implementing nsIAsyncInputStream, async streams + // cannot use `Available() == 0` to indicate EOF, so we re-poll in that + // situation. + if (NS_FAILED(rv)) { + NextStream(); + } + + // Unlock and invoke AsyncWaitInternal to wait again. If this succeeds, + // we'll be called again, otherwise fall through and notify. + MutexAutoUnlock unlock(mLock); + if (NS_SUCCEEDED(AsyncWaitInternal())) { + return NS_OK; + } + } + } + + mAsyncWaitCallback.swap(callback); + mAsyncWaitEventTarget = nullptr; + } + + return callback ? callback->OnInputStreamReady(this) : NS_OK; +} + +void nsMultiplexInputStream::AsyncWaitCompleted() { + nsCOMPtr<nsIInputStreamCallback> callback; + + { + MutexAutoLock lock(mLock); + + // The callback has been nullified in the meantime. + if (!mAsyncWaitCallback) { + return; + } + + mAsyncWaitCallback.swap(callback); + mAsyncWaitEventTarget = nullptr; + } + + callback->OnInputStreamReady(this); +} + +nsresult nsMultiplexInputStreamConstructor(REFNSIID aIID, void** aResult) { + *aResult = nullptr; + + RefPtr<nsMultiplexInputStream> inst = new nsMultiplexInputStream(); + + return inst->QueryInterface(aIID, aResult); +} + +void nsMultiplexInputStream::SerializedComplexity(uint32_t aMaxSize, + uint32_t* aSizeUsed, + uint32_t* aPipes, + uint32_t* aTransferables) { + MutexAutoLock lock(mLock); + bool serializeAsPipe = false; + SerializedComplexityInternal(aMaxSize, aSizeUsed, aPipes, aTransferables, + &serializeAsPipe); +} + +void nsMultiplexInputStream::SerializedComplexityInternal( + uint32_t aMaxSize, uint32_t* aSizeUsed, uint32_t* aPipes, + uint32_t* aTransferables, bool* aSerializeAsPipe) { + mLock.AssertCurrentThreadOwns(); + CheckedUint32 totalSizeUsed = 0; + CheckedUint32 totalPipes = 0; + CheckedUint32 totalTransferables = 0; + CheckedUint32 maxSize = aMaxSize; + + uint32_t streamCount = mStreams.Length(); + + for (uint32_t index = 0; index < streamCount; index++) { + uint32_t sizeUsed = 0; + uint32_t pipes = 0; + uint32_t transferables = 0; + InputStreamHelper::SerializedComplexity(mStreams[index].mOriginalStream, + maxSize.value(), &sizeUsed, &pipes, + &transferables); + + MOZ_ASSERT(maxSize.value() >= sizeUsed); + + maxSize -= sizeUsed; + MOZ_DIAGNOSTIC_ASSERT(maxSize.isValid()); + totalSizeUsed += sizeUsed; + MOZ_DIAGNOSTIC_ASSERT(totalSizeUsed.isValid()); + totalPipes += pipes; + MOZ_DIAGNOSTIC_ASSERT(totalPipes.isValid()); + totalTransferables += transferables; + MOZ_DIAGNOSTIC_ASSERT(totalTransferables.isValid()); + } + + // If the combination of all streams when serialized independently is + // sufficiently complex, we may choose to serialize it as a pipe to limit the + // complexity of the payload. + if (totalTransferables.value() == 0) { + // If there are no transferables within our serialization, and it would + // contain at least one pipe, serialize the entire payload as a pipe for + // simplicity. + *aSerializeAsPipe = totalSizeUsed.value() > 0 && totalPipes.value() > 0; + } else { + // Otherwise, we may want to still serialize in segments to take advantage + // of the efficiency of serializing transferables. We'll only serialize as a + // pipe if the total attachment count exceeds kMaxAttachmentThreshold. + static constexpr uint32_t kMaxAttachmentThreshold = 8; + CheckedUint32 totalAttachments = totalPipes + totalTransferables; + *aSerializeAsPipe = !totalAttachments.isValid() || + totalAttachments.value() > kMaxAttachmentThreshold; + } + + if (*aSerializeAsPipe) { + NS_WARNING( + nsPrintfCString("Choosing to serialize multiplex stream as a pipe " + "(would be %u bytes, %u pipes, %u transferables)", + totalSizeUsed.value(), totalPipes.value(), + totalTransferables.value()) + .get()); + *aSizeUsed = 0; + *aPipes = 1; + *aTransferables = 0; + } else { + *aSizeUsed = totalSizeUsed.value(); + *aPipes = totalPipes.value(); + *aTransferables = totalTransferables.value(); + } +} + +void nsMultiplexInputStream::Serialize(InputStreamParams& aParams, + uint32_t aMaxSize, uint32_t* aSizeUsed) { + MutexAutoLock lock(mLock); + + // Check if we should serialize this stream as a pipe to reduce complexity. + uint32_t dummySizeUsed = 0, dummyPipes = 0, dummyTransferables = 0; + bool serializeAsPipe = false; + SerializedComplexityInternal(aMaxSize, &dummySizeUsed, &dummyPipes, + &dummyTransferables, &serializeAsPipe); + if (serializeAsPipe) { + *aSizeUsed = 0; + MutexAutoUnlock unlock(mLock); + InputStreamHelper::SerializeInputStreamAsPipe(this, aParams); + return; + } + + MultiplexInputStreamParams params; + + CheckedUint32 totalSizeUsed = 0; + CheckedUint32 maxSize = aMaxSize; + + uint32_t streamCount = mStreams.Length(); + if (streamCount) { + nsTArray<InputStreamParams>& streams = params.streams(); + + streams.SetCapacity(streamCount); + for (uint32_t index = 0; index < streamCount; index++) { + uint32_t sizeUsed = 0; + InputStreamHelper::SerializeInputStream(mStreams[index].mOriginalStream, + *streams.AppendElement(), + maxSize.value(), &sizeUsed); + + MOZ_ASSERT(maxSize.value() >= sizeUsed); + + maxSize -= sizeUsed; + MOZ_DIAGNOSTIC_ASSERT(maxSize.isValid()); + + totalSizeUsed += sizeUsed; + MOZ_DIAGNOSTIC_ASSERT(totalSizeUsed.isValid()); + } + } + + params.currentStream() = mCurrentStream; + params.status() = mStatus; + params.startedReadingCurrent() = mStartedReadingCurrent; + + aParams = std::move(params); + + MOZ_ASSERT(aSizeUsed); + *aSizeUsed = totalSizeUsed.value(); +} + +bool nsMultiplexInputStream::Deserialize(const InputStreamParams& aParams) { + if (aParams.type() != InputStreamParams::TMultiplexInputStreamParams) { + NS_ERROR("Received unknown parameters from the other process!"); + return false; + } + + const MultiplexInputStreamParams& params = + aParams.get_MultiplexInputStreamParams(); + + const nsTArray<InputStreamParams>& streams = params.streams(); + + uint32_t streamCount = streams.Length(); + for (uint32_t index = 0; index < streamCount; index++) { + nsCOMPtr<nsIInputStream> stream = + InputStreamHelper::DeserializeInputStream(streams[index]); + if (!stream) { + NS_WARNING("Deserialize failed!"); + return false; + } + + if (NS_FAILED(AppendStream(stream))) { + NS_WARNING("AppendStream failed!"); + return false; + } + } + + MutexAutoLock lock(mLock); + mCurrentStream = params.currentStream(); + mStatus = params.status(); + mStartedReadingCurrent = params.startedReadingCurrent(); + + return true; +} + +NS_IMETHODIMP +nsMultiplexInputStream::GetCloneable(bool* aCloneable) { + MutexAutoLock lock(mLock); + // XXXnsm Cloning a multiplex stream which has started reading is not + // permitted right now. + if (mCurrentStream > 0 || mStartedReadingCurrent) { + *aCloneable = false; + return NS_OK; + } + + uint32_t len = mStreams.Length(); + for (uint32_t i = 0; i < len; ++i) { + nsCOMPtr<nsICloneableInputStream> cis = + do_QueryInterface(mStreams[i].mBufferedStream); + if (!cis || !cis->GetCloneable()) { + *aCloneable = false; + return NS_OK; + } + } + + *aCloneable = true; + return NS_OK; +} + +NS_IMETHODIMP +nsMultiplexInputStream::Clone(nsIInputStream** aClone) { + MutexAutoLock lock(mLock); + + // XXXnsm Cloning a multiplex stream which has started reading is not + // permitted right now. + if (mCurrentStream > 0 || mStartedReadingCurrent) { + return NS_ERROR_FAILURE; + } + + RefPtr<nsMultiplexInputStream> clone = new nsMultiplexInputStream(); + + nsresult rv; + uint32_t len = mStreams.Length(); + for (uint32_t i = 0; i < len; ++i) { + nsCOMPtr<nsICloneableInputStream> substream = + do_QueryInterface(mStreams[i].mBufferedStream); + if (NS_WARN_IF(!substream)) { + return NS_ERROR_FAILURE; + } + + nsCOMPtr<nsIInputStream> clonedSubstream; + rv = substream->Clone(getter_AddRefs(clonedSubstream)); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + rv = clone->AppendStream(clonedSubstream); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + } + + clone.forget(aClone); + return NS_OK; +} + +NS_IMETHODIMP +nsMultiplexInputStream::Length(int64_t* aLength) { + MutexAutoLock lock(mLock); + + if (mCurrentStream > 0 || mStartedReadingCurrent) { + return NS_ERROR_NOT_AVAILABLE; + } + + CheckedInt64 length = 0; + nsresult retval = NS_OK; + + for (uint32_t i = 0, len = mStreams.Length(); i < len; ++i) { + nsCOMPtr<nsIInputStreamLength> substream = + do_QueryInterface(mStreams[i].mBufferedStream); + if (!substream) { + // Let's use available as fallback. + uint64_t streamAvail = 0; + nsresult rv = AvailableMaybeSeek(mStreams[i], &streamAvail); + if (rv == NS_BASE_STREAM_CLOSED) { + continue; + } + + if (NS_WARN_IF(NS_FAILED(rv))) { + mStatus = rv; + return mStatus; + } + + length += streamAvail; + if (!length.isValid()) { + return NS_ERROR_OUT_OF_MEMORY; + } + + continue; + } + + int64_t size = 0; + nsresult rv = substream->Length(&size); + if (rv == NS_BASE_STREAM_CLOSED) { + continue; + } + + if (rv == NS_ERROR_NOT_AVAILABLE) { + return rv; + } + + // If one stream blocks, we all block. + if (rv != NS_BASE_STREAM_WOULD_BLOCK && NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + // We want to return WOULD_BLOCK if there is 1 stream that blocks. But want + // to see if there are other streams with length = -1. + if (rv == NS_BASE_STREAM_WOULD_BLOCK) { + retval = NS_BASE_STREAM_WOULD_BLOCK; + continue; + } + + // If one of the stream doesn't know the size, we all don't know the size. + if (size == -1) { + *aLength = -1; + return NS_OK; + } + + length += size; + if (!length.isValid()) { + return NS_ERROR_OUT_OF_MEMORY; + } + } + + *aLength = length.value(); + return retval; +} + +class nsMultiplexInputStream::AsyncWaitLengthHelper final + : public nsIInputStreamLengthCallback { + public: + NS_DECL_THREADSAFE_ISUPPORTS + + AsyncWaitLengthHelper() + : mStreamNotified(false), mLength(0), mNegativeSize(false) {} + + bool AddStream(nsIAsyncInputStreamLength* aStream) { + return mPendingStreams.AppendElement(aStream, fallible); + } + + bool AddSize(int64_t aSize) { + MOZ_ASSERT(!mNegativeSize); + + mLength += aSize; + return mLength.isValid(); + } + + void NegativeSize() { + MOZ_ASSERT(!mNegativeSize); + mNegativeSize = true; + } + + nsresult Proceed(nsMultiplexInputStream* aParentStream, + nsIEventTarget* aEventTarget, + const MutexAutoLock& aProofOfLock) { + MOZ_ASSERT(!mStream); + + // If we don't need to wait, let's inform the callback immediately. + if (mPendingStreams.IsEmpty() || mNegativeSize) { + RefPtr<nsMultiplexInputStream> parentStream = aParentStream; + int64_t length = -1; + if (!mNegativeSize && mLength.isValid()) { + length = mLength.value(); + } + nsCOMPtr<nsIRunnable> r = NS_NewRunnableFunction( + "AsyncWaitLengthHelper", [parentStream, length]() { + MutexAutoLock lock(parentStream->GetLock()); + parentStream->AsyncWaitCompleted(length, lock); + }); + return aEventTarget->Dispatch(r.forget(), NS_DISPATCH_NORMAL); + } + + // Let's store the callback and the parent stream until we have + // notifications from the async length streams. + + mStream = aParentStream; + + // Let's activate all the pending streams. + for (nsIAsyncInputStreamLength* stream : mPendingStreams) { + nsresult rv = stream->AsyncLengthWait(this, aEventTarget); + if (rv == NS_BASE_STREAM_CLOSED) { + continue; + } + + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + } + + return NS_OK; + } + + NS_IMETHOD + OnInputStreamLengthReady(nsIAsyncInputStreamLength* aStream, + int64_t aLength) override { + MutexAutoLock lock(mStream->GetLock()); + + MOZ_ASSERT(mPendingStreams.Contains(aStream)); + mPendingStreams.RemoveElement(aStream); + + // Already notified. + if (mStreamNotified) { + return NS_OK; + } + + if (aLength == -1) { + mNegativeSize = true; + } else { + mLength += aLength; + if (!mLength.isValid()) { + mNegativeSize = true; + } + } + + // We need to wait. + if (!mNegativeSize && !mPendingStreams.IsEmpty()) { + return NS_OK; + } + + // Let's notify the parent stream. + mStreamNotified = true; + mStream->AsyncWaitCompleted(mNegativeSize ? -1 : mLength.value(), lock); + return NS_OK; + } + + private: + ~AsyncWaitLengthHelper() = default; + + RefPtr<nsMultiplexInputStream> mStream; + bool mStreamNotified; + + CheckedInt64 mLength; + bool mNegativeSize; + + nsTArray<nsCOMPtr<nsIAsyncInputStreamLength>> mPendingStreams; +}; + +NS_IMPL_ISUPPORTS(nsMultiplexInputStream::AsyncWaitLengthHelper, + nsIInputStreamLengthCallback) + +NS_IMETHODIMP +nsMultiplexInputStream::AsyncLengthWait(nsIInputStreamLengthCallback* aCallback, + nsIEventTarget* aEventTarget) { + if (NS_WARN_IF(!aEventTarget)) { + return NS_ERROR_NULL_POINTER; + } + + MutexAutoLock lock(mLock); + + if (mCurrentStream > 0 || mStartedReadingCurrent) { + return NS_ERROR_NOT_AVAILABLE; + } + + if (!aCallback) { + mAsyncWaitLengthCallback = nullptr; + return NS_OK; + } + + // We have a pending operation! Let's use this instead of creating a new one. + if (mAsyncWaitLengthHelper) { + mAsyncWaitLengthCallback = aCallback; + return NS_OK; + } + + RefPtr<AsyncWaitLengthHelper> helper = new AsyncWaitLengthHelper(); + + for (uint32_t i = 0, len = mStreams.Length(); i < len; ++i) { + nsCOMPtr<nsIAsyncInputStreamLength> asyncStream = + do_QueryInterface(mStreams[i].mBufferedStream); + if (asyncStream) { + if (NS_WARN_IF(!helper->AddStream(asyncStream))) { + return NS_ERROR_OUT_OF_MEMORY; + } + continue; + } + + nsCOMPtr<nsIInputStreamLength> stream = + do_QueryInterface(mStreams[i].mBufferedStream); + if (!stream) { + // Let's use available as fallback. + uint64_t streamAvail = 0; + nsresult rv = AvailableMaybeSeek(mStreams[i], &streamAvail); + if (rv == NS_BASE_STREAM_CLOSED) { + continue; + } + + if (NS_WARN_IF(NS_FAILED(rv))) { + mStatus = rv; + return mStatus; + } + + if (NS_WARN_IF(!helper->AddSize(streamAvail))) { + return NS_ERROR_OUT_OF_MEMORY; + } + + continue; + } + + int64_t size = 0; + nsresult rv = stream->Length(&size); + if (rv == NS_BASE_STREAM_CLOSED) { + continue; + } + + MOZ_ASSERT(rv != NS_BASE_STREAM_WOULD_BLOCK, + "A nsILengthInutStream returns NS_BASE_STREAM_WOULD_BLOCK but " + "it doesn't implement nsIAsyncInputStreamLength."); + + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + if (size == -1) { + helper->NegativeSize(); + break; + } + + if (NS_WARN_IF(!helper->AddSize(size))) { + return NS_ERROR_OUT_OF_MEMORY; + } + } + + nsresult rv = helper->Proceed(this, aEventTarget, lock); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + mAsyncWaitLengthHelper = helper; + mAsyncWaitLengthCallback = aCallback; + return NS_OK; +} + +void nsMultiplexInputStream::AsyncWaitCompleted( + int64_t aLength, const MutexAutoLock& aProofOfLock) { + mLock.AssertCurrentThreadOwns(); + + nsCOMPtr<nsIInputStreamLengthCallback> callback; + callback.swap(mAsyncWaitLengthCallback); + + mAsyncWaitLengthHelper = nullptr; + + // Already canceled. + if (!callback) { + return; + } + + MutexAutoUnlock unlock(mLock); + callback->OnInputStreamLengthReady(this, aLength); +} + +#define MAYBE_UPDATE_VALUE_REAL(x, y) \ + if (y) { \ + ++x; \ + } + +#define MAYBE_UPDATE_VALUE(x, y) \ + { \ + nsCOMPtr<y> substream = do_QueryInterface(aStream.mBufferedStream); \ + MAYBE_UPDATE_VALUE_REAL(x, substream) \ + } + +#define MAYBE_UPDATE_BOOL(x, y) \ + if (!x) { \ + nsCOMPtr<y> substream = do_QueryInterface(aStream.mBufferedStream); \ + if (substream) { \ + x = true; \ + } \ + } + +void nsMultiplexInputStream::UpdateQIMap(StreamData& aStream) { + auto length = mStreams.Length(); + + MAYBE_UPDATE_VALUE_REAL(mSeekableStreams, aStream.mSeekableStream) + mIsSeekableStream = (mSeekableStreams == length); + MAYBE_UPDATE_VALUE(mIPCSerializableStreams, nsIIPCSerializableInputStream) + mIsIPCSerializableStream = (mIPCSerializableStreams == length); + MAYBE_UPDATE_VALUE(mCloneableStreams, nsICloneableInputStream) + mIsCloneableStream = (mCloneableStreams == length); + // nsMultiplexInputStream is nsIAsyncInputStream if at least 1 of the + // substream implements that interface + if (!mIsAsyncInputStream && aStream.mAsyncStream) { + mIsAsyncInputStream = true; + } + MAYBE_UPDATE_BOOL(mIsInputStreamLength, nsIInputStreamLength) + MAYBE_UPDATE_BOOL(mIsAsyncInputStreamLength, nsIAsyncInputStreamLength) +} + +#undef MAYBE_UPDATE_VALUE +#undef MAYBE_UPDATE_VALUE_REAL +#undef MAYBE_UPDATE_BOOL + +bool nsMultiplexInputStream::IsSeekable() const { return mIsSeekableStream; } + +bool nsMultiplexInputStream::IsIPCSerializable() const { + return mIsIPCSerializableStream; +} + +bool nsMultiplexInputStream::IsCloneable() const { return mIsCloneableStream; } + +bool nsMultiplexInputStream::IsAsyncInputStream() const { + // nsMultiplexInputStream is nsIAsyncInputStream if at least 1 of the + // substream implements that interface. + return mIsAsyncInputStream; +} + +bool nsMultiplexInputStream::IsInputStreamLength() const { + return mIsInputStreamLength; +} + +bool nsMultiplexInputStream::IsAsyncInputStreamLength() const { + return mIsAsyncInputStreamLength; +} |