michael@0: /* This Source Code Form is subject to the terms of the Mozilla Public
michael@0:  * License, v. 2.0. If a copy of the MPL was not distributed with this
michael@0:  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
michael@0: 
michael@0: #include "mozilla/Attributes.h"
michael@0: #include "mozilla/ReentrantMonitor.h"
michael@0: #include "nsIPipe.h"
michael@0: #include "nsIEventTarget.h"
michael@0: #include "nsISeekableStream.h"
michael@0: #include "nsIProgrammingLanguage.h"
michael@0: #include "nsSegmentedBuffer.h"
michael@0: #include "nsStreamUtils.h"
michael@0: #include "nsCOMPtr.h"
michael@0: #include "nsCRT.h"
michael@0: #include "prlog.h"
michael@0: #include "nsIClassInfoImpl.h"
michael@0: #include "nsAlgorithm.h"
michael@0: #include "nsMemory.h"
michael@0: #include "nsIAsyncInputStream.h"
michael@0: #include "nsIAsyncOutputStream.h"
michael@0: 
michael@0: using namespace mozilla;
michael@0: 
michael@0: #ifdef LOG
michael@0: #undef LOG
michael@0: #endif
michael@0: #if defined(PR_LOGGING)
michael@0: //
michael@0: // set NSPR_LOG_MODULES=nsPipe:5
michael@0: //
michael@0: static PRLogModuleInfo *
michael@0: GetPipeLog()
michael@0: {
michael@0:   static PRLogModuleInfo *sLog;
michael@0:   if (!sLog)
michael@0:     sLog = PR_NewLogModule("nsPipe");
michael@0:   return sLog;
michael@0: }
michael@0: #define LOG(args) PR_LOG(GetPipeLog(), PR_LOG_DEBUG, args)
michael@0: #else
michael@0: #define LOG(args)
michael@0: #endif
michael@0: 
michael@0: #define DEFAULT_SEGMENT_SIZE  4096
michael@0: #define DEFAULT_SEGMENT_COUNT 16
michael@0: 
michael@0: class nsPipe;
michael@0: class nsPipeEvents;
michael@0: class nsPipeInputStream;
michael@0: class nsPipeOutputStream;
michael@0: 
michael@0: //-----------------------------------------------------------------------------
michael@0: 
michael@0: // this class is used to delay notifications until the end of a particular
michael@0: // scope.  it helps avoid the complexity of issuing callbacks while inside
michael@0: // a critical section.
michael@0: class nsPipeEvents
michael@0: {
michael@0: public:
michael@0:     nsPipeEvents() { }
michael@0:    ~nsPipeEvents();
michael@0: 
michael@0:     inline void NotifyInputReady(nsIAsyncInputStream *stream,
michael@0:                                  nsIInputStreamCallback *callback)
michael@0:     {
michael@0:         NS_ASSERTION(!mInputCallback, "already have an input event");
michael@0:         mInputStream = stream;
michael@0:         mInputCallback = callback;
michael@0:     }
michael@0: 
michael@0:     inline void NotifyOutputReady(nsIAsyncOutputStream *stream,
michael@0:                                   nsIOutputStreamCallback *callback)
michael@0:     {
michael@0:         NS_ASSERTION(!mOutputCallback, "already have an output event");
michael@0:         mOutputStream = stream;
michael@0:         mOutputCallback = callback;
michael@0:     }
michael@0: 
michael@0: private:
michael@0:     nsCOMPtr<nsIAsyncInputStream>     mInputStream;
michael@0:     nsCOMPtr<nsIInputStreamCallback>  mInputCallback;
michael@0:     nsCOMPtr<nsIAsyncOutputStream>    mOutputStream;
michael@0:     nsCOMPtr<nsIOutputStreamCallback> mOutputCallback;
michael@0: };
michael@0: 
michael@0: //-----------------------------------------------------------------------------
michael@0: 
michael@0: // the input end of a pipe (allocated as a member of the pipe).
michael@0: class nsPipeInputStream : public nsIAsyncInputStream
michael@0:                         , public nsISeekableStream
michael@0:                         , public nsISearchableInputStream
michael@0:                         , public nsIClassInfo
michael@0: {
michael@0: public:
michael@0:     // since this class will be allocated as a member of the pipe, we do not
michael@0:     // need our own ref count.  instead, we share the lifetime (the ref count)
michael@0:     // of the entire pipe.  this macro is just convenience since it does not
michael@0:     // declare a mRefCount variable; however, don't let the name fool you...
michael@0:     // we are not inheriting from nsPipe ;-)
michael@0:     NS_DECL_ISUPPORTS_INHERITED
michael@0: 
michael@0:     NS_DECL_NSIINPUTSTREAM
michael@0:     NS_DECL_NSIASYNCINPUTSTREAM
michael@0:     NS_DECL_NSISEEKABLESTREAM
michael@0:     NS_DECL_NSISEARCHABLEINPUTSTREAM
michael@0:     NS_DECL_NSICLASSINFO
michael@0: 
michael@0:     nsPipeInputStream(nsPipe *pipe)
michael@0:         : mPipe(pipe)
michael@0:         , mReaderRefCnt(0)
michael@0:         , mLogicalOffset(0)
michael@0:         , mBlocking(true)
michael@0:         , mBlocked(false)
michael@0:         , mAvailable(0)
michael@0:         , mCallbackFlags(0)
michael@0:         { }
michael@0: 
michael@0:     nsresult Fill();
michael@0:     void SetNonBlocking(bool aNonBlocking) { mBlocking = !aNonBlocking; }
michael@0: 
michael@0:     uint32_t Available() { return mAvailable; }
michael@0:     void     ReduceAvailable(uint32_t avail) { mAvailable -= avail; }
michael@0: 
michael@0:     // synchronously wait for the pipe to become readable.
michael@0:     nsresult Wait();
michael@0: 
michael@0:     // these functions return true to indicate that the pipe's monitor should
michael@0:     // be notified, to wake up a blocked reader if any.
michael@0:     bool     OnInputReadable(uint32_t bytesWritten, nsPipeEvents &);
michael@0:     bool     OnInputException(nsresult, nsPipeEvents &);
michael@0: 
michael@0: private:
michael@0:     nsPipe                        *mPipe;
michael@0: 
michael@0:     // separate refcnt so that we know when to close the consumer
michael@0:     mozilla::ThreadSafeAutoRefCnt  mReaderRefCnt;
michael@0:     int64_t                        mLogicalOffset;
michael@0:     bool                           mBlocking;
michael@0: 
michael@0:     // these variables can only be accessed while inside the pipe's monitor
michael@0:     bool                           mBlocked;
michael@0:     uint32_t                       mAvailable;
michael@0:     nsCOMPtr<nsIInputStreamCallback> mCallback;
michael@0:     uint32_t                       mCallbackFlags;
michael@0: };
michael@0: 
michael@0: //-----------------------------------------------------------------------------
michael@0: 
michael@0: // the output end of a pipe (allocated as a member of the pipe).
michael@0: class nsPipeOutputStream : public nsIAsyncOutputStream
michael@0:                          , public nsIClassInfo
michael@0: {
michael@0: public:
michael@0:     // since this class will be allocated as a member of the pipe, we do not
michael@0:     // need our own ref count.  instead, we share the lifetime (the ref count)
michael@0:     // of the entire pipe.  this macro is just convenience since it does not
michael@0:     // declare a mRefCount variable; however, don't let the name fool you...
michael@0:     // we are not inheriting from nsPipe ;-)
michael@0:     NS_DECL_ISUPPORTS_INHERITED
michael@0: 
michael@0:     NS_DECL_NSIOUTPUTSTREAM
michael@0:     NS_DECL_NSIASYNCOUTPUTSTREAM
michael@0:     NS_DECL_NSICLASSINFO
michael@0: 
michael@0:     nsPipeOutputStream(nsPipe *pipe)
michael@0:         : mPipe(pipe)
michael@0:         , mWriterRefCnt(0)
michael@0:         , mLogicalOffset(0)
michael@0:         , mBlocking(true)
michael@0:         , mBlocked(false)
michael@0:         , mWritable(true)
michael@0:         , mCallbackFlags(0)
michael@0:         { }
michael@0: 
michael@0:     void SetNonBlocking(bool aNonBlocking) { mBlocking = !aNonBlocking; }
michael@0:     void SetWritable(bool writable) { mWritable = writable; }
michael@0: 
michael@0:     // synchronously wait for the pipe to become writable.
michael@0:     nsresult Wait();
michael@0: 
michael@0:     // these functions return true to indicate that the pipe's monitor should
michael@0:     // be notified, to wake up a blocked writer if any.
michael@0:     bool     OnOutputWritable(nsPipeEvents &);
michael@0:     bool     OnOutputException(nsresult, nsPipeEvents &);
michael@0: 
michael@0: private:
michael@0:     nsPipe                         *mPipe;
michael@0: 
michael@0:     // separate refcnt so that we know when to close the producer
michael@0:     mozilla::ThreadSafeAutoRefCnt   mWriterRefCnt;
michael@0:     int64_t                         mLogicalOffset;
michael@0:     bool                            mBlocking;
michael@0: 
michael@0:     // these variables can only be accessed while inside the pipe's monitor
michael@0:     bool                            mBlocked;
michael@0:     bool                            mWritable;
michael@0:     nsCOMPtr<nsIOutputStreamCallback> mCallback;
michael@0:     uint32_t                        mCallbackFlags;
michael@0: };
michael@0: 
michael@0: //-----------------------------------------------------------------------------
michael@0: 
michael@0: class nsPipe MOZ_FINAL : public nsIPipe
michael@0: {
michael@0: public:
michael@0:     friend class nsPipeInputStream;
michael@0:     friend class nsPipeOutputStream;
michael@0: 
michael@0:     NS_DECL_THREADSAFE_ISUPPORTS
michael@0:     NS_DECL_NSIPIPE
michael@0: 
michael@0:     // nsPipe methods:
michael@0:     nsPipe();
michael@0: 
michael@0: private:
michael@0:     ~nsPipe();
michael@0: 
michael@0: public:
michael@0:     //
michael@0:     // methods below may only be called while inside the pipe's monitor
michael@0:     //
michael@0: 
michael@0:     void PeekSegment(uint32_t n, char *&cursor, char *&limit);
michael@0: 
michael@0:     //
michael@0:     // methods below may be called while outside the pipe's monitor
michael@0:     //
michael@0:  
michael@0:     nsresult GetReadSegment(const char *&segment, uint32_t &segmentLen);
michael@0:     void     AdvanceReadCursor(uint32_t count);
michael@0: 
michael@0:     nsresult GetWriteSegment(char *&segment, uint32_t &segmentLen);
michael@0:     void     AdvanceWriteCursor(uint32_t count);
michael@0: 
michael@0:     void     OnPipeException(nsresult reason, bool outputOnly = false);
michael@0: 
michael@0: protected:
michael@0:     // We can't inherit from both nsIInputStream and nsIOutputStream
michael@0:     // because they collide on their Close method. Consequently we nest their
michael@0:     // implementations to avoid the extra object allocation.
michael@0:     nsPipeInputStream   mInput;
michael@0:     nsPipeOutputStream  mOutput;
michael@0: 
michael@0:     ReentrantMonitor    mReentrantMonitor;
michael@0:     nsSegmentedBuffer   mBuffer;
michael@0: 
michael@0:     char*               mReadCursor;
michael@0:     char*               mReadLimit;
michael@0: 
michael@0:     int32_t             mWriteSegment;
michael@0:     char*               mWriteCursor;
michael@0:     char*               mWriteLimit;
michael@0: 
michael@0:     nsresult            mStatus;
michael@0:     bool                mInited;
michael@0: };
michael@0: 
michael@0: //
michael@0: // NOTES on buffer architecture:
michael@0: //
michael@0: //       +-----------------+ - - mBuffer.GetSegment(0)
michael@0: //       |                 |
michael@0: //       + - - - - - - - - + - - mReadCursor
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       +-----------------+ - - mReadLimit
michael@0: //                |
michael@0: //       +-----------------+
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       +-----------------+
michael@0: //                |
michael@0: //       +-----------------+ - - mBuffer.GetSegment(mWriteSegment)
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       |/////////////////|
michael@0: //       + - - - - - - - - + - - mWriteCursor
michael@0: //       |                 |
michael@0: //       |                 |
michael@0: //       +-----------------+ - - mWriteLimit
michael@0: //
michael@0: // (shaded region contains data)
michael@0: //
michael@0: // NOTE: on some systems (notably OS/2), the heap allocator uses an arena for
michael@0: // small allocations (e.g., 64 byte allocations).  this means that buffers may
michael@0: // be allocated back-to-back.  in the diagram above, for example, mReadLimit
michael@0: // would actually be pointing at the beginning of the next segment.  when
michael@0: // making changes to this file, please keep this fact in mind.
michael@0: //
michael@0: 
michael@0: //-----------------------------------------------------------------------------
michael@0: // nsPipe methods:
michael@0: //-----------------------------------------------------------------------------
michael@0: 
michael@0: nsPipe::nsPipe()
michael@0:     : mInput(MOZ_THIS_IN_INITIALIZER_LIST())
michael@0:     , mOutput(MOZ_THIS_IN_INITIALIZER_LIST())
michael@0:     , mReentrantMonitor("nsPipe.mReentrantMonitor")
michael@0:     , mReadCursor(nullptr)
michael@0:     , mReadLimit(nullptr)
michael@0:     , mWriteSegment(-1)
michael@0:     , mWriteCursor(nullptr)
michael@0:     , mWriteLimit(nullptr)
michael@0:     , mStatus(NS_OK)
michael@0:     , mInited(false)
michael@0: {
michael@0: }
michael@0: 
michael@0: nsPipe::~nsPipe()
michael@0: {
michael@0: }
michael@0: 
michael@0: NS_IMPL_ISUPPORTS(nsPipe, nsIPipe)
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipe::Init(bool nonBlockingIn,
michael@0:              bool nonBlockingOut,
michael@0:              uint32_t segmentSize,
michael@0:              uint32_t segmentCount)
michael@0: {
michael@0:     mInited = true;
michael@0: 
michael@0:     if (segmentSize == 0)
michael@0:         segmentSize = DEFAULT_SEGMENT_SIZE;
michael@0:     if (segmentCount == 0)
michael@0:         segmentCount = DEFAULT_SEGMENT_COUNT;
michael@0: 
michael@0:     // protect against overflow
michael@0:     uint32_t maxCount = uint32_t(-1) / segmentSize;
michael@0:     if (segmentCount > maxCount)
michael@0:         segmentCount = maxCount;
michael@0: 
michael@0:     nsresult rv = mBuffer.Init(segmentSize, segmentSize * segmentCount);
michael@0:     if (NS_FAILED(rv))
michael@0:         return rv;
michael@0: 
michael@0:     mInput.SetNonBlocking(nonBlockingIn);
michael@0:     mOutput.SetNonBlocking(nonBlockingOut);
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipe::GetInputStream(nsIAsyncInputStream **aInputStream)
michael@0: {
michael@0:     NS_ADDREF(*aInputStream = &mInput);
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipe::GetOutputStream(nsIAsyncOutputStream **aOutputStream)
michael@0: {
michael@0:     if (NS_WARN_IF(!mInited))
michael@0: 	return NS_ERROR_NOT_INITIALIZED;
michael@0:     NS_ADDREF(*aOutputStream = &mOutput);
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: void
michael@0: nsPipe::PeekSegment(uint32_t index, char *&cursor, char *&limit)
michael@0: {
michael@0:     if (index == 0) {
michael@0:         NS_ASSERTION(!mReadCursor || mBuffer.GetSegmentCount(), "unexpected state");
michael@0:         cursor = mReadCursor;
michael@0:         limit = mReadLimit;
michael@0:     }
michael@0:     else {
michael@0:         uint32_t numSegments = mBuffer.GetSegmentCount();
michael@0:         if (index >= numSegments)
michael@0:             cursor = limit = nullptr;
michael@0:         else {
michael@0:             cursor = mBuffer.GetSegment(index);
michael@0:             if (mWriteSegment == (int32_t) index)
michael@0:                 limit = mWriteCursor;
michael@0:             else
michael@0:                 limit = cursor + mBuffer.GetSegmentSize();
michael@0:         }
michael@0:     }
michael@0: }
michael@0: 
michael@0: nsresult
michael@0: nsPipe::GetReadSegment(const char *&segment, uint32_t &segmentLen)
michael@0: {
michael@0:     ReentrantMonitorAutoEnter mon(mReentrantMonitor);
michael@0: 
michael@0:     if (mReadCursor == mReadLimit)
michael@0:         return NS_FAILED(mStatus) ? mStatus : NS_BASE_STREAM_WOULD_BLOCK;
michael@0: 
michael@0:     segment    = mReadCursor;
michael@0:     segmentLen = mReadLimit - mReadCursor;
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: void
michael@0: nsPipe::AdvanceReadCursor(uint32_t bytesRead)
michael@0: {
michael@0:     NS_ASSERTION(bytesRead, "don't call if no bytes read");
michael@0: 
michael@0:     nsPipeEvents events;
michael@0:     {
michael@0:         ReentrantMonitorAutoEnter mon(mReentrantMonitor);
michael@0: 
michael@0:         LOG(("III advancing read cursor by %u\n", bytesRead));
michael@0:         NS_ASSERTION(bytesRead <= mBuffer.GetSegmentSize(), "read too much");
michael@0: 
michael@0:         mReadCursor += bytesRead;
michael@0:         NS_ASSERTION(mReadCursor <= mReadLimit, "read cursor exceeds limit");
michael@0: 
michael@0:         mInput.ReduceAvailable(bytesRead);
michael@0: 
michael@0:         if (mReadCursor == mReadLimit) {
michael@0:             // we've reached the limit of how much we can read from this segment.
michael@0:             // if at the end of this segment, then we must discard this segment.
michael@0: 
michael@0:             // if still writing in this segment then bail because we're not done
michael@0:             // with the segment and have to wait for now...
michael@0:             if (mWriteSegment == 0 && mWriteLimit > mWriteCursor) {
michael@0:                 NS_ASSERTION(mReadLimit == mWriteCursor, "unexpected state");
michael@0:                 return;
michael@0:             }
michael@0: 
michael@0:             // shift write segment index (-1 indicates an empty buffer).
michael@0:             --mWriteSegment;
michael@0: 
michael@0:             // done with this segment
michael@0:             mBuffer.DeleteFirstSegment();
michael@0:             LOG(("III deleting first segment\n"));
michael@0: 
michael@0:             if (mWriteSegment == -1) {
michael@0:                 // buffer is completely empty
michael@0:                 mReadCursor = nullptr;
michael@0:                 mReadLimit = nullptr;
michael@0:                 mWriteCursor = nullptr;
michael@0:                 mWriteLimit = nullptr;
michael@0:             }
michael@0:             else {
michael@0:                 // advance read cursor and limit to next buffer segment
michael@0:                 mReadCursor = mBuffer.GetSegment(0);
michael@0:                 if (mWriteSegment == 0)
michael@0:                     mReadLimit = mWriteCursor;
michael@0:                 else
michael@0:                     mReadLimit = mReadCursor + mBuffer.GetSegmentSize();
michael@0:             }
michael@0: 
michael@0:             // we've free'd up a segment, so notify output stream that pipe has
michael@0:             // room for a new segment.
michael@0:             if (mOutput.OnOutputWritable(events))
michael@0:                 mon.Notify();
michael@0:         }
michael@0:     }
michael@0: }
michael@0: 
michael@0: nsresult
michael@0: nsPipe::GetWriteSegment(char *&segment, uint32_t &segmentLen)
michael@0: {
michael@0:     ReentrantMonitorAutoEnter mon(mReentrantMonitor);
michael@0: 
michael@0:     if (NS_FAILED(mStatus))
michael@0:         return mStatus;
michael@0: 
michael@0:     // write cursor and limit may both be null indicating an empty buffer.
michael@0:     if (mWriteCursor == mWriteLimit) {
michael@0:         char *seg = mBuffer.AppendNewSegment();
michael@0:         // pipe is full
michael@0:         if (seg == nullptr)
michael@0:             return NS_BASE_STREAM_WOULD_BLOCK;
michael@0:         LOG(("OOO appended new segment\n"));
michael@0:         mWriteCursor = seg;
michael@0:         mWriteLimit = mWriteCursor + mBuffer.GetSegmentSize();
michael@0:         ++mWriteSegment;
michael@0:     }
michael@0: 
michael@0:     // make sure read cursor is initialized
michael@0:     if (mReadCursor == nullptr) {
michael@0:         NS_ASSERTION(mWriteSegment == 0, "unexpected null read cursor");
michael@0:         mReadCursor = mReadLimit = mWriteCursor;
michael@0:     }
michael@0: 
michael@0:     // check to see if we can roll-back our read and write cursors to the 
michael@0:     // beginning of the current/first segment.  this is purely an optimization.
michael@0:     if (mReadCursor == mWriteCursor && mWriteSegment == 0) {
michael@0:         char *head = mBuffer.GetSegment(0);
michael@0:         LOG(("OOO rolling back write cursor %u bytes\n", mWriteCursor - head));
michael@0:         mWriteCursor = mReadCursor = mReadLimit = head;
michael@0:     }
michael@0: 
michael@0:     segment    = mWriteCursor;
michael@0:     segmentLen = mWriteLimit - mWriteCursor;
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: void
michael@0: nsPipe::AdvanceWriteCursor(uint32_t bytesWritten)
michael@0: {
michael@0:     NS_ASSERTION(bytesWritten, "don't call if no bytes written");
michael@0: 
michael@0:     nsPipeEvents events;
michael@0:     {
michael@0:         ReentrantMonitorAutoEnter mon(mReentrantMonitor);
michael@0: 
michael@0:         LOG(("OOO advancing write cursor by %u\n", bytesWritten));
michael@0: 
michael@0:         char *newWriteCursor = mWriteCursor + bytesWritten;
michael@0:         NS_ASSERTION(newWriteCursor <= mWriteLimit, "write cursor exceeds limit");
michael@0: 
michael@0:         // update read limit if reading in the same segment
michael@0:         if (mWriteSegment == 0 && mReadLimit == mWriteCursor)
michael@0:             mReadLimit = newWriteCursor;
michael@0: 
michael@0:         mWriteCursor = newWriteCursor;
michael@0: 
michael@0:         // The only way mReadCursor == mWriteCursor is if:
michael@0:         //
michael@0:         // - mReadCursor is at the start of a segment (which, based on how
michael@0:         //   nsSegmentedBuffer works, means that this segment is the "first"
michael@0:         //   segment)
michael@0:         // - mWriteCursor points at the location past the end of the current
michael@0:         //   write segment (so the current write filled the current write
michael@0:         //   segment, so we've incremented mWriteCursor to point past the end
michael@0:         //   of it)
michael@0:         // - the segment to which data has just been written is located
michael@0:         //   exactly one segment's worth of bytes before the first segment
michael@0:         //   where mReadCursor is located
michael@0:         //
michael@0:         // Consequently, the byte immediately after the end of the current
michael@0:         // write segment is the first byte of the first segment, so
michael@0:         // mReadCursor == mWriteCursor.  (Another way to think about this is
michael@0:         // to consider the buffer architecture diagram above, but consider it
michael@0:         // with an arena allocator which allocates from the *end* of the
michael@0:         // arena to the *beginning* of the arena.)
michael@0:         NS_ASSERTION(mReadCursor != mWriteCursor ||
michael@0:                      (mBuffer.GetSegment(0) == mReadCursor &&
michael@0:                       mWriteCursor == mWriteLimit),
michael@0:                      "read cursor is bad");
michael@0: 
michael@0:         // update the writable flag on the output stream
michael@0:         if (mWriteCursor == mWriteLimit) {
michael@0:             if (mBuffer.GetSize() >= mBuffer.GetMaxSize())
michael@0:                 mOutput.SetWritable(false);
michael@0:         }
michael@0: 
michael@0:         // notify input stream that pipe now contains additional data
michael@0:         if (mInput.OnInputReadable(bytesWritten, events))
michael@0:             mon.Notify();
michael@0:     }
michael@0: }
michael@0: 
michael@0: void
michael@0: nsPipe::OnPipeException(nsresult reason, bool outputOnly)
michael@0: {
michael@0:     LOG(("PPP nsPipe::OnPipeException [reason=%x output-only=%d]\n",
michael@0:         reason, outputOnly));
michael@0: 
michael@0:     nsPipeEvents events;
michael@0:     {
michael@0:         ReentrantMonitorAutoEnter mon(mReentrantMonitor);
michael@0: 
michael@0:         // if we've already hit an exception, then ignore this one.
michael@0:         if (NS_FAILED(mStatus))
michael@0:             return;
michael@0: 
michael@0:         mStatus = reason;
michael@0: 
michael@0:         // an output-only exception applies to the input end if the pipe has
michael@0:         // zero bytes available.
michael@0:         if (outputOnly && !mInput.Available())
michael@0:             outputOnly = false;
michael@0: 
michael@0:         if (!outputOnly)
michael@0:             if (mInput.OnInputException(reason, events))
michael@0:                 mon.Notify();
michael@0: 
michael@0:         if (mOutput.OnOutputException(reason, events))
michael@0:             mon.Notify();
michael@0:     }
michael@0: }
michael@0: 
michael@0: //-----------------------------------------------------------------------------
michael@0: // nsPipeEvents methods:
michael@0: //-----------------------------------------------------------------------------
michael@0: 
michael@0: nsPipeEvents::~nsPipeEvents()
michael@0: {
michael@0:     // dispatch any pending events
michael@0: 
michael@0:     if (mInputCallback) {
michael@0:         mInputCallback->OnInputStreamReady(mInputStream);
michael@0:         mInputCallback = 0;
michael@0:         mInputStream = 0;
michael@0:     }
michael@0:     if (mOutputCallback) {
michael@0:         mOutputCallback->OnOutputStreamReady(mOutputStream);
michael@0:         mOutputCallback = 0;
michael@0:         mOutputStream = 0;
michael@0:     }
michael@0: }
michael@0: 
michael@0: //-----------------------------------------------------------------------------
michael@0: // nsPipeInputStream methods:
michael@0: //-----------------------------------------------------------------------------
michael@0: 
michael@0: NS_IMPL_QUERY_INTERFACE(nsPipeInputStream,
michael@0:                         nsIInputStream,
michael@0:                         nsIAsyncInputStream,
michael@0:                         nsISeekableStream,
michael@0:                         nsISearchableInputStream,
michael@0:                         nsIClassInfo)
michael@0: 
michael@0: NS_IMPL_CI_INTERFACE_GETTER(nsPipeInputStream,
michael@0:                             nsIInputStream,
michael@0:                             nsIAsyncInputStream,
michael@0:                             nsISeekableStream,
michael@0:                             nsISearchableInputStream)
michael@0: 
michael@0: NS_IMPL_THREADSAFE_CI(nsPipeInputStream)
michael@0: 
michael@0: nsresult
michael@0: nsPipeInputStream::Wait()
michael@0: {
michael@0:     NS_ASSERTION(mBlocking, "wait on non-blocking pipe input stream");
michael@0: 
michael@0:     ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
michael@0: 
michael@0:     while (NS_SUCCEEDED(mPipe->mStatus) && (mAvailable == 0)) {
michael@0:         LOG(("III pipe input: waiting for data\n"));
michael@0: 
michael@0:         mBlocked = true;
michael@0:         mon.Wait();
michael@0:         mBlocked = false;
michael@0: 
michael@0:         LOG(("III pipe input: woke up [pipe-status=%x available=%u]\n",
michael@0:             mPipe->mStatus, mAvailable));
michael@0:     }
michael@0: 
michael@0:     return mPipe->mStatus == NS_BASE_STREAM_CLOSED ? NS_OK : mPipe->mStatus;
michael@0: }
michael@0: 
michael@0: bool
michael@0: nsPipeInputStream::OnInputReadable(uint32_t bytesWritten, nsPipeEvents &events)
michael@0: {
michael@0:     bool result = false;
michael@0: 
michael@0:     mAvailable += bytesWritten;
michael@0: 
michael@0:     if (mCallback && !(mCallbackFlags & WAIT_CLOSURE_ONLY)) {
michael@0:         events.NotifyInputReady(this, mCallback);
michael@0:         mCallback = 0;
michael@0:         mCallbackFlags = 0;
michael@0:     }
michael@0:     else if (mBlocked)
michael@0:         result = true;
michael@0: 
michael@0:     return result;
michael@0: }
michael@0: 
michael@0: bool
michael@0: nsPipeInputStream::OnInputException(nsresult reason, nsPipeEvents &events)
michael@0: {
michael@0:     LOG(("nsPipeInputStream::OnInputException [this=%x reason=%x]\n",
michael@0:         this, reason));
michael@0: 
michael@0:     bool result = false;
michael@0: 
michael@0:     NS_ASSERTION(NS_FAILED(reason), "huh? successful exception");
michael@0: 
michael@0:     // force count of available bytes to zero.
michael@0:     mAvailable = 0;
michael@0: 
michael@0:     if (mCallback) {
michael@0:         events.NotifyInputReady(this, mCallback);
michael@0:         mCallback = 0;
michael@0:         mCallbackFlags = 0;
michael@0:     }
michael@0:     else if (mBlocked)
michael@0:         result = true;
michael@0: 
michael@0:     return result;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP_(MozExternalRefCountType)
michael@0: nsPipeInputStream::AddRef(void)
michael@0: {
michael@0:     ++mReaderRefCnt;
michael@0:     return mPipe->AddRef();
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP_(MozExternalRefCountType)
michael@0: nsPipeInputStream::Release(void)
michael@0: {
michael@0:     if (--mReaderRefCnt == 0)
michael@0:         Close();
michael@0:     return mPipe->Release();
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::CloseWithStatus(nsresult reason)
michael@0: {
michael@0:     LOG(("III CloseWithStatus [this=%x reason=%x]\n", this, reason));
michael@0: 
michael@0:     if (NS_SUCCEEDED(reason))
michael@0:         reason = NS_BASE_STREAM_CLOSED;
michael@0: 
michael@0:     mPipe->OnPipeException(reason);
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::Close()
michael@0: {
michael@0:     return CloseWithStatus(NS_BASE_STREAM_CLOSED);
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::Available(uint64_t *result)
michael@0: {
michael@0:     // nsPipeInputStream supports under 4GB stream only
michael@0:     ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
michael@0: 
michael@0:     // return error if pipe closed
michael@0:     if (!mAvailable && NS_FAILED(mPipe->mStatus))
michael@0:         return mPipe->mStatus;
michael@0: 
michael@0:     *result = (uint64_t)mAvailable;
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::ReadSegments(nsWriteSegmentFun writer, 
michael@0:                                 void *closure,  
michael@0:                                 uint32_t count,
michael@0:                                 uint32_t *readCount)
michael@0: {
michael@0:     LOG(("III ReadSegments [this=%x count=%u]\n", this, count));
michael@0: 
michael@0:     nsresult rv = NS_OK;
michael@0: 
michael@0:     const char *segment;
michael@0:     uint32_t segmentLen;
michael@0: 
michael@0:     *readCount = 0;
michael@0:     while (count) {
michael@0:         rv = mPipe->GetReadSegment(segment, segmentLen);
michael@0:         if (NS_FAILED(rv)) {
michael@0:             // ignore this error if we've already read something.
michael@0:             if (*readCount > 0) {
michael@0:                 rv = NS_OK;
michael@0:                 break;
michael@0:             }
michael@0:             if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
michael@0:                 // pipe is empty
michael@0:                 if (!mBlocking)
michael@0:                     break;
michael@0:                 // wait for some data to be written to the pipe
michael@0:                 rv = Wait();
michael@0:                 if (NS_SUCCEEDED(rv))
michael@0:                     continue;
michael@0:             }
michael@0:             // ignore this error, just return.
michael@0:             if (rv == NS_BASE_STREAM_CLOSED) {
michael@0:                 rv = NS_OK;
michael@0:                 break;
michael@0:             }
michael@0:             mPipe->OnPipeException(rv);
michael@0:             break;
michael@0:         }
michael@0: 
michael@0:         // read no more than count
michael@0:         if (segmentLen > count)
michael@0:             segmentLen = count;
michael@0: 
michael@0:         uint32_t writeCount, originalLen = segmentLen;
michael@0:         while (segmentLen) {
michael@0:             writeCount = 0;
michael@0: 
michael@0:             rv = writer(this, closure, segment, *readCount, segmentLen, &writeCount);
michael@0: 
michael@0:             if (NS_FAILED(rv) || writeCount == 0) {
michael@0:                 count = 0;
michael@0:                 // any errors returned from the writer end here: do not
michael@0:                 // propagate to the caller of ReadSegments.
michael@0:                 rv = NS_OK;
michael@0:                 break;
michael@0:             }
michael@0: 
michael@0:             NS_ASSERTION(writeCount <= segmentLen, "wrote more than expected");
michael@0:             segment += writeCount;
michael@0:             segmentLen -= writeCount;
michael@0:             count -= writeCount;
michael@0:             *readCount += writeCount;
michael@0:             mLogicalOffset += writeCount;
michael@0:         }
michael@0: 
michael@0:         if (segmentLen < originalLen)
michael@0:             mPipe->AdvanceReadCursor(originalLen - segmentLen);
michael@0:     }
michael@0: 
michael@0:     return rv;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::Read(char* toBuf, uint32_t bufLen, uint32_t *readCount)
michael@0: {
michael@0:     return ReadSegments(NS_CopySegmentToBuffer, toBuf, bufLen, readCount);
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::IsNonBlocking(bool *aNonBlocking)
michael@0: {
michael@0:     *aNonBlocking = !mBlocking;
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::AsyncWait(nsIInputStreamCallback *callback,
michael@0:                              uint32_t flags,
michael@0:                              uint32_t requestedCount,
michael@0:                              nsIEventTarget *target)
michael@0: {
michael@0:     LOG(("III AsyncWait [this=%x]\n", this));
michael@0: 
michael@0:     nsPipeEvents pipeEvents;
michael@0:     {
michael@0:         ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
michael@0: 
michael@0:         // replace a pending callback
michael@0:         mCallback = 0;
michael@0:         mCallbackFlags = 0;
michael@0: 
michael@0:         if (!callback)
michael@0:             return NS_OK;
michael@0: 
michael@0:         nsCOMPtr<nsIInputStreamCallback> proxy;
michael@0:         if (target) {
michael@0:             proxy = NS_NewInputStreamReadyEvent(callback, target);
michael@0:             callback = proxy;
michael@0:         }
michael@0: 
michael@0:         if (NS_FAILED(mPipe->mStatus) ||
michael@0:                 (mAvailable && !(flags & WAIT_CLOSURE_ONLY))) {
michael@0:             // stream is already closed or readable; post event.
michael@0:             pipeEvents.NotifyInputReady(this, callback);
michael@0:         }
michael@0:         else {
michael@0:             // queue up callback object to be notified when data becomes available
michael@0:             mCallback = callback;
michael@0:             mCallbackFlags = flags;
michael@0:         }
michael@0:     }
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::Seek(int32_t whence, int64_t offset)
michael@0: {
michael@0:     NS_NOTREACHED("nsPipeInputStream::Seek");
michael@0:     return NS_ERROR_NOT_IMPLEMENTED;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::Tell(int64_t *offset)
michael@0: {
michael@0:     ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
michael@0: 
michael@0:     // return error if pipe closed
michael@0:     if (!mAvailable && NS_FAILED(mPipe->mStatus))
michael@0:         return mPipe->mStatus;
michael@0: 
michael@0:     *offset = mLogicalOffset;
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::SetEOF()
michael@0: {
michael@0:     NS_NOTREACHED("nsPipeInputStream::SetEOF");
michael@0:     return NS_ERROR_NOT_IMPLEMENTED;
michael@0: }
michael@0: 
michael@0: #define COMPARE(s1, s2, i)                                                 \
michael@0:     (ignoreCase                                                            \
michael@0:      ? nsCRT::strncasecmp((const char *)s1, (const char *)s2, (uint32_t)i) \
michael@0:      : nsCRT::strncmp((const char *)s1, (const char *)s2, (uint32_t)i))
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeInputStream::Search(const char *forString, 
michael@0:                           bool ignoreCase,
michael@0:                           bool *found,
michael@0:                           uint32_t *offsetSearchedTo)
michael@0: {
michael@0:     LOG(("III Search [for=%s ic=%u]\n", forString, ignoreCase));
michael@0: 
michael@0:     ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
michael@0: 
michael@0:     char *cursor1, *limit1;
michael@0:     uint32_t index = 0, offset = 0;
michael@0:     uint32_t strLen = strlen(forString);
michael@0: 
michael@0:     mPipe->PeekSegment(0, cursor1, limit1);
michael@0:     if (cursor1 == limit1) {
michael@0:         *found = false;
michael@0:         *offsetSearchedTo = 0;
michael@0:         LOG(("  result [found=%u offset=%u]\n", *found, *offsetSearchedTo));
michael@0:         return NS_OK;
michael@0:     }
michael@0: 
michael@0:     while (true) {
michael@0:         uint32_t i, len1 = limit1 - cursor1;
michael@0: 
michael@0:         // check if the string is in the buffer segment
michael@0:         for (i = 0; i < len1 - strLen + 1; i++) {
michael@0:             if (COMPARE(&cursor1[i], forString, strLen) == 0) {
michael@0:                 *found = true;
michael@0:                 *offsetSearchedTo = offset + i;
michael@0:                 LOG(("  result [found=%u offset=%u]\n", *found, *offsetSearchedTo));
michael@0:                 return NS_OK;
michael@0:             }
michael@0:         }
michael@0: 
michael@0:         // get the next segment
michael@0:         char *cursor2, *limit2;
michael@0:         uint32_t len2;
michael@0: 
michael@0:         index++;
michael@0:         offset += len1;
michael@0: 
michael@0:         mPipe->PeekSegment(index, cursor2, limit2);
michael@0:         if (cursor2 == limit2) {
michael@0:             *found = false;
michael@0:             *offsetSearchedTo = offset - strLen + 1;
michael@0:             LOG(("  result [found=%u offset=%u]\n", *found, *offsetSearchedTo));
michael@0:             return NS_OK;
michael@0:         }
michael@0:         len2 = limit2 - cursor2;
michael@0: 
michael@0:         // check if the string is straddling the next buffer segment
michael@0:         uint32_t lim = XPCOM_MIN(strLen, len2 + 1);
michael@0:         for (i = 0; i < lim; ++i) {
michael@0:             uint32_t strPart1Len = strLen - i - 1;
michael@0:             uint32_t strPart2Len = strLen - strPart1Len;
michael@0:             const char* strPart2 = &forString[strLen - strPart2Len];
michael@0:             uint32_t bufSeg1Offset = len1 - strPart1Len;
michael@0:             if (COMPARE(&cursor1[bufSeg1Offset], forString, strPart1Len) == 0 &&
michael@0:                 COMPARE(cursor2, strPart2, strPart2Len) == 0) {
michael@0:                 *found = true;
michael@0:                 *offsetSearchedTo = offset - strPart1Len;
michael@0:                 LOG(("  result [found=%u offset=%u]\n", *found, *offsetSearchedTo));
michael@0:                 return NS_OK;
michael@0:             }
michael@0:         }
michael@0: 
michael@0:         // finally continue with the next buffer
michael@0:         cursor1 = cursor2;
michael@0:         limit1 = limit2;
michael@0:     }
michael@0: 
michael@0:     NS_NOTREACHED("can't get here");
michael@0:     return NS_ERROR_UNEXPECTED;    // keep compiler happy
michael@0: }
michael@0: 
michael@0: //-----------------------------------------------------------------------------
michael@0: // nsPipeOutputStream methods:
michael@0: //-----------------------------------------------------------------------------
michael@0: 
michael@0: NS_IMPL_QUERY_INTERFACE(nsPipeOutputStream,
michael@0:                         nsIOutputStream,
michael@0:                         nsIAsyncOutputStream,
michael@0:                         nsIClassInfo)
michael@0: 
michael@0: NS_IMPL_CI_INTERFACE_GETTER(nsPipeOutputStream,
michael@0:                             nsIOutputStream,
michael@0:                             nsIAsyncOutputStream)
michael@0: 
michael@0: NS_IMPL_THREADSAFE_CI(nsPipeOutputStream)
michael@0: 
michael@0: nsresult
michael@0: nsPipeOutputStream::Wait()
michael@0: {
michael@0:     NS_ASSERTION(mBlocking, "wait on non-blocking pipe output stream");
michael@0: 
michael@0:     ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
michael@0: 
michael@0:     if (NS_SUCCEEDED(mPipe->mStatus) && !mWritable) {
michael@0:         LOG(("OOO pipe output: waiting for space\n"));
michael@0:         mBlocked = true;
michael@0:         mon.Wait();
michael@0:         mBlocked = false;
michael@0:         LOG(("OOO pipe output: woke up [pipe-status=%x writable=%u]\n",
michael@0:             mPipe->mStatus, mWritable));
michael@0:     }
michael@0: 
michael@0:     return mPipe->mStatus == NS_BASE_STREAM_CLOSED ? NS_OK : mPipe->mStatus;
michael@0: }
michael@0: 
michael@0: bool
michael@0: nsPipeOutputStream::OnOutputWritable(nsPipeEvents &events)
michael@0: {
michael@0:     bool result = false;
michael@0: 
michael@0:     mWritable = true;
michael@0: 
michael@0:     if (mCallback && !(mCallbackFlags & WAIT_CLOSURE_ONLY)) {
michael@0:         events.NotifyOutputReady(this, mCallback);
michael@0:         mCallback = 0;
michael@0:         mCallbackFlags = 0;
michael@0:     }
michael@0:     else if (mBlocked)
michael@0:         result = true;
michael@0: 
michael@0:     return result;
michael@0: }
michael@0: 
michael@0: bool
michael@0: nsPipeOutputStream::OnOutputException(nsresult reason, nsPipeEvents &events)
michael@0: {
michael@0:     LOG(("nsPipeOutputStream::OnOutputException [this=%x reason=%x]\n",
michael@0:         this, reason));
michael@0: 
michael@0:     bool result = false;
michael@0: 
michael@0:     NS_ASSERTION(NS_FAILED(reason), "huh? successful exception");
michael@0:     mWritable = false;
michael@0: 
michael@0:     if (mCallback) {
michael@0:         events.NotifyOutputReady(this, mCallback);
michael@0:         mCallback = 0;
michael@0:         mCallbackFlags = 0;
michael@0:     }
michael@0:     else if (mBlocked)
michael@0:         result = true;
michael@0: 
michael@0:     return result;
michael@0: }
michael@0: 
michael@0: 
michael@0: NS_IMETHODIMP_(MozExternalRefCountType)
michael@0: nsPipeOutputStream::AddRef()
michael@0: {
michael@0:     ++mWriterRefCnt;
michael@0:     return mPipe->AddRef();
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP_(MozExternalRefCountType)
michael@0: nsPipeOutputStream::Release()
michael@0: {
michael@0:     if (--mWriterRefCnt == 0)
michael@0:         Close();
michael@0:     return mPipe->Release();
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeOutputStream::CloseWithStatus(nsresult reason)
michael@0: {
michael@0:     LOG(("OOO CloseWithStatus [this=%x reason=%x]\n", this, reason));
michael@0: 
michael@0:     if (NS_SUCCEEDED(reason))
michael@0:         reason = NS_BASE_STREAM_CLOSED;
michael@0: 
michael@0:     // input stream may remain open
michael@0:     mPipe->OnPipeException(reason, true);
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeOutputStream::Close()
michael@0: {
michael@0:     return CloseWithStatus(NS_BASE_STREAM_CLOSED);
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeOutputStream::WriteSegments(nsReadSegmentFun reader,
michael@0:                                   void* closure,
michael@0:                                   uint32_t count,
michael@0:                                   uint32_t *writeCount)
michael@0: {
michael@0:     LOG(("OOO WriteSegments [this=%x count=%u]\n", this, count));
michael@0: 
michael@0:     nsresult rv = NS_OK;
michael@0: 
michael@0:     char *segment;
michael@0:     uint32_t segmentLen;
michael@0: 
michael@0:     *writeCount = 0;
michael@0:     while (count) {
michael@0:         rv = mPipe->GetWriteSegment(segment, segmentLen);
michael@0:         if (NS_FAILED(rv)) {
michael@0:             if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
michael@0:                 // pipe is full
michael@0:                 if (!mBlocking) {
michael@0:                     // ignore this error if we've already written something
michael@0:                     if (*writeCount > 0)
michael@0:                         rv = NS_OK;
michael@0:                     break;
michael@0:                 }
michael@0:                 // wait for the pipe to have an empty segment.
michael@0:                 rv = Wait();
michael@0:                 if (NS_SUCCEEDED(rv))
michael@0:                     continue;
michael@0:             }
michael@0:             mPipe->OnPipeException(rv);
michael@0:             break;
michael@0:         }
michael@0: 
michael@0:         // write no more than count
michael@0:         if (segmentLen > count)
michael@0:             segmentLen = count;
michael@0: 
michael@0:         uint32_t readCount, originalLen = segmentLen;
michael@0:         while (segmentLen) {
michael@0:             readCount = 0;
michael@0: 
michael@0:             rv = reader(this, closure, segment, *writeCount, segmentLen, &readCount);
michael@0: 
michael@0:             if (NS_FAILED(rv) || readCount == 0) {
michael@0:                 count = 0;
michael@0:                 // any errors returned from the reader end here: do not
michael@0:                 // propagate to the caller of WriteSegments.
michael@0:                 rv = NS_OK;
michael@0:                 break;
michael@0:             }
michael@0: 
michael@0:             NS_ASSERTION(readCount <= segmentLen, "read more than expected");
michael@0:             segment += readCount;
michael@0:             segmentLen -= readCount;
michael@0:             count -= readCount;
michael@0:             *writeCount += readCount;
michael@0:             mLogicalOffset += readCount;
michael@0:         }
michael@0: 
michael@0:         if (segmentLen < originalLen)
michael@0:             mPipe->AdvanceWriteCursor(originalLen - segmentLen);
michael@0:     }
michael@0: 
michael@0:     return rv;
michael@0: }
michael@0: 
michael@0: static NS_METHOD
michael@0: nsReadFromRawBuffer(nsIOutputStream* outStr,
michael@0:                     void* closure,
michael@0:                     char* toRawSegment,
michael@0:                     uint32_t offset,
michael@0:                     uint32_t count,
michael@0:                     uint32_t *readCount)
michael@0: {
michael@0:     const char* fromBuf = (const char*)closure;
michael@0:     memcpy(toRawSegment, &fromBuf[offset], count);
michael@0:     *readCount = count;
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeOutputStream::Write(const char* fromBuf,
michael@0:                           uint32_t bufLen, 
michael@0:                           uint32_t *writeCount)
michael@0: {
michael@0:     return WriteSegments(nsReadFromRawBuffer, (void*)fromBuf, bufLen, writeCount);
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeOutputStream::Flush(void)
michael@0: {
michael@0:     // nothing to do
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: static NS_METHOD
michael@0: nsReadFromInputStream(nsIOutputStream* outStr,
michael@0:                       void* closure,
michael@0:                       char* toRawSegment, 
michael@0:                       uint32_t offset,
michael@0:                       uint32_t count,
michael@0:                       uint32_t *readCount)
michael@0: {
michael@0:     nsIInputStream* fromStream = (nsIInputStream*)closure;
michael@0:     return fromStream->Read(toRawSegment, count, readCount);
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeOutputStream::WriteFrom(nsIInputStream* fromStream,
michael@0:                               uint32_t count,
michael@0:                               uint32_t *writeCount)
michael@0: {
michael@0:     return WriteSegments(nsReadFromInputStream, fromStream, count, writeCount);
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeOutputStream::IsNonBlocking(bool *aNonBlocking)
michael@0: {
michael@0:     *aNonBlocking = !mBlocking;
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: nsPipeOutputStream::AsyncWait(nsIOutputStreamCallback *callback,
michael@0:                               uint32_t flags,
michael@0:                               uint32_t requestedCount,
michael@0:                               nsIEventTarget *target)
michael@0: {
michael@0:     LOG(("OOO AsyncWait [this=%x]\n", this));
michael@0: 
michael@0:     nsPipeEvents pipeEvents;
michael@0:     {
michael@0:         ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
michael@0: 
michael@0:         // replace a pending callback
michael@0:         mCallback = 0;
michael@0:         mCallbackFlags = 0;
michael@0: 
michael@0:         if (!callback)
michael@0:             return NS_OK;
michael@0: 
michael@0:         nsCOMPtr<nsIOutputStreamCallback> proxy;
michael@0:         if (target) {
michael@0:             proxy = NS_NewOutputStreamReadyEvent(callback, target);
michael@0:             callback = proxy;
michael@0:         }
michael@0: 
michael@0:         if (NS_FAILED(mPipe->mStatus) ||
michael@0:                 (mWritable && !(flags & WAIT_CLOSURE_ONLY))) {
michael@0:             // stream is already closed or writable; post event.
michael@0:             pipeEvents.NotifyOutputReady(this, callback);
michael@0:         }
michael@0:         else {
michael@0:             // queue up callback object to be notified when data becomes available
michael@0:             mCallback = callback;
michael@0:             mCallbackFlags = flags;
michael@0:         }
michael@0:     }
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: ////////////////////////////////////////////////////////////////////////////////
michael@0: 
michael@0: nsresult
michael@0: NS_NewPipe(nsIInputStream **pipeIn,
michael@0:            nsIOutputStream **pipeOut,
michael@0:            uint32_t segmentSize,
michael@0:            uint32_t maxSize,
michael@0:            bool nonBlockingInput,
michael@0:            bool nonBlockingOutput)
michael@0: {
michael@0:     if (segmentSize == 0)
michael@0:         segmentSize = DEFAULT_SEGMENT_SIZE;
michael@0: 
michael@0:     // Handle maxSize of UINT32_MAX as a special case
michael@0:     uint32_t segmentCount;
michael@0:     if (maxSize == UINT32_MAX)
michael@0:         segmentCount = UINT32_MAX;
michael@0:     else
michael@0:         segmentCount = maxSize / segmentSize;
michael@0: 
michael@0:     nsIAsyncInputStream *in;
michael@0:     nsIAsyncOutputStream *out;
michael@0:     nsresult rv = NS_NewPipe2(&in, &out, nonBlockingInput, nonBlockingOutput,
michael@0:                               segmentSize, segmentCount);
michael@0:     if (NS_FAILED(rv)) return rv;
michael@0: 
michael@0:     *pipeIn = in;
michael@0:     *pipeOut = out;
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: nsresult
michael@0: NS_NewPipe2(nsIAsyncInputStream **pipeIn,
michael@0:             nsIAsyncOutputStream **pipeOut,
michael@0:             bool nonBlockingInput,
michael@0:             bool nonBlockingOutput,
michael@0:             uint32_t segmentSize,
michael@0:             uint32_t segmentCount)
michael@0: {
michael@0:     nsresult rv;
michael@0: 
michael@0:     nsPipe *pipe = new nsPipe();
michael@0:     if (!pipe)
michael@0:         return NS_ERROR_OUT_OF_MEMORY;
michael@0: 
michael@0:     rv = pipe->Init(nonBlockingInput,
michael@0:                     nonBlockingOutput,
michael@0:                     segmentSize,
michael@0:                     segmentCount);
michael@0:     if (NS_FAILED(rv)) {
michael@0:         NS_ADDREF(pipe);
michael@0:         NS_RELEASE(pipe);
michael@0:         return rv;
michael@0:     }
michael@0: 
michael@0:     pipe->GetInputStream(pipeIn);
michael@0:     pipe->GetOutputStream(pipeOut);
michael@0:     return NS_OK;
michael@0: }
michael@0: 
michael@0: nsresult
michael@0: nsPipeConstructor(nsISupports *outer, REFNSIID iid, void **result)
michael@0: {
michael@0:     if (outer)
michael@0:         return NS_ERROR_NO_AGGREGATION;
michael@0:     nsPipe *pipe = new nsPipe();
michael@0:     if (!pipe)
michael@0:         return NS_ERROR_OUT_OF_MEMORY;
michael@0:     NS_ADDREF(pipe);
michael@0:     nsresult rv = pipe->QueryInterface(iid, result);
michael@0:     NS_RELEASE(pipe);
michael@0:     return rv;
michael@0: }
michael@0: 
michael@0: ////////////////////////////////////////////////////////////////////////////////