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