/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim:set ts=4 sts=2 sw=2 et cin: */
/* 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 "nsIOService.h"
#include "nsInputStreamPump.h"
#include "nsIStreamTransportService.h"
#include "nsIThreadRetargetableStreamListener.h"
#include "nsThreadUtils.h"
#include "nsCOMPtr.h"
#include "mozilla/Logging.h"
#include "mozilla/NonBlockingAsyncInputStream.h"
#include "mozilla/ProfilerLabels.h"
#include "mozilla/SlicedInputStream.h"
#include "mozilla/StaticPrefs_network.h"
#include "nsIStreamListener.h"
#include "nsILoadGroup.h"
#include "nsNetCID.h"
#include "nsNetUtil.h"
#include "nsStreamUtils.h"
#include <algorithm>

//
// MOZ_LOG=nsStreamPump:5
//
static mozilla::LazyLogModule gStreamPumpLog("nsStreamPump");
#undef LOG
#define LOG(args) MOZ_LOG(gStreamPumpLog, mozilla::LogLevel::Debug, args)

//-----------------------------------------------------------------------------
// nsInputStreamPump methods
//-----------------------------------------------------------------------------

nsInputStreamPump::nsInputStreamPump() : mOffMainThread(!NS_IsMainThread()) {}

nsresult nsInputStreamPump::Create(nsInputStreamPump** result,
                                   nsIInputStream* stream, uint32_t segsize,
                                   uint32_t segcount, bool closeWhenDone,
                                   nsISerialEventTarget* mainThreadTarget) {
  nsresult rv = NS_ERROR_OUT_OF_MEMORY;
  RefPtr<nsInputStreamPump> pump = new nsInputStreamPump();
  if (pump) {
    rv = pump->Init(stream, segsize, segcount, closeWhenDone, mainThreadTarget);
    if (NS_SUCCEEDED(rv)) {
      pump.forget(result);
    }
  }
  return rv;
}

struct PeekData {
  PeekData(nsInputStreamPump::PeekSegmentFun fun, void* closure)
      : mFunc(fun), mClosure(closure) {}

  nsInputStreamPump::PeekSegmentFun mFunc;
  void* mClosure;
};

static nsresult CallPeekFunc(nsIInputStream* aInStream, void* aClosure,
                             const char* aFromSegment, uint32_t aToOffset,
                             uint32_t aCount, uint32_t* aWriteCount) {
  NS_ASSERTION(aToOffset == 0, "Called more than once?");
  NS_ASSERTION(aCount > 0, "Called without data?");

  PeekData* data = static_cast<PeekData*>(aClosure);
  data->mFunc(data->mClosure, reinterpret_cast<const uint8_t*>(aFromSegment),
              aCount);
  return NS_BINDING_ABORTED;
}

nsresult nsInputStreamPump::PeekStream(PeekSegmentFun callback, void* closure) {
  RecursiveMutexAutoLock lock(mMutex);

  MOZ_ASSERT(mAsyncStream, "PeekStream called without stream");

  nsresult rv = CreateBufferedStreamIfNeeded();
  NS_ENSURE_SUCCESS(rv, rv);

  // See if the pipe is closed by checking the return of Available.
  uint64_t dummy64;
  rv = mAsyncStream->Available(&dummy64);
  if (NS_FAILED(rv)) return rv;
  uint32_t dummy = (uint32_t)std::min(dummy64, (uint64_t)UINT32_MAX);

  PeekData data(callback, closure);
  return mAsyncStream->ReadSegments(
      CallPeekFunc, &data, mozilla::net::nsIOService::gDefaultSegmentSize,
      &dummy);
}

nsresult nsInputStreamPump::EnsureWaiting() {
  mMutex.AssertCurrentThreadIn();

  // no need to worry about multiple threads... an input stream pump lives
  // on only one thread at a time.
  MOZ_ASSERT(mAsyncStream);
  if (!mWaitingForInputStreamReady && !mProcessingCallbacks) {
    // Ensure OnStateStop is called on the main thread only when this pump is
    // created on main thread.
    if (mState == STATE_STOP && !mOffMainThread) {
      nsCOMPtr<nsISerialEventTarget> mainThread =
          mLabeledMainThreadTarget
              ? mLabeledMainThreadTarget
              : do_AddRef(mozilla::GetMainThreadSerialEventTarget());
      if (mTargetThread != mainThread) {
        mTargetThread = mainThread;
      }
    }
    MOZ_ASSERT(mTargetThread);
    nsresult rv = mAsyncStream->AsyncWait(this, 0, 0, mTargetThread);
    if (NS_FAILED(rv)) {
      NS_ERROR("AsyncWait failed");
      return rv;
    }
    // Any retargeting during STATE_START or START_TRANSFER is complete
    // after the call to AsyncWait; next callback will be on mTargetThread.
    mRetargeting = false;
    mWaitingForInputStreamReady = true;
  }
  return NS_OK;
}

//-----------------------------------------------------------------------------
// nsInputStreamPump::nsISupports
//-----------------------------------------------------------------------------

// although this class can only be accessed from one thread at a time, we do
// allow its ownership to move from thread to thread, assuming the consumer
// understands the limitations of this.
NS_IMPL_ADDREF(nsInputStreamPump)
NS_IMPL_RELEASE(nsInputStreamPump)
NS_INTERFACE_MAP_BEGIN(nsInputStreamPump)
  NS_INTERFACE_MAP_ENTRY(nsIRequest)
  NS_INTERFACE_MAP_ENTRY(nsIThreadRetargetableRequest)
  NS_INTERFACE_MAP_ENTRY(nsIInputStreamCallback)
  NS_INTERFACE_MAP_ENTRY(nsIInputStreamPump)
  NS_INTERFACE_MAP_ENTRY_CONCRETE(nsInputStreamPump)
  NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsIInputStreamPump)
NS_INTERFACE_MAP_END

//-----------------------------------------------------------------------------
// nsInputStreamPump::nsIRequest
//-----------------------------------------------------------------------------

NS_IMETHODIMP
nsInputStreamPump::GetName(nsACString& result) {
  RecursiveMutexAutoLock lock(mMutex);

  result.Truncate();
  return NS_OK;
}

NS_IMETHODIMP
nsInputStreamPump::IsPending(bool* result) {
  RecursiveMutexAutoLock lock(mMutex);

  *result = (mState != STATE_IDLE && mState != STATE_DEAD);
  return NS_OK;
}

NS_IMETHODIMP
nsInputStreamPump::GetStatus(nsresult* status) {
  RecursiveMutexAutoLock lock(mMutex);

  *status = mStatus;
  return NS_OK;
}

NS_IMETHODIMP nsInputStreamPump::SetCanceledReason(const nsACString& aReason) {
  return SetCanceledReasonImpl(aReason);
}

NS_IMETHODIMP nsInputStreamPump::GetCanceledReason(nsACString& aReason) {
  return GetCanceledReasonImpl(aReason);
}

NS_IMETHODIMP nsInputStreamPump::CancelWithReason(nsresult aStatus,
                                                  const nsACString& aReason) {
  return CancelWithReasonImpl(aStatus, aReason);
}

NS_IMETHODIMP
nsInputStreamPump::Cancel(nsresult status) {
  RecursiveMutexAutoLock lock(mMutex);

  AssertOnThread();

  LOG(("nsInputStreamPump::Cancel [this=%p status=%" PRIx32 "]\n", this,
       static_cast<uint32_t>(status)));

  if (NS_FAILED(mStatus)) {
    LOG(("  already canceled\n"));
    return NS_OK;
  }

  NS_ASSERTION(NS_FAILED(status), "cancel with non-failure status code");
  mStatus = status;

  // close input stream
  if (mAsyncStream) {
    // If mSuspendCount != 0, EnsureWaiting will be called by Resume().
    // Note that while suspended, OnInputStreamReady will
    // not do anything, and also note that calling asyncWait
    // on a closed stream works and will dispatch an event immediately.

    nsCOMPtr<nsIEventTarget> currentTarget = NS_GetCurrentThread();
    if (mTargetThread && currentTarget != mTargetThread) {
      nsresult rv = mTargetThread->Dispatch(NS_NewRunnableFunction(
          "nsInputStreamPump::Cancel", [self = RefPtr{this}, status] {
            RecursiveMutexAutoLock lock(self->mMutex);
            if (!self->mAsyncStream) {
              return;
            }
            self->mAsyncStream->CloseWithStatus(status);
            if (self->mSuspendCount == 0) {
              self->EnsureWaiting();
            }
          }));
      NS_ENSURE_SUCCESS(rv, rv);
    } else {
      mAsyncStream->CloseWithStatus(status);
      if (mSuspendCount == 0) {
        EnsureWaiting();
      }
    }
  }
  return NS_OK;
}

NS_IMETHODIMP
nsInputStreamPump::Suspend() {
  RecursiveMutexAutoLock lock(mMutex);

  LOG(("nsInputStreamPump::Suspend [this=%p]\n", this));
  NS_ENSURE_TRUE(mState != STATE_IDLE && mState != STATE_DEAD,
                 NS_ERROR_UNEXPECTED);
  ++mSuspendCount;
  return NS_OK;
}

NS_IMETHODIMP
nsInputStreamPump::Resume() {
  RecursiveMutexAutoLock lock(mMutex);

  LOG(("nsInputStreamPump::Resume [this=%p]\n", this));
  NS_ENSURE_TRUE(mSuspendCount > 0, NS_ERROR_UNEXPECTED);
  NS_ENSURE_TRUE(mState != STATE_IDLE && mState != STATE_DEAD,
                 NS_ERROR_UNEXPECTED);

  // There is a brief in-between state when we null out mAsyncStream in
  // OnStateStop() before calling OnStopRequest, and only afterwards set
  // STATE_DEAD, which we need to handle gracefully.
  if (--mSuspendCount == 0 && mAsyncStream) {
    EnsureWaiting();
  }
  return NS_OK;
}

NS_IMETHODIMP
nsInputStreamPump::GetLoadFlags(nsLoadFlags* aLoadFlags) {
  RecursiveMutexAutoLock lock(mMutex);

  *aLoadFlags = mLoadFlags;
  return NS_OK;
}

NS_IMETHODIMP
nsInputStreamPump::SetLoadFlags(nsLoadFlags aLoadFlags) {
  RecursiveMutexAutoLock lock(mMutex);

  mLoadFlags = aLoadFlags;
  return NS_OK;
}

NS_IMETHODIMP
nsInputStreamPump::GetTRRMode(nsIRequest::TRRMode* aTRRMode) {
  return GetTRRModeImpl(aTRRMode);
}

NS_IMETHODIMP
nsInputStreamPump::SetTRRMode(nsIRequest::TRRMode aTRRMode) {
  return SetTRRModeImpl(aTRRMode);
}

NS_IMETHODIMP
nsInputStreamPump::GetLoadGroup(nsILoadGroup** aLoadGroup) {
  RecursiveMutexAutoLock lock(mMutex);

  *aLoadGroup = do_AddRef(mLoadGroup).take();
  return NS_OK;
}

NS_IMETHODIMP
nsInputStreamPump::SetLoadGroup(nsILoadGroup* aLoadGroup) {
  RecursiveMutexAutoLock lock(mMutex);

  mLoadGroup = aLoadGroup;
  return NS_OK;
}

//-----------------------------------------------------------------------------
// nsInputStreamPump::nsIInputStreamPump implementation
//-----------------------------------------------------------------------------

NS_IMETHODIMP
nsInputStreamPump::Init(nsIInputStream* stream, uint32_t segsize,
                        uint32_t segcount, bool closeWhenDone,
                        nsISerialEventTarget* mainThreadTarget) {
  // probably we can't be multithread-accessed yet
  RecursiveMutexAutoLock lock(mMutex);
  NS_ENSURE_TRUE(mState == STATE_IDLE, NS_ERROR_IN_PROGRESS);

  mStream = stream;
  mSegSize = segsize;
  mSegCount = segcount;
  mCloseWhenDone = closeWhenDone;
  mLabeledMainThreadTarget = mainThreadTarget;
  if (mOffMainThread && mLabeledMainThreadTarget) {
    MOZ_ASSERT(
        false,
        "Init stream pump off main thread with a main thread event target.");
    return NS_ERROR_FAILURE;
  }

  return NS_OK;
}

NS_IMETHODIMP
nsInputStreamPump::AsyncRead(nsIStreamListener* listener) {
  RecursiveMutexAutoLock lock(mMutex);

  // This ensures only one thread can interact with a pump at a time
  NS_ENSURE_TRUE(mState == STATE_IDLE, NS_ERROR_IN_PROGRESS);
  NS_ENSURE_ARG_POINTER(listener);
  MOZ_ASSERT(NS_IsMainThread() || mOffMainThread,
             "nsInputStreamPump should be read from the "
             "main thread only.");

  nsresult rv = NS_MakeAsyncNonBlockingInputStream(
      mStream.forget(), getter_AddRefs(mAsyncStream), mCloseWhenDone, mSegSize,
      mSegCount);
  if (NS_WARN_IF(NS_FAILED(rv))) {
    return rv;
  }

  MOZ_ASSERT(mAsyncStream);

  // mStreamOffset now holds the number of bytes currently read.
  mStreamOffset = 0;

  // grab event queue (we must do this here by contract, since all notifications
  // must go to the thread which called AsyncRead)
  if (NS_IsMainThread() && mLabeledMainThreadTarget) {
    mTargetThread = mLabeledMainThreadTarget;
  } else {
    mTargetThread = mozilla::GetCurrentSerialEventTarget();
  }
  NS_ENSURE_STATE(mTargetThread);

  rv = EnsureWaiting();
  if (NS_FAILED(rv)) return rv;

  if (mLoadGroup) mLoadGroup->AddRequest(this, nullptr);

  mState = STATE_START;
  mListener = listener;
  return NS_OK;
}

//-----------------------------------------------------------------------------
// nsInputStreamPump::nsIInputStreamCallback implementation
//-----------------------------------------------------------------------------

NS_IMETHODIMP
nsInputStreamPump::OnInputStreamReady(nsIAsyncInputStream* stream) {
  LOG(("nsInputStreamPump::OnInputStreamReady [this=%p]\n", this));

  AUTO_PROFILER_LABEL("nsInputStreamPump::OnInputStreamReady", NETWORK);

  // this function has been called from a PLEvent, so we can safely call
  // any listener or progress sink methods directly from here.

  for (;;) {
    // There should only be one iteration of this loop happening at a time.
    // To prevent AsyncWait() (called during callbacks or on other threads)
    // from creating a parallel OnInputStreamReady(), we use:
    // -- a mutex; and
    // -- a boolean mProcessingCallbacks to detect parallel loops
    //    when exiting the mutex for callbacks.
    RecursiveMutexAutoLock lock(mMutex);

    // Prevent parallel execution during callbacks, while out of mutex.
    if (mProcessingCallbacks) {
      MOZ_ASSERT(!mProcessingCallbacks);
      break;
    }
    mProcessingCallbacks = true;
    if (mSuspendCount || mState == STATE_IDLE || mState == STATE_DEAD) {
      mWaitingForInputStreamReady = false;
      mProcessingCallbacks = false;
      break;
    }

    uint32_t nextState;
    switch (mState) {
      case STATE_START:
        nextState = OnStateStart();
        break;
      case STATE_TRANSFER:
        nextState = OnStateTransfer();
        break;
      case STATE_STOP:
        mRetargeting = false;
        nextState = OnStateStop();
        break;
      default:
        nextState = 0;
        MOZ_ASSERT_UNREACHABLE("Unknown enum value.");
        return NS_ERROR_UNEXPECTED;
    }

    bool stillTransferring =
        (mState == STATE_TRANSFER && nextState == STATE_TRANSFER);
    if (stillTransferring) {
      NS_ASSERTION(NS_SUCCEEDED(mStatus),
                   "Should not have failed status for ongoing transfer");
    } else {
      NS_ASSERTION(mState != nextState,
                   "Only OnStateTransfer can be called more than once.");
    }
    if (mRetargeting) {
      NS_ASSERTION(mState != STATE_STOP,
                   "Retargeting should not happen during OnStateStop.");
    }

    // Set mRetargeting so EnsureWaiting will be called. It ensures that
    // OnStateStop is called on the main thread.
    if (nextState == STATE_STOP && !NS_IsMainThread() && !mOffMainThread) {
      mRetargeting = true;
    }

    // Unset mProcessingCallbacks here (while we have lock) so our own call to
    // EnsureWaiting isn't blocked by it.
    mProcessingCallbacks = false;

    // We must break the loop if suspended during one of the previous
    // operation.
    if (mSuspendCount) {
      mState = nextState;
      mWaitingForInputStreamReady = false;
      break;
    }

    // Wait asynchronously if there is still data to transfer, or we're
    // switching event delivery to another thread.
    if (stillTransferring || mRetargeting) {
      mState = nextState;
      mWaitingForInputStreamReady = false;
      nsresult rv = EnsureWaiting();
      if (NS_SUCCEEDED(rv)) break;

      // Failure to start asynchronous wait: stop transfer.
      // Do not set mStatus if it was previously set to report a failure.
      if (NS_SUCCEEDED(mStatus)) {
        mStatus = rv;
      }
      nextState = STATE_STOP;
    }

    mState = nextState;
  }
  return NS_OK;
}

uint32_t nsInputStreamPump::OnStateStart() {
  mMutex.AssertCurrentThreadIn();

  AUTO_PROFILER_LABEL("nsInputStreamPump::OnStateStart", NETWORK);

  LOG(("  OnStateStart [this=%p]\n", this));

  nsresult rv;

  // need to check the reason why the stream is ready.  this is required
  // so our listener can check our status from OnStartRequest.
  // XXX async streams should have a GetStatus method!
  if (NS_SUCCEEDED(mStatus)) {
    uint64_t avail;
    rv = mAsyncStream->Available(&avail);
    if (NS_FAILED(rv) && rv != NS_BASE_STREAM_CLOSED) mStatus = rv;
  }

  {
    nsCOMPtr<nsIStreamListener> listener = mListener;
    // We're on the writing thread
    AssertOnThread();

    // Note: Must exit mutex for call to OnStartRequest to avoid
    // deadlocks when calls to RetargetDeliveryTo for multiple
    // nsInputStreamPumps are needed (e.g. nsHttpChannel).
    RecursiveMutexAutoUnlock unlock(mMutex);
    rv = listener->OnStartRequest(this);
  }

  // an error returned from OnStartRequest should cause us to abort; however,
  // we must not stomp on mStatus if already canceled.
  if (NS_FAILED(rv) && NS_SUCCEEDED(mStatus)) mStatus = rv;

  return NS_SUCCEEDED(mStatus) ? STATE_TRANSFER : STATE_STOP;
}

uint32_t nsInputStreamPump::OnStateTransfer() {
  mMutex.AssertCurrentThreadIn();

  AUTO_PROFILER_LABEL("nsInputStreamPump::OnStateTransfer", NETWORK);

  LOG(("  OnStateTransfer [this=%p]\n", this));

  // if canceled, go directly to STATE_STOP...
  if (NS_FAILED(mStatus)) return STATE_STOP;

  nsresult rv = CreateBufferedStreamIfNeeded();
  if (NS_WARN_IF(NS_FAILED(rv))) {
    return STATE_STOP;
  }

  uint64_t avail;
  rv = mAsyncStream->Available(&avail);
  LOG(("  Available returned [stream=%p rv=%" PRIx32 " avail=%" PRIu64 "]\n",
       mAsyncStream.get(), static_cast<uint32_t>(rv), avail));

  if (rv == NS_BASE_STREAM_CLOSED) {
    rv = NS_OK;
    avail = 0;
  } else if (NS_SUCCEEDED(rv) && avail) {
    // we used to limit avail to 16K - we were afraid some ODA handlers
    // might assume they wouldn't get more than 16K at once
    // we're removing that limit since it speeds up local file access.
    // Now there's an implicit 64K limit of 4 16K segments
    // NOTE: ok, so the story is as follows.  OnDataAvailable impls
    //       are by contract supposed to consume exactly |avail| bytes.
    //       however, many do not... mailnews... stream converters...
    //       cough, cough.  the input stream pump is fairly tolerant
    //       in this regard; however, if an ODA does not consume any
    //       data from the stream, then we could potentially end up in
    //       an infinite loop.  we do our best here to try to catch
    //       such an error.  (see bug 189672)

    // in most cases this QI will succeed (mAsyncStream is almost always
    // a nsPipeInputStream, which implements nsITellableStream::Tell).
    int64_t offsetBefore;
    nsCOMPtr<nsITellableStream> tellable = do_QueryInterface(mAsyncStream);
    if (tellable && NS_FAILED(tellable->Tell(&offsetBefore))) {
      MOZ_ASSERT_UNREACHABLE("Tell failed on readable stream");
      offsetBefore = 0;
    }

    uint32_t odaAvail = avail > UINT32_MAX ? UINT32_MAX : uint32_t(avail);

    LOG(("  calling OnDataAvailable [offset=%" PRIu64 " count=%" PRIu64
         "(%u)]\n",
         mStreamOffset, avail, odaAvail));

    {
      // We may be called on non-MainThread even if mOffMainThread is
      // false, due to RetargetDeliveryTo(), so don't use AssertOnThread()
      if (mTargetThread) {
        MOZ_ASSERT(mTargetThread->IsOnCurrentThread());
      } else {
        MOZ_ASSERT(NS_IsMainThread());
      }

      nsCOMPtr<nsIStreamListener> listener = mListener;
      // Note: Must exit mutex for call to OnStartRequest to avoid
      // deadlocks when calls to RetargetDeliveryTo for multiple
      // nsInputStreamPumps are needed (e.g. nsHttpChannel).
      RecursiveMutexAutoUnlock unlock(mMutex);
      // We're on the writing thread for mListener and mAsyncStream.
      // mStreamOffset is only touched in OnStateTransfer, and AsyncRead
      // shouldn't be called during OnDataAvailable()

      MOZ_PUSH_IGNORE_THREAD_SAFETY
      rv = listener->OnDataAvailable(this, mAsyncStream, mStreamOffset,
                                     odaAvail);
      MOZ_POP_THREAD_SAFETY
    }

    // don't enter this code if ODA failed or called Cancel
    if (NS_SUCCEEDED(rv) && NS_SUCCEEDED(mStatus)) {
      // test to see if this ODA failed to consume data
      if (tellable) {
        // NOTE: if Tell fails, which can happen if the stream is
        // now closed, then we assume that everything was read.
        int64_t offsetAfter;
        if (NS_FAILED(tellable->Tell(&offsetAfter))) {
          offsetAfter = offsetBefore + odaAvail;
        }
        if (offsetAfter > offsetBefore) {
          mStreamOffset += (offsetAfter - offsetBefore);
        } else if (mSuspendCount == 0) {
          //
          // possible infinite loop if we continue pumping data!
          //
          // NOTE: although not allowed by nsIStreamListener, we
          // will allow the ODA impl to Suspend the pump.  IMAP
          // does this :-(
          //
          NS_ERROR("OnDataAvailable implementation consumed no data");
          mStatus = NS_ERROR_UNEXPECTED;
        }
      } else {
        mStreamOffset += odaAvail;  // assume ODA behaved well
      }
    }
  }

  // an error returned from Available or OnDataAvailable should cause us to
  // abort; however, we must not stop on mStatus if already canceled.

  if (NS_SUCCEEDED(mStatus)) {
    if (NS_FAILED(rv)) {
      mStatus = rv;
    } else if (avail) {
      // if stream is now closed, advance to STATE_STOP right away.
      // Available may return 0 bytes available at the moment; that
      // would not mean that we are done.
      // XXX async streams should have a GetStatus method!
      rv = mAsyncStream->Available(&avail);
      if (NS_SUCCEEDED(rv)) return STATE_TRANSFER;
      if (rv != NS_BASE_STREAM_CLOSED) mStatus = rv;
    }
  }
  return STATE_STOP;
}

nsresult nsInputStreamPump::CallOnStateStop() {
  RecursiveMutexAutoLock lock(mMutex);

  MOZ_ASSERT(NS_IsMainThread(),
             "CallOnStateStop should only be called on the main thread.");

  mState = OnStateStop();
  return NS_OK;
}

uint32_t nsInputStreamPump::OnStateStop() {
  mMutex.AssertCurrentThreadIn();

  if (!NS_IsMainThread() && !mOffMainThread) {
    // This method can be called on a different thread if nsInputStreamPump
    // is used off the main-thread.
    if (NS_SUCCEEDED(mStatus) && mListener &&
        mozilla::StaticPrefs::network_send_OnDataFinished_nsInputStreamPump()) {
      nsCOMPtr<nsIThreadRetargetableStreamListener> retargetableListener =
          do_QueryInterface(mListener);
      if (retargetableListener) {
        retargetableListener->OnDataFinished(mStatus);
      }
    }
    nsresult rv = mLabeledMainThreadTarget->Dispatch(
        mozilla::NewRunnableMethod("nsInputStreamPump::CallOnStateStop", this,
                                   &nsInputStreamPump::CallOnStateStop));
    NS_ENSURE_SUCCESS(rv, STATE_DEAD);
    return STATE_DEAD;
  }

  AUTO_PROFILER_LABEL("nsInputStreamPump::OnStateStop", NETWORK);

  LOG(("  OnStateStop [this=%p status=%" PRIx32 "]\n", this,
       static_cast<uint32_t>(mStatus)));

  // if an error occurred, we must be sure to pass the error onto the async
  // stream.  in some cases, this is redundant, but since close is idempotent,
  // this is OK.  otherwise, be sure to honor the "close-when-done" option.

  if (!mAsyncStream || !mListener) {
    MOZ_ASSERT(mAsyncStream, "null mAsyncStream: OnStateStop called twice?");
    MOZ_ASSERT(mListener, "null mListener: OnStateStop called twice?");
    return STATE_DEAD;
  }

  if (NS_FAILED(mStatus)) {
    mAsyncStream->CloseWithStatus(mStatus);
  } else if (mCloseWhenDone) {
    mAsyncStream->Close();
  }

  mAsyncStream = nullptr;
  mIsPending = false;
  {
    // We're on the writing thread.
    // We believe that mStatus can't be changed on us here.
    AssertOnThread();

    nsCOMPtr<nsIStreamListener> listener = mListener;
    nsresult status = mStatus;
    // Note: Must exit mutex for call to OnStartRequest to avoid
    // deadlocks when calls to RetargetDeliveryTo for multiple
    // nsInputStreamPumps are needed (e.g. nsHttpChannel).
    RecursiveMutexAutoUnlock unlock(mMutex);

    listener->OnStopRequest(this, status);
  }
  mTargetThread = nullptr;
  mListener = nullptr;

  if (mLoadGroup) mLoadGroup->RemoveRequest(this, nullptr, mStatus);

  return STATE_DEAD;
}

nsresult nsInputStreamPump::CreateBufferedStreamIfNeeded() {
  if (mAsyncStreamIsBuffered) {
    return NS_OK;
  }

  // ReadSegments is not available for any nsIAsyncInputStream. In order to use
  // it, we wrap a nsIBufferedInputStream around it, if needed.

  if (NS_InputStreamIsBuffered(mAsyncStream)) {
    mAsyncStreamIsBuffered = true;
    return NS_OK;
  }

  nsCOMPtr<nsIInputStream> stream;
  nsresult rv = NS_NewBufferedInputStream(getter_AddRefs(stream),
                                          mAsyncStream.forget(), 4096);
  NS_ENSURE_SUCCESS(rv, rv);

  // A buffered inputStream must implement nsIAsyncInputStream.
  mAsyncStream = do_QueryInterface(stream);
  MOZ_DIAGNOSTIC_ASSERT(mAsyncStream);
  mAsyncStreamIsBuffered = true;

  return NS_OK;
}

//-----------------------------------------------------------------------------
// nsIThreadRetargetableRequest
//-----------------------------------------------------------------------------

NS_IMETHODIMP
nsInputStreamPump::RetargetDeliveryTo(nsISerialEventTarget* aNewTarget) {
  RecursiveMutexAutoLock lock(mMutex);

  NS_ENSURE_ARG(aNewTarget);
  NS_ENSURE_TRUE(mState == STATE_START || mState == STATE_TRANSFER,
                 NS_ERROR_UNEXPECTED);

  // If canceled, do not retarget. Return with canceled status.
  if (NS_FAILED(mStatus)) {
    return mStatus;
  }

  if (aNewTarget == mTargetThread) {
    NS_WARNING("Retargeting delivery to same thread");
    return NS_OK;
  }

  if (mOffMainThread) {
    // Don't support retargeting if this pump is already used off the main
    // thread.
    return NS_ERROR_FAILURE;
  }

  // Ensure that |mListener| and any subsequent listeners can be retargeted
  // to another thread.
  nsresult rv = NS_OK;
  nsCOMPtr<nsIThreadRetargetableStreamListener> retargetableListener =
      do_QueryInterface(mListener, &rv);
  if (NS_SUCCEEDED(rv) && retargetableListener) {
    rv = retargetableListener->CheckListenerChain();
    if (NS_SUCCEEDED(rv)) {
      mTargetThread = aNewTarget;
      mRetargeting = true;
    }
  }
  LOG(
      ("nsInputStreamPump::RetargetDeliveryTo [this=%p aNewTarget=%p] "
       "%s listener [%p] rv[%" PRIx32 "]",
       this, aNewTarget, (mTargetThread == aNewTarget ? "success" : "failure"),
       (nsIStreamListener*)mListener, static_cast<uint32_t>(rv)));
  return rv;
}

NS_IMETHODIMP
nsInputStreamPump::GetDeliveryTarget(nsISerialEventTarget** aNewTarget) {
  RecursiveMutexAutoLock lock(mMutex);

  nsCOMPtr<nsISerialEventTarget> target = mTargetThread;
  target.forget(aNewTarget);
  return NS_OK;
}