Adding upstream version 1:115.7.0.upstream/1%115.7.0upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

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;
+}