The Tor Browser: xpcom/io/nsMultiplexInputStream.cpp@b8a032363ba2 (annotated)

xpcom/io/nsMultiplexInputStream.cpp@b8a032363ba2 (annotated)

xpcom/io/nsMultiplexInputStream.cpp

Thu, 22 Jan 2015 13:21:57 +0100

author: Michael Schloh von Bennewitz <michael@schloh.com>
date: Thu, 22 Jan 2015 13:21:57 +0100
branch: TOR_BUG_9701
changeset 15: b8a032363ba2
permissions: -rw-r--r--

Incorporate requested changes from Mozilla in review:
https://bugzilla.mozilla.org/show_bug.cgi?id=1123480#c6

 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
 /* 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/MathAlgorithms.h"
 #include "base/basictypes.h"
 #include "nsMultiplexInputStream.h"
 #include "nsIMultiplexInputStream.h"
 #include "nsISeekableStream.h"
 #include "nsCOMPtr.h"
 #include "nsCOMArray.h"
 #include "nsIClassInfoImpl.h"
 #include "nsIIPCSerializableInputStream.h"
 #include "mozilla/ipc/InputStreamUtils.h"
 using namespace mozilla::ipc;
 using mozilla::DeprecatedAbs;
 class nsMultiplexInputStream MOZ_FINAL : public nsIMultiplexInputStream,
                                          public nsISeekableStream,
                                          public nsIIPCSerializableInputStream
 {
 public:
     nsMultiplexInputStream();
     NS_DECL_THREADSAFE_ISUPPORTS
     NS_DECL_NSIINPUTSTREAM
     NS_DECL_NSIMULTIPLEXINPUTSTREAM
     NS_DECL_NSISEEKABLESTREAM
     NS_DECL_NSIIPCSERIALIZABLEINPUTSTREAM
 private:
     ~nsMultiplexInputStream() {}
     struct ReadSegmentsState {
         nsIInputStream* mThisStream;
         uint32_t mOffset;
         nsWriteSegmentFun mWriter;
         void* mClosure;
         bool mDone;
     };
     static NS_METHOD ReadSegCb(nsIInputStream* aIn, void* aClosure,
                                const char* aFromRawSegment, uint32_t aToOffset,
                                uint32_t aCount, uint32_t *aWriteCount);
     nsTArray<nsCOMPtr<nsIInputStream> > mStreams;
     uint32_t mCurrentStream;
     bool mStartedReadingCurrent;
     nsresult mStatus;
 };
 NS_IMPL_ADDREF(nsMultiplexInputStream)
 NS_IMPL_RELEASE(nsMultiplexInputStream)
 NS_IMPL_CLASSINFO(nsMultiplexInputStream, nullptr, nsIClassInfo::THREADSAFE,
                   NS_MULTIPLEXINPUTSTREAM_CID)
 NS_IMPL_QUERY_INTERFACE_CI(nsMultiplexInputStream,
                            nsIMultiplexInputStream,
                            nsIInputStream,
                            nsISeekableStream,
                            nsIIPCSerializableInputStream)
 NS_IMPL_CI_INTERFACE_GETTER(nsMultiplexInputStream,
                             nsIMultiplexInputStream,
                             nsIInputStream,
                             nsISeekableStream)
 nsMultiplexInputStream::nsMultiplexInputStream()
     : mCurrentStream(0),
       mStartedReadingCurrent(false),
       mStatus(NS_OK)
 {
 }
 /* readonly attribute unsigned long count; */
 NS_IMETHODIMP
 nsMultiplexInputStream::GetCount(uint32_t *aCount)
 {
     *aCount = mStreams.Length();
     return NS_OK;
 }
 #ifdef DEBUG
 static bool
 SeekableStreamAtBeginning(nsIInputStream *aStream)
 {
     int64_t streamPos;
     nsCOMPtr<nsISeekableStream> stream = do_QueryInterface(aStream);
     if (stream && NS_SUCCEEDED(stream->Tell(&streamPos)) && streamPos != 0) {
         return false;
     }
     return true;
 }
 #endif
 /* void appendStream (in nsIInputStream stream); */
 NS_IMETHODIMP
 nsMultiplexInputStream::AppendStream(nsIInputStream *aStream)
 {
     NS_ASSERTION(SeekableStreamAtBeginning(aStream), "Appended stream not at beginning.");
     return mStreams.AppendElement(aStream) ? NS_OK : NS_ERROR_OUT_OF_MEMORY;
 }
 /* void insertStream (in nsIInputStream stream, in unsigned long index); */
 NS_IMETHODIMP
 nsMultiplexInputStream::InsertStream(nsIInputStream *aStream, uint32_t aIndex)
 {
     NS_ASSERTION(SeekableStreamAtBeginning(aStream), "Inserted stream not at beginning.");
     mStreams.InsertElementAt(aIndex, aStream);
     if (mCurrentStream > aIndex ||
         (mCurrentStream == aIndex && mStartedReadingCurrent))
         ++mCurrentStream;
     return NS_OK;
 }
 /* void removeStream (in unsigned long index); */
 NS_IMETHODIMP
 nsMultiplexInputStream::RemoveStream(uint32_t aIndex)
 {
     mStreams.RemoveElementAt(aIndex);
     if (mCurrentStream > aIndex)
         --mCurrentStream;
     else if (mCurrentStream == aIndex)
         mStartedReadingCurrent = false;
     return NS_OK;
 }
 /* nsIInputStream getStream (in unsigned long index); */
 NS_IMETHODIMP
 nsMultiplexInputStream::GetStream(uint32_t aIndex, nsIInputStream **_retval)
 {
     *_retval = mStreams.SafeElementAt(aIndex, nullptr);
     if (NS_WARN_IF(!*_retval))
         return NS_ERROR_NOT_AVAILABLE;
     NS_ADDREF(*_retval);
     return NS_OK;
 }
 /* void close (); */
 NS_IMETHODIMP
 nsMultiplexInputStream::Close()
 {
     mStatus = NS_BASE_STREAM_CLOSED;
     nsresult rv = NS_OK;
     uint32_t len = mStreams.Length();
     for (uint32_t i = 0; i < len; ++i) {
         nsresult rv2 = mStreams[i]->Close();
         // We still want to close all streams, but we should return an error
         if (NS_FAILED(rv2))
             rv = rv2;
     }
     return rv;
 }
 /* unsigned long long available (); */
 NS_IMETHODIMP
 nsMultiplexInputStream::Available(uint64_t *_retval)
 {
     if (NS_FAILED(mStatus))
         return mStatus;
     nsresult rv;
     uint64_t avail = 0;
     uint32_t len = mStreams.Length();
     for (uint32_t i = mCurrentStream; i < len; i++) {
         uint64_t streamAvail;
         rv = mStreams[i]->Available(&streamAvail);
         if (NS_WARN_IF(NS_FAILED(rv)))
             return rv;
         avail += streamAvail;
     }
     *_retval = avail;
     return NS_OK;
 }
 /* [noscript] unsigned long read (in charPtr buf, in unsigned long count); */
 NS_IMETHODIMP
 nsMultiplexInputStream::Read(char * aBuf, uint32_t aCount, uint32_t *_retval)
 {
     // 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).
     *_retval = 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]->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) {
             NS_NOTREACHED("Input stream's Read method returned NS_BASE_STREAM_CLOSED");
             rv = NS_OK;
             read = 0;
         }
         else if (NS_FAILED(rv))
             break;
         if (read == 0) {
             ++mCurrentStream;
             mStartedReadingCurrent = false;
         }
         else {
             NS_ASSERTION(aCount >= read, "Read more than requested");
             *_retval += read;
             aCount -= read;
             aBuf += read;
             mStartedReadingCurrent = true;
         }
     }
     return *_retval ? NS_OK : rv;
 }
 /* [noscript] unsigned long readSegments (in nsWriteSegmentFun writer,
  *                                        in voidPtr closure,
  *                                        in unsigned long count); */
 NS_IMETHODIMP
 nsMultiplexInputStream::ReadSegments(nsWriteSegmentFun aWriter, void *aClosure,
                                      uint32_t aCount, uint32_t *_retval)
 {
     if (mStatus == NS_BASE_STREAM_CLOSED) {
         *_retval = 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]->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) {
             NS_NOTREACHED("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) {
             ++mCurrentStream;
             mStartedReadingCurrent = false;
         }
         else {
             NS_ASSERTION(aCount >= read, "Read more than requested");
             state.mOffset += read;
             aCount -= read;
             mStartedReadingCurrent = true;
         }
     }
     // if we successfully read some data, then this call succeeded.
     *_retval = state.mOffset;
     return state.mOffset ? NS_OK : rv;
 }
 NS_METHOD
 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;
 }
 /* readonly attribute boolean nonBlocking; */
 NS_IMETHODIMP
 nsMultiplexInputStream::IsNonBlocking(bool *aNonBlocking)
 {
     uint32_t len = mStreams.Length();
     if (len == 0) {
         // Claim to be non-blocking, since we won't block the caller.
         // On the other hand we'll never return NS_BASE_STREAM_WOULD_BLOCK,
         // so maybe we should claim to be blocking?  It probably doesn't
         // matter in practice.
         *aNonBlocking = true;
         return NS_OK;
     }
     for (uint32_t i = 0; i < len; ++i) {
         nsresult rv = mStreams[i]->IsNonBlocking(aNonBlocking);
         if (NS_WARN_IF(NS_FAILED(rv)))
             return rv;
         // If one is non-blocking the entire stream becomes non-blocking
         // (except that we don't implement nsIAsyncInputStream, so there's
         //  not much for the caller to do if Read returns "would block")
         if (*aNonBlocking)
             return NS_OK;
     }
     return NS_OK;
 }
 /* void seek (in int32_t whence, in int32_t offset); */
 NS_IMETHODIMP
 nsMultiplexInputStream::Seek(int32_t aWhence, int64_t aOffset)
 {
     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 =
                 do_QueryInterface(mStreams[i]);
             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;
                     continue;
                 }
                 else {
                     break;
                 }
             }
             // Get position in current stream
             int64_t streamPos;
             if (i > oldCurrentStream ||
                 (i == oldCurrentStream && !oldStartedReadingCurrent)) {
                 streamPos = 0;
             }
             else {
                 rv = stream->Tell(&streamPos);
                 if (NS_WARN_IF(NS_FAILED(rv)))
                     return rv;
             }
             // See if we need to seek current stream forward or backward
             if (remaining < streamPos) {
                 rv = stream->Seek(NS_SEEK_SET, remaining);
                 if (NS_WARN_IF(NS_FAILED(rv)))
                     return rv;
                 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 = mStreams[i]->Available(&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;
                     mCurrentStream = i;
                     mStartedReadingCurrent = true;
                     remaining -= newPos;
                     NS_ASSERTION(remaining >= 0, "Remaining invalid");
                 }
             }
             else {
                 NS_ASSERTION(remaining == streamPos, "Huh?");
                 remaining = 0;
             }
         }
         return NS_OK;
     }
     if (aWhence == NS_SEEK_CUR && aOffset > 0) {
         int64_t remaining = aOffset;
         for (uint32_t i = mCurrentStream; remaining && i < mStreams.Length(); ++i) {
             nsCOMPtr<nsISeekableStream> stream =
                 do_QueryInterface(mStreams[i]);
             uint64_t avail;
             rv = mStreams[i]->Available(&avail);
             if (NS_WARN_IF(NS_FAILED(rv)))
                 return rv;
             int64_t seek = XPCOM_MIN((int64_t)avail, remaining);
             rv = stream->Seek(NS_SEEK_CUR, seek);
             if (NS_WARN_IF(NS_FAILED(rv)))
                 return rv;
             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) {
             nsCOMPtr<nsISeekableStream> stream =
                 do_QueryInterface(mStreams[i]);
             int64_t pos;
             rv = stream->Tell(&pos);
             if (NS_WARN_IF(NS_FAILED(rv)))
                 return rv;
             int64_t seek = XPCOM_MIN(pos, remaining);
             rv = stream->Seek(NS_SEEK_CUR, -seek);
             if (NS_WARN_IF(NS_FAILED(rv)))
                 return rv;
             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;
         for (uint32_t i = mStreams.Length() - 1; i != (uint32_t)-1; --i) {
             nsCOMPtr<nsISeekableStream> stream =
                 do_QueryInterface(mStreams[i]);
             // See if all remaining streams should be seeked to end
             if (remaining == 0) {
                 if (i >= oldCurrentStream) {
                     rv = stream->Seek(NS_SEEK_END, 0);
                     if (NS_WARN_IF(NS_FAILED(rv)))
                         return rv;
                 }
                 else {
                     break;
                 }
             }
             // Get position in current stream
             int64_t streamPos;
             if (i < oldCurrentStream) {
                 streamPos = 0;
             } else {
                 uint64_t avail;
                 rv = mStreams[i]->Available(&avail);
                 if (NS_WARN_IF(NS_FAILED(rv)))
                     return rv;
                 streamPos = avail;
             }
             // See if we have enough data in the current stream.
             if (DeprecatedAbs(remaining) < streamPos) {
                 rv = stream->Seek(NS_SEEK_END, remaining);
                 if (NS_WARN_IF(NS_FAILED(rv)))
                     return rv;
                 mCurrentStream = i;
                 mStartedReadingCurrent = true;
                 remaining = 0;
             } else if (DeprecatedAbs(remaining) > streamPos) {
                 if (i > oldCurrentStream ||
                     (i == oldCurrentStream && !oldStartedReadingCurrent)) {
                     // We're already at start so no need to seek this stream
                     remaining += streamPos;
                 } else {
                     int64_t avail;
                     rv = stream->Tell(&avail);
                     if (NS_WARN_IF(NS_FAILED(rv)))
                         return rv;
                     int64_t newPos = streamPos + XPCOM_MIN(avail, DeprecatedAbs(remaining));
                     rv = stream->Seek(NS_SEEK_END, -newPos);
                     if (NS_WARN_IF(NS_FAILED(rv)))
                         return rv;
                     mCurrentStream = i;
                     mStartedReadingCurrent = true;
                     remaining += newPos;
                 }
             }
             else {
                 NS_ASSERTION(remaining == streamPos, "Huh?");
                 remaining = 0;
             }
         }
         return NS_OK;
     }
     // other Seeks not implemented yet
     return NS_ERROR_NOT_IMPLEMENTED;
 }
 /* uint32_t tell (); */
 NS_IMETHODIMP
 nsMultiplexInputStream::Tell(int64_t *_retval)
 {
     if (NS_FAILED(mStatus))
         return mStatus;
     nsresult rv;
     int64_t ret64 = 0;
     uint32_t i, last;
     last = mStartedReadingCurrent ? mCurrentStream+1 : mCurrentStream;
     for (i = 0; i < last; ++i) {
         nsCOMPtr<nsISeekableStream> stream = do_QueryInterface(mStreams[i]);
         if (NS_WARN_IF(!stream))
             return NS_ERROR_NO_INTERFACE;
         int64_t pos;
         rv = stream->Tell(&pos);
         if (NS_WARN_IF(NS_FAILED(rv)))
             return rv;
         ret64 += pos;
     }
     *_retval =  ret64;
     return NS_OK;
 }
 /* void setEOF (); */
 NS_IMETHODIMP
 nsMultiplexInputStream::SetEOF()
 {
     return NS_ERROR_NOT_IMPLEMENTED;
 }
 nsresult
 nsMultiplexInputStreamConstructor(nsISupports *outer,
                                   REFNSIID iid,
                                   void **result)
 {
     *result = nullptr;
     if (outer)
         return NS_ERROR_NO_AGGREGATION;
     nsMultiplexInputStream *inst = new nsMultiplexInputStream();
     if (!inst)
         return NS_ERROR_OUT_OF_MEMORY;
     NS_ADDREF(inst);
     nsresult rv = inst->QueryInterface(iid, result);
     NS_RELEASE(inst);
     return rv;
 }
 void
 nsMultiplexInputStream::Serialize(InputStreamParams& aParams,
                                   FileDescriptorArray& aFileDescriptors)
 {
     MultiplexInputStreamParams params;
     uint32_t streamCount = mStreams.Length();
     if (streamCount) {
         InfallibleTArray<InputStreamParams>& streams = params.streams();
         streams.SetCapacity(streamCount);
         for (uint32_t index = 0; index < streamCount; index++) {
             InputStreamParams childStreamParams;
             SerializeInputStream(mStreams[index], childStreamParams,
                                  aFileDescriptors);
             streams.AppendElement(childStreamParams);
         }
     }
     params.currentStream() = mCurrentStream;
     params.status() = mStatus;
     params.startedReadingCurrent() = mStartedReadingCurrent;
     aParams = params;
 }
 bool
 nsMultiplexInputStream::Deserialize(const InputStreamParams& aParams,
                                     const FileDescriptorArray& aFileDescriptors)
 {
     if (aParams.type() !=
             InputStreamParams::TMultiplexInputStreamParams) {
         NS_ERROR("Received unknown parameters from the other process!");
         return false;
     }
     const MultiplexInputStreamParams& params =
         aParams.get_MultiplexInputStreamParams();
     const InfallibleTArray<InputStreamParams>& streams = params.streams();
     uint32_t streamCount = streams.Length();
     for (uint32_t index = 0; index < streamCount; index++) {
         nsCOMPtr<nsIInputStream> stream =
             DeserializeInputStream(streams[index], aFileDescriptors);
         if (!stream) {
             NS_WARNING("Deserialize failed!");
             return false;
         }
         if (NS_FAILED(AppendStream(stream))) {
             NS_WARNING("AppendStream failed!");
             return false;
         }
     }
     mCurrentStream = params.currentStream();
     mStatus = params.status();
     mStartedReadingCurrent = params.startedReadingCurrent();
     return true;
 }

The Tor Browser / annotate

xpcom/io/nsMultiplexInputStream.cpp@b8a032363ba2 (annotated)

xpcom/io/nsMultiplexInputStream.cpp