mozilla
/
pjs
зеркало из https://github.com/mozilla/pjs.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность
pjs/netwerk/test/TestSocketTransport.cpp

721 строка
16 KiB
C++
Исходник Ответственный История

/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*-
*
* The contents of this file are subject to the Netscape 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/NPL/
*
* 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.org code.
*
* The Initial Developer of the Original Code is Netscape
* Communications Corporation. Portions created by Netscape are
* Copyright (C) 1998 Netscape Communications Corporation. All
* Rights Reserved.
*
* Contributor(s):
*/
#include <stdio.h>
#ifdef WIN32
//#define USE_TIMERS // Only use nsITimer on Windows (for now...)
#include <windows.h>
#endif
#ifdef XP_OS2
#include <os2.h>
#endif
#include "nspr.h"
#ifdef XP_MAC
#include "pprio.h" // PR_Init_Log
#endif
#define NSPIPE2
#include "nsISocketTransportService.h"
#include "nsIEventQueueService.h"
#include "nsIServiceManager.h"
#include "nsIChannel.h"
#include "nsIStreamObserver.h"
#include "nsIStreamListener.h"
#ifndef NSPIPE2
#include "nsIBuffer.h"
#else
#include "nsIPipe.h"
#endif
#include "nsIBufferInputStream.h"
#include "nsIBufferOutputStream.h"
#include "nsIRunnable.h"
#include "nsIThread.h"
#include "nsITimer.h"
#include "nsIInterfaceRequestor.h"
#include "nsIProgressEventSink.h"
#include "nsCRT.h"
#if defined(XP_MAC)
#include "macstdlibextras.h"
// Set up the toolbox and (if DEBUG) the console. Do this in a static initializer,
// to make it as unlikely as possible that somebody calls printf() before we get initialized.
static struct MacInitializer { MacInitializer() { InitializeMacToolbox(); InitializeSIOUX(true); } } gInitializer;
#endif // XP_MAC
// forward declarations...
class TestConnection;
static NS_DEFINE_CID(kSocketTransportServiceCID, NS_SOCKETTRANSPORTSERVICE_CID);
static NS_DEFINE_CID(kEventQueueServiceCID, NS_EVENTQUEUESERVICE_CID);
//static PRTime gElapsedTime;
static int gKeepRunning = 1;
#define NUM_TEST_THREADS 15
#define TRANSFER_AMOUNT 64
static TestConnection* gConnections[NUM_TEST_THREADS];
static nsIThread* gThreads[NUM_TEST_THREADS];
//static nsITimer* gPeriodicTimer;
static PRBool gVerbose = PR_TRUE;
void Pump_PLEvents(nsIEventQueueService * eventQService);
void Pump_PLEvents(nsIEventQueueService * eventQService)
{
nsIEventQueue* eventQ = nsnull;
eventQService->GetThreadEventQueue(NS_CURRENT_THREAD, &eventQ);
while ( gKeepRunning ) {
#ifdef WIN32
MSG msg;
if (GetMessage(&msg, NULL, 0, 0)) {
TranslateMessage(&msg);
DispatchMessage(&msg);
} else {
gKeepRunning = FALSE;
}
#elif defined(XP_OS2)
QMSG qmsg;
if (WinGetMsg(0, &qmsg, 0, 0, 0))
WinDispatchMsg(0, &qmsg);
else
gKeepRunning = FALSE;
#else
nsresult rv;
PLEvent *gEvent;
rv = eventQ->GetEvent(&gEvent);
rv = eventQ->HandleEvent(gEvent);
#endif /* !WIN32 */
}
}
// -----
//
// TestConnection class...
//
// -----
class TestConnection : public nsIRunnable,
public nsIStreamListener,
public nsIInterfaceRequestor,
public nsIProgressEventSink
{
public:
TestConnection(const char* aHostName, PRInt32 aPort,
PRBool aAsyncFlag, PRBool testAsyncRead);
virtual ~TestConnection();
NS_DECL_ISUPPORTS
NS_DECL_NSIRUNNABLE
NS_DECL_NSISTREAMOBSERVER
NS_DECL_NSISTREAMLISTENER
// TestConnection methods...
nsresult WriteBuffer(void);
nsresult ReadBuffer(void);
nsresult Process(void);
nsresult Suspend(void);
nsresult Resume(void);
void SetAsyncOpenCompleted() { mAsyncOpenCompleted = PR_TRUE; }
NS_IMETHOD GetInterface(const nsIID & uuid, void * *result) {
if (uuid.Equals(NS_GET_IID(nsIProgressEventSink))) {
*result = (nsIProgressEventSink*)this;
NS_ADDREF_THIS();
return NS_OK;
}
return NS_NOINTERFACE;
}
NS_IMETHOD OnProgress(nsIChannel *channel, nsISupports *ctxt, PRUint32 aProgress, PRUint32 aProgressMax) {
putc('+', stderr);
return NS_OK;
}
NS_IMETHOD OnStatus(nsIChannel *channel, nsISupports *ctxt, const PRUnichar *aMsg) {
putc('?', stderr);
return NS_OK;
}
protected:
#ifndef NSPIPE2
nsIBuffer* mBuffer;
#else
nsIBufferOutputStream* mOut;
#endif
nsIBufferInputStream* mStream;
nsIInputStream* mInStream;
nsIOutputStream* mOutStream;
nsIChannel* mTransport;
PRBool mIsAsync;
PRInt32 mBufferLength;
char mBufferChar;
PRInt32 mBytesRead;
PRBool mTestAsyncOpen;
PRBool mAsyncOpenCompleted;
};
////////////////////////////////////////////////////////////////////////////////
class TestConnectionOpenObserver : public nsIStreamObserver
{
public:
NS_DECL_ISUPPORTS
NS_DECL_NSISTREAMOBSERVER
TestConnectionOpenObserver(TestConnection* test)
: mTestConnection(test) {
NS_INIT_REFCNT();
}
virtual ~TestConnectionOpenObserver() {}
protected:
TestConnection* mTestConnection;
};
NS_IMPL_ISUPPORTS1(TestConnectionOpenObserver, nsIStreamObserver);
NS_IMETHODIMP
TestConnectionOpenObserver::OnStartRequest(nsIChannel* channel, nsISupports* context)
{
if (gVerbose)
printf("\n+++ TestConnectionOpenObserver::OnStartRequest +++. Context = %p\n", context);
mTestConnection->SetAsyncOpenCompleted();
return NS_OK;
}
NS_IMETHODIMP
TestConnectionOpenObserver::OnStopRequest(nsIChannel* channel,
nsISupports* context,
nsresult aStatus,
const PRUnichar* aMsg)
{
if (gVerbose || NS_FAILED(aStatus))
printf("\n+++ TestConnectionOpenObserver::OnStopRequest (status = %x) +++."
"\tContext = %p\n",
aStatus, context);
return NS_OK;
}
////////////////////////////////////////////////////////////////////////////////
NS_IMETHODIMP
TestConnection::OnStartRequest(nsIChannel* channel, nsISupports* context)
{
NS_ASSERTION(!mTestAsyncOpen || mAsyncOpenCompleted, "AsyncOpen failed");
if (gVerbose)
printf("\n+++ TestConnection::OnStartRequest +++. Context = %p\n", context);
return NS_OK;
}
NS_IMETHODIMP
TestConnection::OnDataAvailable(nsIChannel* channel, nsISupports* context,
nsIInputStream *aIStream,
PRUint32 aSourceOffset,
PRUint32 aLength)
{
nsresult rv;
NS_ASSERTION(!mTestAsyncOpen || mAsyncOpenCompleted, "AsyncOpen failed");
char buf[TRANSFER_AMOUNT];
PRUint32 amt;
if (gVerbose)
printf("\n+++ TestConnection::OnDavaAvailable +++."
"\tContext = %p length = %d\n",
context, aLength);
while (aLength > 0) {
PRInt32 cnt = PR_MIN(TRANSFER_AMOUNT, aLength);
rv = aIStream->Read(buf, cnt, &amt);
NS_ASSERTION(NS_SUCCEEDED(rv), "Read failed");
mBytesRead += amt;
aLength -= amt;
buf[amt] = '\0';
if (gVerbose)
puts(buf);
}
if (mBufferLength == mBytesRead) {
mBytesRead = 0;
WriteBuffer();
}
return NS_OK;
}
NS_IMETHODIMP
TestConnection::OnStopRequest(nsIChannel* channel,
nsISupports* context,
nsresult aStatus,
const PRUnichar* aMsg)
{
NS_ASSERTION(!mTestAsyncOpen || mAsyncOpenCompleted, "AsyncOpen failed");
if (gVerbose || NS_FAILED(aStatus))
printf("\n+++ TestConnection::OnStopRequest (status = %x) +++."
"\tContext = %p\n",
aStatus, context);
return NS_OK;
}
TestConnection::TestConnection(const char* aHostName, PRInt32 aPort,
PRBool aAsyncFlag, PRBool testAsyncRead)
{
nsresult rv;
NS_INIT_REFCNT();
mIsAsync = aAsyncFlag;
mTestAsyncOpen = testAsyncRead;
mAsyncOpenCompleted = PR_FALSE;
mBufferLength = TRANSFER_AMOUNT;
mBufferChar = 'a';
mBytesRead = 0;
mTransport = nsnull;
#ifndef NSPIPE2
mBuffer = nsnull;
#else
mOut = nsnull;
#endif
mStream = nsnull;
mInStream = nsnull;
mOutStream = nsnull;
// Create a socket transport...
NS_WITH_SERVICE(nsISocketTransportService, sts, kSocketTransportServiceCID, &rv);
if (NS_SUCCEEDED(rv)) {
rv = sts->CreateTransport(aHostName, aPort, aHostName, 0, 0, &mTransport);
if (NS_SUCCEEDED(rv)) {
// Set up the notification callbacks to provide a progress event sink.
// That way we exercise the progress notification proxy code.
rv = mTransport->SetNotificationCallbacks(this);
}
}
if (NS_SUCCEEDED(rv)) {
if (mIsAsync) {
// Create a stream for the data being written to the server...
if (NS_SUCCEEDED(rv)) {
#ifndef NSPIPE2
rv = NS_NewBuffer(&mBuffer, 1024, 4096, nsnull);
rv = NS_NewBufferInputStream(&mStream, mBuffer);
#else
rv = NS_NewPipe(&mStream, &mOut, nsnull, 1024, 4096);
#endif
}
}
// Synchronous transport...
else {
rv = mTransport->OpenInputStream(&mInStream);
rv = mTransport->OpenOutputStream(&mOutStream);
}
}
}
TestConnection::~TestConnection()
{
NS_IF_RELEASE(mTransport);
// Async resources...
NS_IF_RELEASE(mStream);
#ifndef NSPIPE2
NS_IF_RELEASE(mBuffer);
#else
NS_IF_RELEASE(mOut);
#endif
// Sync resources...
NS_IF_RELEASE(mInStream);
NS_IF_RELEASE(mOutStream);
}
NS_IMPL_THREADSAFE_ISUPPORTS5(TestConnection,
nsIRunnable,
nsIStreamListener,
nsIStreamObserver,
nsIInterfaceRequestor,
nsIProgressEventSink);
NS_IMETHODIMP
TestConnection::Run(void)
{
nsresult rv = NS_OK;
// Create the Event Queue for this thread...
NS_WITH_SERVICE(nsIEventQueueService, eventQService, kEventQueueServiceCID, &rv);
if (NS_FAILED(rv)) return rv;
rv = eventQService->CreateThreadEventQueue();
if (NS_FAILED(rv)) return rv;
//
// Make sure that all resources were allocated in the constructor...
//
if (!mTransport) {
rv = NS_ERROR_FAILURE;
}
if (NS_SUCCEEDED(rv)) {
if (mIsAsync) {
if (mTestAsyncOpen) {
TestConnectionOpenObserver* obs = new TestConnectionOpenObserver(this);
if (obs == nsnull)
return NS_ERROR_OUT_OF_MEMORY;
NS_ADDREF(obs);
rv = mTransport->AsyncOpen(obs, nsnull);
NS_RELEASE(obs);
if (NS_FAILED(rv)) {
printf("Error: AsyncOpen failed...");
}
}
//
// Initiate an async read...
//
rv = mTransport->AsyncRead(this, mTransport);
if (NS_FAILED(rv)) {
printf("Error: AsyncRead failed...");
} else {
rv = WriteBuffer();
}
Pump_PLEvents(eventQService);
}
else {
while (NS_SUCCEEDED(rv)) {
rv = WriteBuffer();
if (NS_SUCCEEDED(rv)) {
rv = ReadBuffer();
}
}
}
}
if (gVerbose)
printf("Transport thread exiting...\n");
return rv;
}
nsresult TestConnection::WriteBuffer(void)
{
nsresult rv;
char *buffer;
PRInt32 size;
PRUint32 bytesWritten;
if (mBufferChar == 'z') {
mBufferChar = 'a';
} else {
mBufferChar++;
}
if (gVerbose)
printf("\n+++ Request is: %c. Context = %p\n", mBufferChar, mTransport);
// Create and fill a test buffer of data...
buffer = (char*)PR_Malloc(mBufferLength + 4);
if (buffer) {
for (size=0; size<mBufferLength-2; size++) {
buffer[size] = mBufferChar;
}
buffer[size++] = '\r';
buffer[size++] = '\n';
buffer[size] = 0;
//
// Async case...
//
if (mStream) {
#if 0
rv = mStream->Fill(buffer, size, &bytesWritten);
#else
#ifndef NSPIPE2
rv = mBuffer->Write(buffer, size, &bytesWritten);
#else
rv = mOut->Write(buffer, size, &bytesWritten);
#endif
#endif
// Write the buffer to the server...
if (NS_SUCCEEDED(rv)) {
rv = mTransport->SetTransferCount(bytesWritten);
if (NS_SUCCEEDED(rv)) {
rv = mTransport->AsyncWrite(mStream, /* mOutputObserver */ nsnull, mTransport);
}
}
// Wait for the write to complete...
if (NS_FAILED(rv)) {
printf("Error: AsyncWrite failed...");
}
}
//
// Synchronous case...
//
else if (mOutStream) {
rv = mOutStream->Write(buffer, size, &bytesWritten);
NS_ASSERTION((PRUint32)size == bytesWritten, "Not enough was written...");
}
PR_Free(buffer);
} else {
rv = NS_ERROR_OUT_OF_MEMORY;
}
return rv;
}
nsresult TestConnection::ReadBuffer(void)
{
nsresult rv = NS_OK;
//
// Synchronous case...
//
if (mInStream) {
char *buffer;
PRUint32 bytesRead;
buffer = (char*)PR_Malloc(mBufferLength + 4);
if (buffer) {
rv = mInStream->Read(buffer, mBufferLength, &bytesRead);
if (NS_SUCCEEDED(rv) && bytesRead > 0) {
buffer[bytesRead] = '\0';
if (gVerbose)
printf("TestConnection::ReadBuffer. Read %d bytes\n", bytesRead);
puts(buffer);
}
PR_Free(buffer);
}
}
return rv;
}
nsresult TestConnection::Process(void)
{
return NS_OK;
}
nsresult TestConnection::Suspend(void)
{
nsresult rv;
if (mTransport) {
rv = mTransport->Suspend();
} else {
rv = NS_ERROR_FAILURE;
}
return rv;
}
nsresult TestConnection::Resume(void)
{
nsresult rv;
if (mTransport) {
rv = mTransport->Resume();
} else {
rv = NS_ERROR_FAILURE;
}
return rv;
}
#if defined(USE_TIMERS)
void TimerCallback(nsITimer* aTimer, void* aClosure)
{
static PRBool flag = PR_FALSE;
int i;
if (flag) {
printf("Resuming connections...\n");
} else {
printf("Suspending connections...\n");
}
for (i=0; i<NUM_TEST_THREADS; i++) {
TestConnection* connection;
connection = gConnections[i];
if (connection) {
if (flag) {
connection->Resume();
} else {
connection->Suspend();
}
}
}
flag = !flag;
NS_RELEASE(gPeriodicTimer);
if (NS_OK == NS_NewTimer(&gPeriodicTimer)) {
gPeriodicTimer->Init(TimerCallback, nsnull, 1000*5);
}
}
#endif /* USE_TIMERS */
nsresult NS_AutoregisterComponents();
nsresult NS_AutoregisterComponents()
{
nsresult rv = nsComponentManager::AutoRegister(nsIComponentManager::NS_Startup, NULL /* default */);
return rv;
}
int
main(int argc, char* argv[])
{
nsresult rv;
// -----
//
// Parse the command line args...
//
// -----
/// if (argc < 3) {
/// printf("usage: %s [-sync|-asyncopen|-silent] <host> <path>\n", argv[0]);
/// return -1;
/// }
PRBool bIsAsync = PR_TRUE;
PRBool bTestAsyncOpen = PR_FALSE;
char* hostName = nsnull;
int i;
for (i=1; i<argc; i++) {
// Turn on synchronous mode...
if (PL_strcasecmp(argv[i], "-sync") == 0) {
bIsAsync = PR_FALSE;
continue;
}
if (PL_strcasecmp(argv[i], "-asyncopen") == 0) {
bTestAsyncOpen = PR_TRUE;
continue;
}
if (PL_strcasecmp(argv[i], "-silent") == 0) {
gVerbose = PR_FALSE;
continue;
}
hostName = argv[i];
}
if (!hostName) {
hostName = "chainsaw";
}
printf("Using %s as echo server...\n", hostName);
#ifdef XP_MAC
(void) PR_PutEnv("NSPR_LOG_MODULES=nsSocketTransport:5");
(void) PR_PutEnv("NSPR_LOG_FILE=nspr.log");
PR_Init_Log();
#endif
// -----
//
// Initialize XPCom...
//
// -----
rv = NS_AutoregisterComponents();
if (NS_FAILED(rv)) return rv;
// Create the Event Queue for this thread...
NS_WITH_SERVICE(nsIEventQueueService, eventQService, kEventQueueServiceCID, &rv);
if (NS_FAILED(rv)) return rv;
rv = eventQService->CreateThreadEventQueue();
if (NS_FAILED(rv)) return rv;
//
// Create the connections and threads...
//
for (i=0; i<NUM_TEST_THREADS; i++) {
gConnections[i] = new TestConnection(hostName, 7, bIsAsync, bTestAsyncOpen);
rv = NS_NewThread(&gThreads[i], gConnections[i], 0, PR_JOINABLE_THREAD);
}
#if defined(USE_TIMERS)
//
// Start up the timer to test Suspend/Resume APIs on the transport...
//
if (NS_OK == NS_NewTimer(&gPeriodicTimer)) {
gPeriodicTimer->Init(TimerCallback, nsnull, 1000);
}
#endif /* USE_TIMERS */
// Enter the message pump to allow the URL load to proceed.
Pump_PLEvents(eventQService);
PRTime endTime;
endTime = PR_Now();
// printf("Elapsed time: %d\n", (PRInt32)(endTime/1000UL - gElapsedTime/1000UL));
return 0;
}
Ссылка в новой задаче Показать git blame Копировать постоянную ссылку