/* 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 "nsStreamTransportService.h"

#include "nsXPCOMCIDInternal.h"

#include "nsNetSegmentUtils.h"

#include "nsTransportUtils.h"

#include "nsStreamUtils.h"

#include "nsError.h"

#include "nsNetCID.h"

#include "nsIAsyncInputStream.h"

#include "nsIAsyncOutputStream.h"

#include "nsISeekableStream.h"

#include "nsIPipe.h"

#include "nsITransport.h"

#include "nsIObserverService.h"

#include "nsThreadPool.h"

#include "mozilla/Services.h"

namespace mozilla {

namespace net {

//-----------------------------------------------------------------------------

// nsInputStreamTransport

//

// Implements nsIInputStream as a wrapper around the real input stream.  This

// allows the transport to support seeking, range-limiting, progress reporting,

// and close-when-done semantics while utilizing NS_AsyncCopy.

//-----------------------------------------------------------------------------

class nsInputStreamTransport : public nsITransport

                             , public nsIInputStream

{

public:

    // Record refcount changes to ensure that stream transports are destroyed

    // on consistent threads when recording/replaying.

    NS_DECL_THREADSAFE_ISUPPORTS_WITH_RECORDING(recordreplay::Behavior::Preserve)

    NS_DECL_NSITRANSPORT

    NS_DECL_NSIINPUTSTREAM

    nsInputStreamTransport(nsIInputStream *source,

                           bool closeWhenDone)

        : mSource(source)

        , mOffset(0)

        , mCloseWhenDone(closeWhenDone)

        , mInProgress(false)

    {

    }

private:

    virtual ~nsInputStreamTransport() = default;

    nsCOMPtr<nsIAsyncInputStream>   mPipeIn;

    // while the copy is active, these members may only be accessed from the

    // nsIInputStream implementation.

    nsCOMPtr<nsITransportEventSink> mEventSink;

    nsCOMPtr<nsIInputStream>        mSource;

    int64_t                         mOffset;

    bool                            mCloseWhenDone;

    // this variable serves as a lock to prevent the state of the transport

    // from being modified once the copy is in progress.

    bool                            mInProgress;

};

NS_IMPL_ISUPPORTS(nsInputStreamTransport,

                  nsITransport,

                  nsIInputStream)

/** nsITransport **/

NS_IMETHODIMP

nsInputStreamTransport::OpenInputStream(uint32_t flags,

                                        uint32_t segsize,

                                        uint32_t segcount,

                                        nsIInputStream **result)

{

    NS_ENSURE_TRUE(!mInProgress, NS_ERROR_IN_PROGRESS);

    nsresult rv;

    nsCOMPtr<nsIEventTarget> target =

            do_GetService(NS_STREAMTRANSPORTSERVICE_CONTRACTID, &rv);

    if (NS_FAILED(rv)) return rv;

    // XXX if the caller requests an unbuffered stream, then perhaps

    //     we'd want to simply return mSource; however, then we would

    //     not be reading mSource on a background thread.  is this ok?

    bool nonblocking = !(flags & OPEN_BLOCKING);

    net_ResolveSegmentParams(segsize, segcount);

    nsCOMPtr<nsIAsyncOutputStream> pipeOut;

    rv = NS_NewPipe2(getter_AddRefs(mPipeIn),

                     getter_AddRefs(pipeOut),

                     nonblocking, true,

                     segsize, segcount);

    if (NS_FAILED(rv)) return rv;

    mInProgress = true;

    // startup async copy process...

    rv = NS_AsyncCopy(this, pipeOut, target,

                      NS_ASYNCCOPY_VIA_WRITESEGMENTS, segsize);

    if (NS_SUCCEEDED(rv))

        NS_ADDREF(*result = mPipeIn);

    return rv;

}

NS_IMETHODIMP

nsInputStreamTransport::OpenOutputStream(uint32_t flags,

                                         uint32_t segsize,

                                         uint32_t segcount,

                                         nsIOutputStream **result)

{

    // this transport only supports reading!

    MOZ_ASSERT_UNREACHABLE("nsInputStreamTransport::OpenOutputStream");

    return NS_ERROR_UNEXPECTED;

}

NS_IMETHODIMP

nsInputStreamTransport::Close(nsresult reason)

{

    if (NS_SUCCEEDED(reason))

        reason = NS_BASE_STREAM_CLOSED;

    return mPipeIn->CloseWithStatus(reason);

}

NS_IMETHODIMP

nsInputStreamTransport::SetEventSink(nsITransportEventSink *sink,

                                     nsIEventTarget *target)

{

    NS_ENSURE_TRUE(!mInProgress, NS_ERROR_IN_PROGRESS);

    if (target)

        return net_NewTransportEventSinkProxy(getter_AddRefs(mEventSink),

                                              sink, target);

    mEventSink = sink;

    return NS_OK;

}

/** nsIInputStream **/

NS_IMETHODIMP

nsInputStreamTransport::Close()

{

    if (mCloseWhenDone)

        mSource->Close();

    // make additional reads return early...

    mOffset = 0;

    return NS_OK;

}

NS_IMETHODIMP

nsInputStreamTransport::Available(uint64_t *result)

{

    return NS_ERROR_NOT_IMPLEMENTED;

}

NS_IMETHODIMP

nsInputStreamTransport::Read(char *buf, uint32_t count, uint32_t *result)

{

    nsresult rv = mSource->Read(buf, count, result);

    if (NS_SUCCEEDED(rv)) {

        mOffset += *result;

        if (mEventSink)

            mEventSink->OnTransportStatus(this, NS_NET_STATUS_READING, mOffset,

                                          -1);

    }

    return rv;

}

NS_IMETHODIMP

nsInputStreamTransport::ReadSegments(nsWriteSegmentFun writer, void *closure,

                                     uint32_t count, uint32_t *result)

{

    return NS_ERROR_NOT_IMPLEMENTED;

}

NS_IMETHODIMP

nsInputStreamTransport::IsNonBlocking(bool *result)

{

    *result = false;

    return NS_OK;

}

//-----------------------------------------------------------------------------

// nsStreamTransportService

//-----------------------------------------------------------------------------

nsStreamTransportService::~nsStreamTransportService()

{

    NS_ASSERTION(!mPool, "thread pool wasn't shutdown");

}

nsresult

nsStreamTransportService::Init()

{

    mPool = new nsThreadPool();

    // Configure the pool

    mPool->SetName(NS_LITERAL_CSTRING("StreamTrans"));

    mPool->SetThreadLimit(25);

    mPool->SetIdleThreadLimit(1);

    mPool->SetIdleThreadTimeout(PR_SecondsToInterval(30));

    nsCOMPtr<nsIObserverService> obsSvc =

        mozilla::services::GetObserverService();

    if (obsSvc)

        obsSvc->AddObserver(this, "xpcom-shutdown-threads", false);

    return NS_OK;

}

NS_IMPL_ISUPPORTS(nsStreamTransportService,

                  nsIStreamTransportService,

                  nsIEventTarget,

                  nsIObserver)

NS_IMETHODIMP

nsStreamTransportService::DispatchFromScript(nsIRunnable *task, uint32_t flags)

{

  nsCOMPtr<nsIRunnable> event(task);

  return Dispatch(event.forget(), flags);

}

NS_IMETHODIMP

nsStreamTransportService::Dispatch(already_AddRefed<nsIRunnable> task, uint32_t flags)

{

    nsCOMPtr<nsIRunnable> event(task); // so it gets released on failure paths

    nsCOMPtr<nsIThreadPool> pool;

    {

        mozilla::MutexAutoLock lock(mShutdownLock);

        if (mIsShutdown) {

            return NS_ERROR_NOT_INITIALIZED;

        }

        pool = mPool;

    }

    NS_ENSURE_TRUE(pool, NS_ERROR_NOT_INITIALIZED);

    return pool->Dispatch(event.forget(), flags);

}

NS_IMETHODIMP

nsStreamTransportService::DelayedDispatch(already_AddRefed<nsIRunnable>, uint32_t)

{

    return NS_ERROR_NOT_IMPLEMENTED;

}

NS_IMETHODIMP_(bool)

nsStreamTransportService::IsOnCurrentThreadInfallible()

{

    nsCOMPtr<nsIThreadPool> pool;

    {

        mozilla::MutexAutoLock lock(mShutdownLock);

        pool = mPool;

    }

    if (!pool) {

      return false;

    }

    return pool->IsOnCurrentThread();

}

NS_IMETHODIMP

nsStreamTransportService::IsOnCurrentThread(bool *result)

{

    nsCOMPtr<nsIThreadPool> pool;

    {

        mozilla::MutexAutoLock lock(mShutdownLock);

        if (mIsShutdown) {

            return NS_ERROR_NOT_INITIALIZED;

        }

        pool = mPool;

    }

    NS_ENSURE_TRUE(pool, NS_ERROR_NOT_INITIALIZED);

    return pool->IsOnCurrentThread(result);

}

NS_IMETHODIMP

nsStreamTransportService::CreateInputTransport(nsIInputStream *stream,

                                               bool closeWhenDone,

                                               nsITransport **result)

{

    nsInputStreamTransport *trans =

        new nsInputStreamTransport(stream, closeWhenDone);

    if (!trans)

        return NS_ERROR_OUT_OF_MEMORY;

    NS_ADDREF(*result = trans);

    return NS_OK;

}

NS_IMETHODIMP

nsStreamTransportService::Observe(nsISupports *subject, const char *topic,

                                  const char16_t *data)

{

  NS_ASSERTION(strcmp(topic, "xpcom-shutdown-threads") == 0, "oops");

  {

    mozilla::MutexAutoLock lock(mShutdownLock);

    mIsShutdown = true;

  }

  if (mPool) {

    mPool->Shutdown();

    mPool = nullptr;

  }

  return NS_OK;

}

class AvailableEvent final : public Runnable

{

public:

  AvailableEvent(nsIInputStream* stream,

                 nsIInputAvailableCallback* callback)

    : Runnable("net::AvailableEvent")

    , mStream(stream)

    , mCallback(callback)

    , mDoingCallback(false)

    , mSize(0)

    , mResultForCallback(NS_OK)

  {

    mCallbackTarget = GetCurrentThreadEventTarget();

  }

    NS_IMETHOD Run() override

    {

        if (mDoingCallback) {

            // pong

            mCallback->OnInputAvailableComplete(mSize, mResultForCallback);

            mCallback = nullptr;

        } else {

            // ping

            mResultForCallback = mStream->Available(&mSize);

            mStream = nullptr;

            mDoingCallback = true;

            nsCOMPtr<nsIRunnable> event(this); // overly cute

            mCallbackTarget->Dispatch(event.forget(), NS_DISPATCH_NORMAL);

            mCallbackTarget = nullptr;

        }

        return NS_OK;

    }

private:

    virtual ~AvailableEvent() = default;

    nsCOMPtr<nsIInputStream> mStream;

    nsCOMPtr<nsIInputAvailableCallback> mCallback;

    nsCOMPtr<nsIEventTarget> mCallbackTarget;

    bool mDoingCallback;

    uint64_t mSize;

    nsresult mResultForCallback;

};

NS_IMETHODIMP

nsStreamTransportService::InputAvailable(nsIInputStream *stream,

                                         nsIInputAvailableCallback *callback)

{

    nsCOMPtr<nsIThreadPool> pool;

    {

        mozilla::MutexAutoLock lock(mShutdownLock);

        if (mIsShutdown) {

            return NS_ERROR_NOT_INITIALIZED;

        }

        pool = mPool;

    }

    nsCOMPtr<nsIRunnable> event = new AvailableEvent(stream, callback);

    return pool->Dispatch(event.forget(), NS_DISPATCH_NORMAL);

}

} // namespace net

} // namespace mozilla