631 строка
20 KiB
C++
631 строка
20 KiB
C++
/* vim:set ts=4 sw=4 sts=4 et cin: */
|
|
/* ***** BEGIN LICENSE BLOCK *****
|
|
* Version: MPL 1.1/GPL 2.0/LGPL 2.1
|
|
*
|
|
* The contents of this file are subject to the Mozilla Public License Version
|
|
* 1.1 (the "License"); you may not use this file except in compliance with
|
|
* the License. You may obtain a copy of the License at
|
|
* http://www.mozilla.org/MPL/
|
|
*
|
|
* Software distributed under the License is distributed on an "AS IS" basis,
|
|
* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
|
|
* for the specific language governing rights and limitations under the
|
|
* License.
|
|
*
|
|
* The Original Code is Mozilla.
|
|
*
|
|
* The Initial Developer of the Original Code is
|
|
* Netscape Communications Corporation.
|
|
* Portions created by the Initial Developer are Copyright (C) 2002
|
|
* the Initial Developer. All Rights Reserved.
|
|
*
|
|
* Contributor(s):
|
|
* Darin Fisher <darin@netscape.com>
|
|
*
|
|
* Alternatively, the contents of this file may be used under the terms of
|
|
* either the GNU General Public License Version 2 or later (the "GPL"), or
|
|
* the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
|
|
* in which case the provisions of the GPL or the LGPL are applicable instead
|
|
* of those above. If you wish to allow use of your version of this file only
|
|
* under the terms of either the GPL or the LGPL, and not to allow others to
|
|
* use your version of this file under the terms of the MPL, indicate your
|
|
* decision by deleting the provisions above and replace them with the notice
|
|
* and other provisions required by the GPL or the LGPL. If you do not delete
|
|
* the provisions above, a recipient may use your version of this file under
|
|
* the terms of any one of the MPL, the GPL or the LGPL.
|
|
*
|
|
* ***** END LICENSE BLOCK ***** */
|
|
|
|
#include "nsStreamUtils.h"
|
|
#include "nsCOMPtr.h"
|
|
#include "nsIPipe.h"
|
|
#include "nsIEventTarget.h"
|
|
#include "nsAutoLock.h"
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
class nsInputStreamReadyEvent : public PLEvent
|
|
, public nsIInputStreamCallback
|
|
{
|
|
public:
|
|
NS_DECL_ISUPPORTS
|
|
|
|
nsInputStreamReadyEvent(nsIInputStreamCallback *callback,
|
|
nsIEventTarget *target)
|
|
: mCallback(callback)
|
|
, mTarget(target)
|
|
{
|
|
}
|
|
|
|
private:
|
|
~nsInputStreamReadyEvent()
|
|
{
|
|
if (mCallback) {
|
|
nsresult rv;
|
|
//
|
|
// whoa!! looks like we never posted this event. take care to
|
|
// release mCallback on the correct thread. if mTarget lives on the
|
|
// calling thread, then we are ok. otherwise, we have to try to
|
|
// proxy the Release over the right thread. if that thread is dead,
|
|
// then there's nothing we can do... better to leak than crash.
|
|
//
|
|
PRBool val;
|
|
rv = mTarget->IsOnCurrentThread(&val);
|
|
if (NS_FAILED(rv) || !val) {
|
|
nsCOMPtr<nsIInputStreamCallback> event;
|
|
NS_NewInputStreamReadyEvent(getter_AddRefs(event), mCallback, mTarget);
|
|
mCallback = 0;
|
|
if (event) {
|
|
rv = event->OnInputStreamReady(nsnull);
|
|
if (NS_FAILED(rv)) {
|
|
NS_NOTREACHED("leaking stream event");
|
|
nsISupports *sup = event;
|
|
NS_ADDREF(sup);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
public:
|
|
NS_IMETHOD OnInputStreamReady(nsIAsyncInputStream *stream)
|
|
{
|
|
mStream = stream;
|
|
|
|
// will be released when event is handled
|
|
NS_ADDREF_THIS();
|
|
|
|
PL_InitEvent(this, nsnull, EventHandler, EventCleanup);
|
|
|
|
if (NS_FAILED(mTarget->PostEvent(this))) {
|
|
NS_WARNING("PostEvent failed");
|
|
NS_RELEASE_THIS();
|
|
return NS_ERROR_FAILURE;
|
|
}
|
|
|
|
return NS_OK;
|
|
}
|
|
|
|
private:
|
|
nsCOMPtr<nsIAsyncInputStream> mStream;
|
|
nsCOMPtr<nsIInputStreamCallback> mCallback;
|
|
nsCOMPtr<nsIEventTarget> mTarget;
|
|
|
|
PR_STATIC_CALLBACK(void *) EventHandler(PLEvent *plevent)
|
|
{
|
|
nsInputStreamReadyEvent *ev = (nsInputStreamReadyEvent *) plevent;
|
|
// bypass event delivery if this is a cleanup event...
|
|
if (ev->mCallback)
|
|
ev->mCallback->OnInputStreamReady(ev->mStream);
|
|
ev->mCallback = 0;
|
|
return NULL;
|
|
}
|
|
|
|
PR_STATIC_CALLBACK(void) EventCleanup(PLEvent *plevent)
|
|
{
|
|
nsInputStreamReadyEvent *ev = (nsInputStreamReadyEvent *) plevent;
|
|
NS_RELEASE(ev);
|
|
}
|
|
};
|
|
|
|
NS_IMPL_THREADSAFE_ISUPPORTS1(nsInputStreamReadyEvent,
|
|
nsIInputStreamCallback)
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
class nsOutputStreamReadyEvent : public PLEvent
|
|
, public nsIOutputStreamCallback
|
|
{
|
|
public:
|
|
NS_DECL_ISUPPORTS
|
|
|
|
nsOutputStreamReadyEvent(nsIOutputStreamCallback *callback,
|
|
nsIEventTarget *target)
|
|
: mCallback(callback)
|
|
, mTarget(target)
|
|
{
|
|
}
|
|
|
|
private:
|
|
~nsOutputStreamReadyEvent()
|
|
{
|
|
if (mCallback) {
|
|
nsresult rv;
|
|
//
|
|
// whoa!! looks like we never posted this event. take care to
|
|
// release mCallback on the correct thread. if mTarget lives on the
|
|
// calling thread, then we are ok. otherwise, we have to try to
|
|
// proxy the Release over the right thread. if that thread is dead,
|
|
// then there's nothing we can do... better to leak than crash.
|
|
//
|
|
PRBool val;
|
|
rv = mTarget->IsOnCurrentThread(&val);
|
|
if (NS_FAILED(rv) || !val) {
|
|
nsCOMPtr<nsIOutputStreamCallback> event;
|
|
NS_NewOutputStreamReadyEvent(getter_AddRefs(event), mCallback, mTarget);
|
|
mCallback = 0;
|
|
if (event) {
|
|
rv = event->OnOutputStreamReady(nsnull);
|
|
if (NS_FAILED(rv)) {
|
|
NS_NOTREACHED("leaking stream event");
|
|
nsISupports *sup = event;
|
|
NS_ADDREF(sup);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
public:
|
|
void Init(nsIOutputStreamCallback *callback, nsIEventTarget *target)
|
|
{
|
|
mCallback = callback;
|
|
mTarget = target;
|
|
|
|
PL_InitEvent(this, nsnull, EventHandler, EventCleanup);
|
|
}
|
|
|
|
NS_IMETHOD OnOutputStreamReady(nsIAsyncOutputStream *stream)
|
|
{
|
|
mStream = stream;
|
|
|
|
// this will be released when the event is handled
|
|
NS_ADDREF_THIS();
|
|
|
|
PL_InitEvent(this, nsnull, EventHandler, EventCleanup);
|
|
|
|
if (NS_FAILED(mTarget->PostEvent(this))) {
|
|
NS_WARNING("PostEvent failed");
|
|
NS_RELEASE_THIS();
|
|
return NS_ERROR_FAILURE;
|
|
}
|
|
|
|
return NS_OK;
|
|
}
|
|
|
|
private:
|
|
nsCOMPtr<nsIAsyncOutputStream> mStream;
|
|
nsCOMPtr<nsIOutputStreamCallback> mCallback;
|
|
nsCOMPtr<nsIEventTarget> mTarget;
|
|
|
|
PR_STATIC_CALLBACK(void *) EventHandler(PLEvent *plevent)
|
|
{
|
|
nsOutputStreamReadyEvent *ev = (nsOutputStreamReadyEvent *) plevent;
|
|
if (ev->mCallback)
|
|
ev->mCallback->OnOutputStreamReady(ev->mStream);
|
|
ev->mCallback = 0;
|
|
return NULL;
|
|
}
|
|
|
|
PR_STATIC_CALLBACK(void) EventCleanup(PLEvent *ev)
|
|
{
|
|
nsOutputStreamReadyEvent *event = (nsOutputStreamReadyEvent *) ev;
|
|
NS_RELEASE(event);
|
|
}
|
|
};
|
|
|
|
NS_IMPL_THREADSAFE_ISUPPORTS1(nsOutputStreamReadyEvent,
|
|
nsIOutputStreamCallback)
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
NS_COM nsresult
|
|
NS_NewInputStreamReadyEvent(nsIInputStreamCallback **event,
|
|
nsIInputStreamCallback *callback,
|
|
nsIEventTarget *target)
|
|
{
|
|
nsInputStreamReadyEvent *ev = new nsInputStreamReadyEvent(callback, target);
|
|
if (!ev)
|
|
return NS_ERROR_OUT_OF_MEMORY;
|
|
NS_ADDREF(*event = ev);
|
|
return NS_OK;
|
|
}
|
|
|
|
NS_COM nsresult
|
|
NS_NewOutputStreamReadyEvent(nsIOutputStreamCallback **event,
|
|
nsIOutputStreamCallback *callback,
|
|
nsIEventTarget *target)
|
|
{
|
|
nsOutputStreamReadyEvent *ev = new nsOutputStreamReadyEvent(callback, target);
|
|
if (!ev)
|
|
return NS_ERROR_OUT_OF_MEMORY;
|
|
NS_ADDREF(*event = ev);
|
|
return NS_OK;
|
|
}
|
|
|
|
//-----------------------------------------------------------------------------
|
|
// NS_AsyncCopy implementation
|
|
|
|
// abstract stream copier...
|
|
class nsAStreamCopier : public nsIInputStreamCallback
|
|
, public nsIOutputStreamCallback
|
|
{
|
|
public:
|
|
NS_DECL_ISUPPORTS
|
|
|
|
nsAStreamCopier()
|
|
: mLock(nsnull)
|
|
, mCallback(nsnull)
|
|
, mClosure(nsnull)
|
|
, mChunkSize(0)
|
|
, mEventInProcess(PR_FALSE)
|
|
, mEventIsPending(PR_FALSE)
|
|
{
|
|
}
|
|
|
|
// virtual since subclasses call superclass Release()
|
|
virtual ~nsAStreamCopier()
|
|
{
|
|
if (mLock)
|
|
PR_DestroyLock(mLock);
|
|
}
|
|
|
|
// kick off the async copy...
|
|
nsresult Start(nsIInputStream *source,
|
|
nsIOutputStream *sink,
|
|
nsIEventTarget *target,
|
|
nsAsyncCopyCallbackFun callback,
|
|
void *closure,
|
|
PRUint32 chunksize)
|
|
{
|
|
mSource = source;
|
|
mSink = sink;
|
|
mTarget = target;
|
|
mCallback = callback;
|
|
mClosure = closure;
|
|
mChunkSize = chunksize;
|
|
|
|
mLock = PR_NewLock();
|
|
if (!mLock)
|
|
return NS_ERROR_OUT_OF_MEMORY;
|
|
|
|
mAsyncSource = do_QueryInterface(mSource);
|
|
mAsyncSink = do_QueryInterface(mSink);
|
|
|
|
return PostContinuationEvent();
|
|
}
|
|
|
|
// implemented by subclasses, returns number of bytes copied and
|
|
// sets source and sink condition before returning.
|
|
virtual PRUint32 DoCopy(nsresult *sourceCondition, nsresult *sinkCondition) = 0;
|
|
|
|
void Process()
|
|
{
|
|
if (!mSource || !mSink)
|
|
return;
|
|
|
|
nsresult sourceCondition, sinkCondition;
|
|
|
|
// ok, copy data from source to sink.
|
|
for (;;) {
|
|
PRUint32 n = DoCopy(&sourceCondition, &sinkCondition);
|
|
if (NS_FAILED(sourceCondition) || NS_FAILED(sinkCondition) || n == 0) {
|
|
if (sourceCondition == NS_BASE_STREAM_WOULD_BLOCK && mAsyncSource) {
|
|
// need to wait for more data from source. while waiting for
|
|
// more source data, be sure to observe failures on output end.
|
|
mAsyncSource->AsyncWait(this, 0, 0, nsnull);
|
|
|
|
if (mAsyncSink)
|
|
mAsyncSink->AsyncWait(this,
|
|
nsIAsyncOutputStream::WAIT_CLOSURE_ONLY,
|
|
0, nsnull);
|
|
}
|
|
else if (sinkCondition == NS_BASE_STREAM_WOULD_BLOCK && mAsyncSink) {
|
|
// need to wait for more room in the sink. while waiting for
|
|
// more room in the sink, be sure to observer failures on the
|
|
// input end.
|
|
mAsyncSink->AsyncWait(this, 0, 0, nsnull);
|
|
|
|
if (mAsyncSource)
|
|
mAsyncSource->AsyncWait(this,
|
|
nsIAsyncInputStream::WAIT_CLOSURE_ONLY,
|
|
0, nsnull);
|
|
}
|
|
else {
|
|
// close source
|
|
if (mAsyncSource)
|
|
mAsyncSource->CloseWithStatus(sinkCondition);
|
|
else
|
|
mSource->Close();
|
|
mAsyncSource = nsnull;
|
|
mSource = nsnull;
|
|
|
|
// close sink
|
|
if (mAsyncSink)
|
|
mAsyncSink->CloseWithStatus(sourceCondition);
|
|
else
|
|
mSink->Close();
|
|
mAsyncSink = nsnull;
|
|
mSink = nsnull;
|
|
|
|
// notify state complete...
|
|
if (mCallback) {
|
|
nsresult status = sourceCondition;
|
|
if (NS_SUCCEEDED(status))
|
|
status = sinkCondition;
|
|
if (status == NS_BASE_STREAM_CLOSED)
|
|
status = NS_OK;
|
|
mCallback(mClosure, status);
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
NS_IMETHOD OnInputStreamReady(nsIAsyncInputStream *source)
|
|
{
|
|
PostContinuationEvent();
|
|
return NS_OK;
|
|
}
|
|
|
|
NS_IMETHOD OnOutputStreamReady(nsIAsyncOutputStream *sink)
|
|
{
|
|
PostContinuationEvent();
|
|
return NS_OK;
|
|
}
|
|
|
|
PR_STATIC_CALLBACK(void*) HandleContinuationEvent(PLEvent *event)
|
|
{
|
|
nsAStreamCopier *self = (nsAStreamCopier *) event->owner;
|
|
self->Process();
|
|
|
|
// clear "in process" flag and post any pending continuation event
|
|
nsAutoLock lock(self->mLock);
|
|
self->mEventInProcess = PR_FALSE;
|
|
if (self->mEventIsPending) {
|
|
self->mEventIsPending = PR_FALSE;
|
|
self->PostContinuationEvent_Locked();
|
|
}
|
|
return nsnull;
|
|
}
|
|
|
|
PR_STATIC_CALLBACK(void) DestroyContinuationEvent(PLEvent *event)
|
|
{
|
|
nsAStreamCopier *self = (nsAStreamCopier *) event->owner;
|
|
NS_RELEASE(self);
|
|
delete event;
|
|
}
|
|
|
|
nsresult PostContinuationEvent()
|
|
{
|
|
// we cannot post a continuation event if there is currently
|
|
// an event in process. doing so could result in Process being
|
|
// run simultaneously on multiple threads, so we mark the event
|
|
// as pending, and if an event is already in process then we
|
|
// just let that existing event take care of posting the real
|
|
// continuation event.
|
|
|
|
nsAutoLock lock(mLock);
|
|
return PostContinuationEvent_Locked();
|
|
}
|
|
|
|
nsresult PostContinuationEvent_Locked()
|
|
{
|
|
nsresult rv = NS_OK;
|
|
if (mEventInProcess)
|
|
mEventIsPending = PR_TRUE;
|
|
else {
|
|
PLEvent *event = new PLEvent;
|
|
if (!event)
|
|
rv = NS_ERROR_OUT_OF_MEMORY;
|
|
else {
|
|
NS_ADDREF_THIS();
|
|
PL_InitEvent(event, this,
|
|
HandleContinuationEvent,
|
|
DestroyContinuationEvent);
|
|
|
|
rv = mTarget->PostEvent(event);
|
|
if (NS_SUCCEEDED(rv))
|
|
mEventInProcess = PR_TRUE;
|
|
else {
|
|
NS_ERROR("unable to post continuation event");
|
|
PL_DestroyEvent(event);
|
|
}
|
|
}
|
|
}
|
|
return rv;
|
|
}
|
|
|
|
protected:
|
|
nsCOMPtr<nsIInputStream> mSource;
|
|
nsCOMPtr<nsIOutputStream> mSink;
|
|
nsCOMPtr<nsIAsyncInputStream> mAsyncSource;
|
|
nsCOMPtr<nsIAsyncOutputStream> mAsyncSink;
|
|
nsCOMPtr<nsIEventTarget> mTarget;
|
|
PRLock *mLock;
|
|
nsAsyncCopyCallbackFun mCallback;
|
|
void *mClosure;
|
|
PRUint32 mChunkSize;
|
|
PRPackedBool mEventInProcess;
|
|
PRPackedBool mEventIsPending;
|
|
};
|
|
|
|
NS_IMPL_THREADSAFE_ISUPPORTS2(nsAStreamCopier,
|
|
nsIInputStreamCallback,
|
|
nsIOutputStreamCallback)
|
|
|
|
class nsStreamCopierIB : public nsAStreamCopier
|
|
{
|
|
public:
|
|
nsStreamCopierIB() : nsAStreamCopier() {}
|
|
virtual ~nsStreamCopierIB() {}
|
|
|
|
struct ReadSegmentsState {
|
|
nsIOutputStream *mSink;
|
|
nsresult mSinkCondition;
|
|
};
|
|
|
|
static NS_METHOD ConsumeInputBuffer(nsIInputStream *inStr,
|
|
void *closure,
|
|
const char *buffer,
|
|
PRUint32 offset,
|
|
PRUint32 count,
|
|
PRUint32 *countWritten)
|
|
{
|
|
ReadSegmentsState *state = (ReadSegmentsState *) closure;
|
|
|
|
nsresult rv = state->mSink->Write(buffer, count, countWritten);
|
|
if (NS_FAILED(rv))
|
|
state->mSinkCondition = rv;
|
|
else if (*countWritten == 0)
|
|
state->mSinkCondition = NS_BASE_STREAM_CLOSED;
|
|
|
|
return state->mSinkCondition;
|
|
}
|
|
|
|
PRUint32 DoCopy(nsresult *sourceCondition, nsresult *sinkCondition)
|
|
{
|
|
ReadSegmentsState state;
|
|
state.mSink = mSink;
|
|
state.mSinkCondition = NS_OK;
|
|
|
|
PRUint32 n;
|
|
*sourceCondition =
|
|
mSource->ReadSegments(ConsumeInputBuffer, &state, mChunkSize, &n);
|
|
*sinkCondition = state.mSinkCondition;
|
|
return n;
|
|
}
|
|
};
|
|
|
|
class nsStreamCopierOB : public nsAStreamCopier
|
|
{
|
|
public:
|
|
nsStreamCopierOB() : nsAStreamCopier() {}
|
|
virtual ~nsStreamCopierOB() {}
|
|
|
|
struct WriteSegmentsState {
|
|
nsIInputStream *mSource;
|
|
nsresult mSourceCondition;
|
|
};
|
|
|
|
static NS_METHOD FillOutputBuffer(nsIOutputStream *outStr,
|
|
void *closure,
|
|
char *buffer,
|
|
PRUint32 offset,
|
|
PRUint32 count,
|
|
PRUint32 *countRead)
|
|
{
|
|
WriteSegmentsState *state = (WriteSegmentsState *) closure;
|
|
|
|
nsresult rv = state->mSource->Read(buffer, count, countRead);
|
|
if (NS_FAILED(rv))
|
|
state->mSourceCondition = rv;
|
|
else if (*countRead == 0)
|
|
state->mSourceCondition = NS_BASE_STREAM_CLOSED;
|
|
|
|
return state->mSourceCondition;
|
|
}
|
|
|
|
PRUint32 DoCopy(nsresult *sourceCondition, nsresult *sinkCondition)
|
|
{
|
|
WriteSegmentsState state;
|
|
state.mSource = mSource;
|
|
state.mSourceCondition = NS_OK;
|
|
|
|
PRUint32 n;
|
|
*sinkCondition =
|
|
mSink->WriteSegments(FillOutputBuffer, &state, mChunkSize, &n);
|
|
*sourceCondition = state.mSourceCondition;
|
|
return n;
|
|
}
|
|
};
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
NS_COM nsresult
|
|
NS_AsyncCopy(nsIInputStream *source,
|
|
nsIOutputStream *sink,
|
|
nsIEventTarget *target,
|
|
nsAsyncCopyMode mode,
|
|
PRUint32 chunkSize,
|
|
nsAsyncCopyCallbackFun callback,
|
|
void *closure)
|
|
{
|
|
NS_ASSERTION(target, "non-null target required");
|
|
|
|
nsresult rv;
|
|
nsAStreamCopier *copier;
|
|
|
|
if (mode == NS_ASYNCCOPY_VIA_READSEGMENTS)
|
|
copier = new nsStreamCopierIB();
|
|
else
|
|
copier = new nsStreamCopierOB();
|
|
|
|
if (!copier)
|
|
return NS_ERROR_OUT_OF_MEMORY;
|
|
|
|
// Start() takes an owning ref to the copier...
|
|
NS_ADDREF(copier);
|
|
rv = copier->Start(source, sink, target, callback, closure, chunkSize);
|
|
NS_RELEASE(copier);
|
|
|
|
return rv;
|
|
}
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
static NS_METHOD
|
|
TestInputStream(nsIInputStream *inStr,
|
|
void *closure,
|
|
const char *buffer,
|
|
PRUint32 offset,
|
|
PRUint32 count,
|
|
PRUint32 *countWritten)
|
|
{
|
|
PRBool *result = NS_REINTERPRET_CAST(PRBool *, closure);
|
|
*result = PR_TRUE;
|
|
return NS_ERROR_ABORT; // don't call me anymore
|
|
}
|
|
|
|
NS_COM PRBool
|
|
NS_InputStreamIsBuffered(nsIInputStream *stream)
|
|
{
|
|
PRBool result = PR_FALSE;
|
|
PRUint32 n;
|
|
stream->ReadSegments(TestInputStream, &result, 1, &n);
|
|
return result;
|
|
}
|
|
|
|
static NS_METHOD
|
|
TestOutputStream(nsIOutputStream *outStr,
|
|
void *closure,
|
|
char *buffer,
|
|
PRUint32 offset,
|
|
PRUint32 count,
|
|
PRUint32 *countRead)
|
|
{
|
|
PRBool *result = NS_REINTERPRET_CAST(PRBool *, closure);
|
|
*result = PR_TRUE;
|
|
return NS_ERROR_ABORT; // don't call me anymore
|
|
}
|
|
|
|
NS_COM PRBool
|
|
NS_OutputStreamIsBuffered(nsIOutputStream *stream)
|
|
{
|
|
PRBool result = PR_FALSE;
|
|
PRUint32 n;
|
|
stream->WriteSegments(TestOutputStream, &result, 1, &n);
|
|
return result;
|
|
}
|