[wxWidgets.git] / samples / sockets / baseclient.cpp

/////////////////////////////////////////////////////////////////////////////
// Name:        samples/sockbase/client.cpp
// Purpose:     Sockets sample for wxBase
// Author:      Lukasz Michalski
// Modified by:
// Created:     27.06.2005
// RCS-ID:      $Id$
// Copyright:   (c) 2005 Lukasz Michalski <lmichalski@sf.net>
// Licence:     wxWindows license
/////////////////////////////////////////////////////////////////////////////

// ============================================================================
// declarations
// ============================================================================

// ----------------------------------------------------------------------------
// headers
// ----------------------------------------------------------------------------

#include "wx/wx.h"
#include "wx/socket.h"
#include "wx/event.h"
#include "wx/list.h"
#include "wx/cmdline.h"
#include "wx/ffile.h"
#include "wx/datetime.h"
#include "wx/timer.h"
#include "wx/thread.h"

const wxEventType wxEVT_WORKER = wxNewEventType();
#define EVT_WORKER(func) DECLARE_EVENT_TABLE_ENTRY( wxEVT_WORKER, -1, -1, (wxObjectEventFunction) (wxEventFunction) (WorkerEventFunction) & func, (wxObject *) NULL ),

const int timeout_val = 1000;

class WorkerEvent : public wxEvent {
public:
    typedef enum {
        CONNECTING,
        SENDING,
        RECEIVING,
        DISCONNECTING,
        DONE
    } evt_type;
    WorkerEvent(void* pSender, evt_type type)
    {
        SetId(-1);
        SetEventType(wxEVT_WORKER);
        m_sender = pSender;
        m_eventType = type;
        m_isFailed = false;
    }

    void setFailed() { m_isFailed = true; }
    bool isFailed() const { return m_isFailed; }

    virtual wxEvent* Clone() const
    {
        return new WorkerEvent(*this);
    }
    void* m_sender;
    bool m_isFailed;
    wxString m_workerIdent;
    evt_type m_eventType;
};

typedef void (wxEvtHandler::*WorkerEventFunction)(WorkerEvent&);

class ThreadWorker;
class EventWorker;

WX_DECLARE_LIST(ThreadWorker, TList);
WX_DECLARE_LIST(EventWorker, EList);

class Client : public wxApp {
    DECLARE_EVENT_TABLE();
public:
    void RemoveEventWorker(EventWorker* p_worker);
private:
    typedef enum
    {
      THREADS,
      EVENTS
    } workMode;

    typedef enum
    {
      SEND_RANDOM,
      SEND_MESSAGE,
      STRESS_TEST
    } sendType;

    workMode m_workMode;
    sendType m_sendType;
    wxString m_message;
    wxString m_host;
    long m_stressWorkers;

    virtual bool OnInit();
    virtual int OnRun();
    virtual int OnExit();
    void OnInitCmdLine(wxCmdLineParser& pParser);
    bool OnCmdLineParsed(wxCmdLineParser& pParser);
    void OnWorkerEvent(WorkerEvent& pEvent);
    void OnTimerEvent(wxTimerEvent& pEvent);

    void StartWorker(workMode pMode, const wxString& pMessage);
    void StartWorker(workMode pMode);
    char* CreateBuffer(int *msgsize);

    void dumpStatistics();

    TList m_threadWorkers;
    EList m_eventWorkers;

    unsigned m_statConnecting;
    unsigned m_statSending;
    unsigned m_statReceiving;
    unsigned m_statDisconnecting;
    unsigned m_statDone;
    unsigned m_statFailed;

    wxTimer mTimer;
};

DECLARE_APP(Client);

class ThreadWorker : public wxThread
{
public:
    ThreadWorker(const wxString& p_host, char* p_buf, int p_size);
    virtual ExitCode Entry();
private:
    wxString m_host;
    wxSocketClient* m_clientSocket;
    char* m_inbuf;
    char* m_outbuf;
    int m_outsize;
    int m_insize;
    wxString m_workerIdent;
};

class EventWorker : public wxEvtHandler
{
    DECLARE_EVENT_TABLE();
public:
    EventWorker(const wxString& p_host, char* p_buf, int p_size);
    void Run();
    virtual ~EventWorker();
private:
    wxString m_host;
    wxSocketClient* m_clientSocket;
    char* m_inbuf;
    char* m_outbuf;
    int m_outsize;
    int m_written;
    int m_insize;
    int m_readed;

    WorkerEvent::evt_type m_currentType;
    bool m_doneSent;
    wxIPV4address m_localaddr;

    void OnSocketEvent(wxSocketEvent& pEvent);
    void SendEvent(bool failed);
};

/******************* Implementation ******************/
IMPLEMENT_APP_CONSOLE(Client);

#include <wx/listimpl.cpp>
WX_DEFINE_LIST(TList);
WX_DEFINE_LIST(EList);

wxString
CreateIdent(const wxIPV4address& addr)
{
    return wxString::Format(wxT("%s:%d"),addr.IPAddress().c_str(),addr.Service());
}

void
Client::OnInitCmdLine(wxCmdLineParser& pParser)
{
    wxApp::OnInitCmdLine(pParser);
    pParser.AddSwitch(wxT("e"),wxT("event"),_("Use event based worker (default)"),wxCMD_LINE_PARAM_OPTIONAL);
    pParser.AddSwitch(wxT("t"),wxT("thread"),_("Use thread based worker"),wxCMD_LINE_PARAM_OPTIONAL);
    pParser.AddSwitch(wxT("r"),wxT("random"),_("Send radnom data (default)"),wxCMD_LINE_PARAM_OPTIONAL);
    pParser.AddOption(wxT("m"),wxT("message"),_("Send message from <str>"),wxCMD_LINE_VAL_STRING,wxCMD_LINE_PARAM_OPTIONAL);
    pParser.AddOption(wxT("f"),wxT("file"),_("Send contents of <file>"),wxCMD_LINE_VAL_STRING,wxCMD_LINE_PARAM_OPTIONAL);
    pParser.AddOption(wxT("H"),wxT("hostname"),_("IP or name of host to connect to"),wxCMD_LINE_VAL_STRING,wxCMD_LINE_PARAM_OPTIONAL);
    pParser.AddOption(wxT("s"),wxT("stress"),_("stress test with <num> concurrent connections"),wxCMD_LINE_VAL_NUMBER,wxCMD_LINE_PARAM_OPTIONAL);
}


bool
Client::OnCmdLineParsed(wxCmdLineParser& pParser)
{
    wxString fname;
    m_workMode = EVENTS;
    m_stressWorkers = 50;

    if (pParser.Found(_("verbose")))
    {
        wxLog::AddTraceMask(wxT("wxSocket"));
        wxLog::AddTraceMask(wxT("epolldispatcher"));
        wxLog::AddTraceMask(wxT("selectdispatcher"));
        wxLog::AddTraceMask(wxT("thread"));
        wxLog::AddTraceMask(wxT("events"));
    }

    if (pParser.Found(wxT("t")))
        m_workMode = THREADS;
    m_sendType = SEND_RANDOM;

    if (pParser.Found(wxT("m"),&m_message))
        m_sendType = SEND_MESSAGE;
    else if (pParser.Found(wxT("f"),&fname))
    {
        wxFFile file(fname);
        if (!file.IsOpened()) {
            wxLogError(wxT("Cannot open file %s"),fname.c_str());
            return false;
        };
        if (!file.ReadAll(&m_message)) {
            wxLogError(wxT("Cannot read conten of file %s"),fname.c_str());
            return false;
        };
        m_sendType = SEND_MESSAGE;
    };

    if (pParser.Found(wxT("s"),&m_stressWorkers))
        m_sendType = STRESS_TEST;

    m_host = wxT("127.0.0.1");
    pParser.Found(wxT("H"),&m_host);
    return wxApp::OnCmdLineParsed(pParser);
};

bool
Client::OnInit()
{
    if (!wxApp::OnInit())
        return false;
    srand(wxDateTime::Now().GetTicks());
    mTimer.SetOwner(this);
    m_statConnecting = 0;
    m_statSending = 0;
    m_statReceiving = 0;
    m_statDisconnecting = 0;
    m_statDone = 0;
    m_statFailed = 0;
    return true;
}

int
Client::OnRun()
{
    switch(m_sendType)
    {
        case STRESS_TEST:
            switch(m_workMode)
            {
                case THREADS:
                    for (int i = 0; i < m_stressWorkers; i++) {
                        if (m_message.empty())
                            StartWorker(THREADS);
                        else
                            StartWorker(THREADS, m_message);
                    }
                    break;
                case EVENTS:
                    for (int i = 0; i < m_stressWorkers; i++) {
                        if (m_message.empty())
                            StartWorker(EVENTS);
                        else
                            StartWorker(EVENTS, m_message);
                    }
                    break;
                default:
                    for (int i = 0; i < m_stressWorkers; i++) {
                        if (m_message.empty())
                            StartWorker(i % 5 == 0 ? THREADS : EVENTS);
                        else
                            StartWorker(i % 5 == 0 ? THREADS : EVENTS, m_message);
                    }
                break;
            }
        break;
        case SEND_MESSAGE:
            StartWorker(m_workMode,m_message);
        break;
        case SEND_RANDOM:
            StartWorker(m_workMode);
        break;
    }
    mTimer.Start(timeout_val,true);
    return wxApp::OnRun();
}

int
Client::OnExit()
{
    for(EList::compatibility_iterator it = m_eventWorkers.GetFirst(); it ; it->GetNext()) {
        delete it->GetData();
    }
    return 0;
}

// Create buffer to be sent by client. Buffer contains test indicator 
// message size and place for data
// msgsize parameter contains size of data in bytes and 
// if input value does not fit into 250 bytes then
// on exit is updated to new value that is multiply of 1024 bytes
char*
Client::CreateBuffer(int* msgsize)
{
    int bufsize = 0;
    char* buf;
    //if message should have more than 256 bytes then set it as
    //test3 for compatibility with GUI server sample
    if ((*msgsize) > 250) 
    {
        //send at least one kb of data
        int size = (*msgsize)/1024 + 1;
        //returned buffer will contain test indicator, message size in kb and data
        bufsize = size*1024+2;
        buf = new char[bufsize];
        buf[0] = 0xDE; //second byte contains size in kilobytes
        buf[1] = (char)(size);
        *msgsize = size*1024;
    }
    else
    {
        //returned buffer will contain test indicator, message size in kb and data
        bufsize = (*msgsize)+2;
        buf = new char[bufsize];
        buf[0] = 0xBE; //second byte contains size in bytes
        buf[1] = (char)(*msgsize);
    }
    return buf;
}

void
Client::StartWorker(workMode pMode) {
    int msgsize = 1 + (int) (250000.0 * (rand() / (RAND_MAX + 1.0)));
    char* buf = CreateBuffer(&msgsize);

    //fill data part of buffer with random bytes
    for (int i = 2; i < (msgsize); i++) {
        buf[i] = i % 256;
    }

    if (pMode == THREADS) {
        ThreadWorker* c = new ThreadWorker(m_host,buf,msgsize+2);
        if (c->Create() != wxTHREAD_NO_ERROR) {
            wxLogError(wxT("Cannot create more threads"));
        } else {
            c->Run();
            m_threadWorkers.Append(c);
        }
    } else {
        EventWorker* e = new EventWorker(m_host,buf,msgsize+2);
        e->Run();
        m_eventWorkers.Append(e);
    }
    m_statConnecting++;
}

void
Client::StartWorker(workMode pMode, const wxString& pMessage) {
    char* tmpbuf = strdup(pMessage.mb_str());
    int msgsize = strlen(tmpbuf);
    char* buf = CreateBuffer(&msgsize);
    memset(buf+2,0x0,msgsize);
    memcpy(buf+2,tmpbuf,msgsize);
    free(tmpbuf);

    if (pMode == THREADS) {
        ThreadWorker* c = new ThreadWorker(m_host,buf,msgsize+2);
        if (c->Create() != wxTHREAD_NO_ERROR) {
            wxLogError(wxT("Cannot create more threads"));
        } else {
            c->Run();
            m_threadWorkers.Append(c);
        }
    } else {
        EventWorker* e = new EventWorker(m_host,buf,msgsize+2);
        e->Run();
        m_eventWorkers.Append(e);
    }
    m_statConnecting++;
}

void
Client::OnWorkerEvent(WorkerEvent& pEvent) {
    switch (pEvent.m_eventType) {
        case WorkerEvent::CONNECTING:
            if (pEvent.isFailed())
            {
                m_statConnecting--;
                m_statFailed++;
            }
        break;
        case WorkerEvent::SENDING:
            if (pEvent.isFailed())
            {
                m_statFailed++;
                m_statSending--;
            }
            else
            {
                m_statConnecting--;
                m_statSending++;
            }
        break;
        case WorkerEvent::RECEIVING:
            if (pEvent.isFailed())
            {
                m_statReceiving--;
                m_statFailed++;
            }
            else
            {
                m_statSending--;
                m_statReceiving++;
            }
        break;
        case WorkerEvent::DISCONNECTING:
            if (pEvent.isFailed())
            {
                m_statDisconnecting--;
                m_statFailed++;
            }
            else
            {
                m_statReceiving--;
                m_statDisconnecting++;
            }
        break;
        case WorkerEvent::DONE:
            m_statDone++;
            m_statDisconnecting--;
        break;
    };

    if (pEvent.isFailed() || pEvent.m_eventType == WorkerEvent::DONE)
    {
        for(TList::compatibility_iterator it = m_threadWorkers.GetFirst(); it ; it = it->GetNext()) {
            if (it->GetData() == pEvent.m_sender) {
                m_threadWorkers.DeleteNode(it);
                break;
            }
        }
        for(EList::compatibility_iterator it = m_eventWorkers.GetFirst(); it ; it = it->GetNext())
        {
            if (it->GetData() == pEvent.m_sender) {
                delete it->GetData();
                m_eventWorkers.DeleteNode(it);
                break;
            }
        }
        if ((m_threadWorkers.GetCount() == 0) && (m_eventWorkers.GetCount() == 0))
        {
            mTimer.Stop();
            dumpStatistics();
            wxSleep(2);
            ExitMainLoop();
        }
        else
        {
            mTimer.Start(timeout_val,true);
        }
    }
}

void
Client::RemoveEventWorker(EventWorker* p_worker) {
    for(EList::compatibility_iterator it = m_eventWorkers.GetFirst(); it ; it = it->GetNext()) {
        if (it->GetData() == p_worker) {
            //wxLogDebug(wxT("Deleting event worker"));
            delete it->GetData();
            m_eventWorkers.DeleteNode(it);
            return;
        }
    }
}

void
Client::dumpStatistics() {
    wxString msg(
        wxString::Format(_("Connecting:\t%d\nSending\t\t%d\nReceiving\t%d\nDisconnecting:\t%d\nDone:\t\t%d\nFailed:\t\t%d\n"),
                m_statConnecting,
                m_statSending,
                m_statReceiving,
                m_statDisconnecting,
                m_statDone,
                m_statFailed
                ));

    wxLogMessage(wxT("Current status:\n%s\n"),msg.c_str());
}

void
Client::OnTimerEvent(wxTimerEvent&) {
    dumpStatistics();
}

BEGIN_EVENT_TABLE(Client,wxEvtHandler)
    EVT_WORKER(Client::OnWorkerEvent)
    EVT_TIMER(wxID_ANY,Client::OnTimerEvent)
END_EVENT_TABLE()


EventWorker::EventWorker(const wxString& p_host, char* p_buf, int p_size)
  : m_host(p_host),
    m_outbuf(p_buf),
    m_outsize(p_size),
    m_written(0),
    m_readed(0)
{
    m_clientSocket = new wxSocketClient(wxSOCKET_NOWAIT);
    m_clientSocket->SetEventHandler(*this);
    m_insize = m_outsize - 2;
    m_inbuf = new char[m_insize];
}

void
EventWorker::Run() {
    wxIPV4address ca;
    ca.Hostname(m_host);
    ca.Service(3000);
    m_clientSocket->SetNotify(wxSOCKET_CONNECTION_FLAG|wxSOCKET_LOST_FLAG|wxSOCKET_OUTPUT_FLAG|wxSOCKET_INPUT_FLAG);
    m_clientSocket->Notify(true);
    m_currentType = WorkerEvent::CONNECTING;
    m_doneSent = false;
    //wxLogMessage(wxT("EventWorker: Connecting....."));
    m_clientSocket->Connect(ca,false);
}

void
EventWorker::OnSocketEvent(wxSocketEvent& pEvent) {
    switch(pEvent.GetSocketEvent()) {
        case wxSOCKET_INPUT:
            //wxLogDebug(wxT("EventWorker: INPUT"));
            do {
                if (m_readed == m_insize)
                    return; //event already posted
                m_clientSocket->Read(m_inbuf + m_readed, m_insize - m_readed);
                if (m_clientSocket->Error())
                {
                    if (m_clientSocket->LastError() != wxSOCKET_WOULDBLOCK)
                    {
                        wxLogError(wxT("%s: read error"),CreateIdent(m_localaddr).c_str());
                        SendEvent(true);
                    }
                }

                m_readed += m_clientSocket->LastCount();
                //wxLogDebug(wxT("EventWorker: readed %d bytes, %d bytes to do"),m_clientSocket->LastCount(), m_insize - m_readed);
                if (m_readed == m_insize)
                {
                    if (!memcmp(m_inbuf,m_outbuf,m_insize)) {
                        wxLogError(wxT("%s: data mismatch"),CreateIdent(m_localaddr).c_str());
                        SendEvent(true);
                    }
                    m_currentType = WorkerEvent::DISCONNECTING;
                    wxLogDebug(wxT("%s: DISCONNECTING"),CreateIdent(m_localaddr).c_str());
                    SendEvent(false);

                    //wxLogDebug(wxT("EventWorker %p closing"),this);
                    m_clientSocket->Close();

                    m_currentType = WorkerEvent::DONE;
                    wxLogDebug(wxT("%s: DONE"),CreateIdent(m_localaddr).c_str());
                    SendEvent(false);
                }
            } while (!m_clientSocket->Error());
        break;
        case wxSOCKET_OUTPUT:
            //wxLogDebug(wxT("EventWorker: OUTPUT"));
            do {
                if (m_written == m_outsize)
                    return;
                if (m_written == 0)
                {
                    m_currentType = WorkerEvent::SENDING;
                    wxLogDebug(wxT("%s: SENDING"),CreateIdent(m_localaddr).c_str());
                }
                m_clientSocket->Write(m_outbuf + m_written, m_outsize - m_written);
                if (m_clientSocket->Error())
                {
                    if (m_clientSocket->LastError() != wxSOCKET_WOULDBLOCK) {
                        wxLogError(wxT("%s: Write error"),CreateIdent(m_localaddr).c_str());
                        SendEvent(true);
                    }
                }
                m_written += m_clientSocket->LastCount();
                if (m_written != m_outsize)
                {
                    //wxLogDebug(wxT("EventWorker: written %d bytes, %d bytes to do"),m_clientSocket->LastCount(),m_outsize - m_written);
                }
                else
                {
                    //wxLogDebug(wxT("EventWorker %p SENDING->RECEIVING"),this);
                    m_currentType = WorkerEvent::RECEIVING;
                    wxLogDebug(wxT("%s: RECEIVING"),CreateIdent(m_localaddr).c_str());
                    SendEvent(false);
                }
            } while(!m_clientSocket->Error());
        break;
        case wxSOCKET_CONNECTION:
        {
            //wxLogMessage(wxT("EventWorker: got connection"));
            wxLogMessage(wxT("%s: starting writing message (2 bytes for signature and %d bytes of data to write)"),CreateIdent(m_localaddr).c_str(),m_outsize-2);
            if (!m_clientSocket->GetLocal(m_localaddr))
                wxLogError(_("Cannot get peer data for socket %p"),m_clientSocket);
            m_currentType = WorkerEvent::SENDING;
            wxLogDebug(wxT("%s: CONNECTING"),CreateIdent(m_localaddr).c_str());
            SendEvent(false);
        }
        break;
        case wxSOCKET_LOST:
        {
            wxLogError(_("%s: connection lost"),CreateIdent(m_localaddr).c_str());
            SendEvent(true);
        }
        break;
    }
}

void
EventWorker::SendEvent(bool failed) {
    if (m_doneSent)
        return;
    WorkerEvent e(this,m_currentType);
    if (failed) e.setFailed();
    wxGetApp().AddPendingEvent(e);
    m_doneSent = failed || m_currentType == WorkerEvent::DONE;
};

EventWorker::~EventWorker() {
    m_clientSocket->Destroy();
    delete [] m_outbuf;
    delete [] m_inbuf;
}

BEGIN_EVENT_TABLE(EventWorker,wxEvtHandler)
    EVT_SOCKET(wxID_ANY,EventWorker::OnSocketEvent)
END_EVENT_TABLE()


ThreadWorker::ThreadWorker(const wxString& p_host, char* p_buf, int p_size)
  : wxThread(wxTHREAD_DETACHED),
    m_host(p_host),
    m_outbuf(p_buf),
    m_outsize(p_size)
{
    m_clientSocket = new wxSocketClient(wxSOCKET_BLOCK|wxSOCKET_WAITALL);
    m_insize = m_outsize - 2;
    m_inbuf = new char[m_insize];
}

wxThread::ExitCode ThreadWorker::Entry()
{
    wxIPV4address ca;
    ca.Hostname(m_host);
    ca.Service(5678);
    //wxLogDebug(wxT("ThreadWorker: Connecting....."));
    m_clientSocket->SetTimeout(60);
    bool failed = false;
    WorkerEvent::evt_type etype = WorkerEvent::CONNECTING;
    if (!m_clientSocket->Connect(ca)) {
        wxLogError(wxT("Cannot connect to %s:%d"),ca.IPAddress().c_str(), ca.Service());
        failed = true;
    } else {
        //wxLogMessage(wxT("ThreadWorker: Connected. Sending %d bytes of data"),m_outsize);
        etype = WorkerEvent::SENDING;
        WorkerEvent e(this,etype);
        wxGetApp().AddPendingEvent(e);
        int to_process = m_outsize;
        do {
            m_clientSocket->Write(m_outbuf,m_outsize);
            if (m_clientSocket->Error()) {
                wxLogError(wxT("ThreadWorker: Write error"));
                failed  = true;
            }
            to_process -= m_clientSocket->LastCount();
            //wxLogDebug(wxT("EventWorker: written %d bytes, %d bytes to do"),m_clientSocket->LastCount(),to_process);
        } while(!m_clientSocket->Error() && to_process != 0);

        if (!failed) {
            etype = WorkerEvent::RECEIVING;
            WorkerEvent e(this,etype);
            wxGetApp().AddPendingEvent(e);
            to_process = m_insize;
            do {
                m_clientSocket->Read(m_inbuf,m_insize);
                if (m_clientSocket->Error()) {
                    wxLogError(wxT("ThreadWorker: Read error"));
                    failed = true;
                    break;
                }
                to_process -= m_clientSocket->LastCount();
                //wxLogDebug(wxT("EventWorker: readed %d bytes, %d bytes to do"),m_clientSocket->LastCount(),to_process);
            } while(!m_clientSocket->Error() && to_process != 0);
        }

        char* outdat = (char*)m_outbuf+2;
        if (!failed && (memcmp(m_inbuf,outdat,m_insize) != 0))
        {
            wxLogError(wxT("Data mismatch"));
            failed = true;
        }
    }
    //wxLogDebug(wxT("ThreadWorker: Finished"));
    if (!failed) {
        etype = WorkerEvent::DISCONNECTING;
        WorkerEvent e(this,etype);
        wxGetApp().AddPendingEvent(e);
    };
    m_clientSocket->Close();
    m_clientSocket->Destroy();
    m_clientSocket = NULL;
    delete [] m_outbuf;
    delete [] m_inbuf;
    if (!failed)
        etype = WorkerEvent::DONE;
    WorkerEvent e(this,etype);
    if (failed) e.setFailed();
    wxGetApp().AddPendingEvent(e);
    return 0;
}
Commit	Line	Data
2804f77d VZ	1	/////////////////////////////////////////////////////////////////////////////
	2	// Name: samples/sockbase/client.cpp
	3	// Purpose: Sockets sample for wxBase
	4	// Author: Lukasz Michalski
	5	// Modified by:
	6	// Created: 27.06.2005
	7	// RCS-ID: $Id$
	8	// Copyright: (c) 2005 Lukasz Michalski <lmichalski@sf.net>
	9	// Licence: wxWindows license
	10	/////////////////////////////////////////////////////////////////////////////
	11
	12	// ============================================================================
	13	// declarations
	14	// ============================================================================
	15
	16	// ----------------------------------------------------------------------------
	17	// headers
	18	// ----------------------------------------------------------------------------
	19
	20	#include "wx/wx.h"
	21	#include "wx/socket.h"
	22	#include "wx/event.h"
	23	#include "wx/list.h"
	24	#include "wx/cmdline.h"
	25	#include "wx/ffile.h"
	26	#include "wx/datetime.h"
	27	#include "wx/timer.h"
	28	#include "wx/thread.h"
	29
	30	const wxEventType wxEVT_WORKER = wxNewEventType();
	31	#define EVT_WORKER(func) DECLARE_EVENT_TABLE_ENTRY( wxEVT_WORKER, -1, -1, (wxObjectEventFunction) (wxEventFunction) (WorkerEventFunction) & func, (wxObject *) NULL ),
	32
	33	const int timeout_val = 1000;
	34
	35	class WorkerEvent : public wxEvent {
	36	public:
	37	typedef enum {
	38	CONNECTING,
	39	SENDING,
	40	RECEIVING,
	41	DISCONNECTING,
	42	DONE
	43	} evt_type;
	44	WorkerEvent(void* pSender, evt_type type)
	45	{
	46	SetId(-1);
	47	SetEventType(wxEVT_WORKER);
	48	m_sender = pSender;
	49	m_eventType = type;
	50	m_isFailed = false;
	51	}
	52
	53	void setFailed() { m_isFailed = true; }
	54	bool isFailed() const { return m_isFailed; }
	55
	56	virtual wxEvent* Clone() const
	57	{
	58	return new WorkerEvent(*this);
	59	}
	60	void* m_sender;
	61	bool m_isFailed;
	62	wxString m_workerIdent;
	63	evt_type m_eventType;
	64	};
65
66	typedef void (wxEvtHandler::*WorkerEventFunction)(WorkerEvent&);
67
68	class ThreadWorker;
69	class EventWorker;
70
71	WX_DECLARE_LIST(ThreadWorker, TList);
72	WX_DECLARE_LIST(EventWorker, EList);
73
74	class Client : public wxApp {
75	DECLARE_EVENT_TABLE();
76	public:
77	void RemoveEventWorker(EventWorker* p_worker);
78	private:
79	typedef enum
80	{
81	THREADS,
82	EVENTS
83	} workMode;
84
85	typedef enum
86	{
87	SEND_RANDOM,
88	SEND_MESSAGE,
89	STRESS_TEST
90	} sendType;
91
92	workMode m_workMode;
93	sendType m_sendType;
94	wxString m_message;
95	wxString m_host;
96	long m_stressWorkers;
97
98	virtual bool OnInit();
99	virtual int OnRun();
100	virtual int OnExit();
101	void OnInitCmdLine(wxCmdLineParser& pParser);
102	bool OnCmdLineParsed(wxCmdLineParser& pParser);
103	void OnWorkerEvent(WorkerEvent& pEvent);
104	void OnTimerEvent(wxTimerEvent& pEvent);
105
106	void StartWorker(workMode pMode, const wxString& pMessage);
107	void StartWorker(workMode pMode);
108	char* CreateBuffer(int *msgsize);
109
110	void dumpStatistics();
111
112	TList m_threadWorkers;
113	EList m_eventWorkers;
114
115	unsigned m_statConnecting;
116	unsigned m_statSending;
117	unsigned m_statReceiving;
118	unsigned m_statDisconnecting;
119	unsigned m_statDone;
120	unsigned m_statFailed;
121
122	wxTimer mTimer;
123	};
124
125	DECLARE_APP(Client);
126
127	class ThreadWorker : public wxThread
128	{
129	public:
130	ThreadWorker(const wxString& p_host, char* p_buf, int p_size);
131	virtual ExitCode Entry();
132	private:
133	wxString m_host;
134	wxSocketClient* m_clientSocket;
135	char* m_inbuf;
136	char* m_outbuf;
137	int m_outsize;
138	int m_insize;
139	wxString m_workerIdent;
140	};
141
142	class EventWorker : public wxEvtHandler
143	{
144	DECLARE_EVENT_TABLE();
145	public:
146	EventWorker(const wxString& p_host, char* p_buf, int p_size);
147	void Run();
148	virtual ~EventWorker();
149	private:
150	wxString m_host;
151	wxSocketClient* m_clientSocket;
152	char* m_inbuf;
153	char* m_outbuf;
154	int m_outsize;
155	int m_written;
156	int m_insize;
157	int m_readed;
158
159	WorkerEvent::evt_type m_currentType;
160	bool m_doneSent;
161	wxIPV4address m_localaddr;
162
163	void OnSocketEvent(wxSocketEvent& pEvent);
164	void SendEvent(bool failed);
165	};
166
167	/***************** Implementation ****************/
168	IMPLEMENT_APP_CONSOLE(Client);
169
170	#include <wx/listimpl.cpp>
171	WX_DEFINE_LIST(TList);
172	WX_DEFINE_LIST(EList);
173
174	wxString
175	CreateIdent(const wxIPV4address& addr)
176	{
177	return wxString::Format(wxT("%s:%d"),addr.IPAddress().c_str(),addr.Service());
178	}
179
180	void
181	Client::OnInitCmdLine(wxCmdLineParser& pParser)
182	{
183	wxApp::OnInitCmdLine(pParser);
184	pParser.AddSwitch(wxT("e"),wxT("event"),_("Use event based worker (default)"),wxCMD_LINE_PARAM_OPTIONAL);
185	pParser.AddSwitch(wxT("t"),wxT("thread"),_("Use thread based worker"),wxCMD_LINE_PARAM_OPTIONAL);
186	pParser.AddSwitch(wxT("r"),wxT("random"),_("Send radnom data (default)"),wxCMD_LINE_PARAM_OPTIONAL);
187	pParser.AddOption(wxT("m"),wxT("message"),_("Send message from <str>"),wxCMD_LINE_VAL_STRING,wxCMD_LINE_PARAM_OPTIONAL);
188	pParser.AddOption(wxT("f"),wxT("file"),_("Send contents of <file>"),wxCMD_LINE_VAL_STRING,wxCMD_LINE_PARAM_OPTIONAL);
189	pParser.AddOption(wxT("H"),wxT("hostname"),_("IP or name of host to connect to"),wxCMD_LINE_VAL_STRING,wxCMD_LINE_PARAM_OPTIONAL);
190	pParser.AddOption(wxT("s"),wxT("stress"),_("stress test with <num> concurrent connections"),wxCMD_LINE_VAL_NUMBER,wxCMD_LINE_PARAM_OPTIONAL);
191	}
192
193
194	bool
195	Client::OnCmdLineParsed(wxCmdLineParser& pParser)
196	{
197	wxString fname;
198	m_workMode = EVENTS;
199	m_stressWorkers = 50;
200
201	if (pParser.Found(_("verbose")))
202	{
203	wxLog::AddTraceMask(wxT("wxSocket"));
204	wxLog::AddTraceMask(wxT("epolldispatcher"));
205	wxLog::AddTraceMask(wxT("selectdispatcher"));
206	wxLog::AddTraceMask(wxT("thread"));
207	wxLog::AddTraceMask(wxT("events"));
208	}
209
210	if (pParser.Found(wxT("t")))
211	m_workMode = THREADS;
212	m_sendType = SEND_RANDOM;
213
214	if (pParser.Found(wxT("m"),&m_message))
215	m_sendType = SEND_MESSAGE;
216	else if (pParser.Found(wxT("f"),&fname))
217	{
218	wxFFile file(fname);
219	if (!file.IsOpened()) {
220	wxLogError(wxT("Cannot open file %s"),fname.c_str());
221	return false;
222	};
223	if (!file.ReadAll(&m_message)) {
224	wxLogError(wxT("Cannot read conten of file %s"),fname.c_str());
225	return false;
226	};
227	m_sendType = SEND_MESSAGE;
228	};
229
230	if (pParser.Found(wxT("s"),&m_stressWorkers))
231	m_sendType = STRESS_TEST;
232
233	m_host = wxT("127.0.0.1");
234	pParser.Found(wxT("H"),&m_host);
235	return wxApp::OnCmdLineParsed(pParser);
236	};
237
238	bool
239	Client::OnInit()
240	{
241	if (!wxApp::OnInit())
242	return false;
243	srand(wxDateTime::Now().GetTicks());
244	mTimer.SetOwner(this);
245	m_statConnecting = 0;
246	m_statSending = 0;
247	m_statReceiving = 0;
248	m_statDisconnecting = 0;
249	m_statDone = 0;
250	m_statFailed = 0;
251	return true;
252	}
253
254	int
255	Client::OnRun()
256	{
257	switch(m_sendType)
258	{
259	case STRESS_TEST:
260	switch(m_workMode)
261	{
262	case THREADS:
263	for (int i = 0; i < m_stressWorkers; i++) {
264	if (m_message.empty())
265	StartWorker(THREADS);
266	else
267	StartWorker(THREADS, m_message);
268	}
269	break;
270	case EVENTS:
271	for (int i = 0; i < m_stressWorkers; i++) {
272	if (m_message.empty())
273	StartWorker(EVENTS);
274	else
275	StartWorker(EVENTS, m_message);
276	}
277	break;
278	default:
279	for (int i = 0; i < m_stressWorkers; i++) {
280	if (m_message.empty())
281	StartWorker(i % 5 == 0 ? THREADS : EVENTS);
282	else
283	StartWorker(i % 5 == 0 ? THREADS : EVENTS, m_message);
284	}
285	break;
286	}
287	break;
288	case SEND_MESSAGE:
289	StartWorker(m_workMode,m_message);
290	break;
291	case SEND_RANDOM:
292	StartWorker(m_workMode);
293	break;
294	}
295	mTimer.Start(timeout_val,true);
296	return wxApp::OnRun();
297	}
298
299	int
300	Client::OnExit()
301	{
302	for(EList::compatibility_iterator it = m_eventWorkers.GetFirst(); it ; it->GetNext()) {
303	delete it->GetData();
304	}
305	return 0;
306	}
307
308	// Create buffer to be sent by client. Buffer contains test indicator
309	// message size and place for data
310	// msgsize parameter contains size of data in bytes and
311	// if input value does not fit into 250 bytes then
312	// on exit is updated to new value that is multiply of 1024 bytes
313	char*
314	Client::CreateBuffer(int* msgsize)
315	{
316	int bufsize = 0;
317	char* buf;
318	//if message should have more than 256 bytes then set it as
319	//test3 for compatibility with GUI server sample
320	if ((*msgsize) > 250)
321	{
322	//send at least one kb of data
323	int size = (*msgsize)/1024 + 1;
324	//returned buffer will contain test indicator, message size in kb and data
325	bufsize = size*1024+2;
326	buf = new char[bufsize];
327	buf[0] = 0xDE; //second byte contains size in kilobytes
328	buf[1] = (char)(size);
329	msgsize = size1024;
330	}
331	else
332	{
333	//returned buffer will contain test indicator, message size in kb and data
334	bufsize = (*msgsize)+2;
335	buf = new char[bufsize];
336	buf[0] = 0xBE; //second byte contains size in bytes
337	buf[1] = (char)(*msgsize);
338	}
339	return buf;
340	}
341
342	void
343	Client::StartWorker(workMode pMode) {
344	int msgsize = 1 + (int) (250000.0 * (rand() / (RAND_MAX + 1.0)));
345	char* buf = CreateBuffer(&msgsize);
346
347	//fill data part of buffer with random bytes
348	for (int i = 2; i < (msgsize); i++) {
349	buf[i] = i % 256;
350	}
351
352	if (pMode == THREADS) {
353	ThreadWorker* c = new ThreadWorker(m_host,buf,msgsize+2);
354	if (c->Create() != wxTHREAD_NO_ERROR) {
355	wxLogError(wxT("Cannot create more threads"));
356	} else {
357	c->Run();
358	m_threadWorkers.Append(c);
359	}
360	} else {
361	EventWorker* e = new EventWorker(m_host,buf,msgsize+2);
362	e->Run();
363	m_eventWorkers.Append(e);
364	}
365	m_statConnecting++;
366	}
367
368	void
369	Client::StartWorker(workMode pMode, const wxString& pMessage) {
370	char* tmpbuf = strdup(pMessage.mb_str());
371	int msgsize = strlen(tmpbuf);
372	char* buf = CreateBuffer(&msgsize);
373	memset(buf+2,0x0,msgsize);
374	memcpy(buf+2,tmpbuf,msgsize);
375	free(tmpbuf);
376
377	if (pMode == THREADS) {
378	ThreadWorker* c = new ThreadWorker(m_host,buf,msgsize+2);
379	if (c->Create() != wxTHREAD_NO_ERROR) {
380	wxLogError(wxT("Cannot create more threads"));
381	} else {
382	c->Run();
383	m_threadWorkers.Append(c);
384	}
385	} else {
386	EventWorker* e = new EventWorker(m_host,buf,msgsize+2);
387	e->Run();
388	m_eventWorkers.Append(e);
389	}
390	m_statConnecting++;
391	}
392
393	void
394	Client::OnWorkerEvent(WorkerEvent& pEvent) {
395	switch (pEvent.m_eventType) {
396	case WorkerEvent::CONNECTING:
397	if (pEvent.isFailed())
398	{
399	m_statConnecting--;
400	m_statFailed++;
401	}
402	break;
403	case WorkerEvent::SENDING:
404	if (pEvent.isFailed())
405	{
406	m_statFailed++;
407	m_statSending--;
408	}
409	else
410	{
411	m_statConnecting--;
412	m_statSending++;
413	}
414	break;
415	case WorkerEvent::RECEIVING:
416	if (pEvent.isFailed())
417	{
418	m_statReceiving--;
419	m_statFailed++;
420	}
421	else
422	{
423	m_statSending--;
424	m_statReceiving++;
425	}
426	break;
427	case WorkerEvent::DISCONNECTING:
428	if (pEvent.isFailed())
429	{
430	m_statDisconnecting--;
431	m_statFailed++;
432	}
433	else
434	{
435	m_statReceiving--;
436	m_statDisconnecting++;
437	}
438	break;
439	case WorkerEvent::DONE:
440	m_statDone++;
441	m_statDisconnecting--;
442	break;
443	};
444
445	if (pEvent.isFailed() \|\| pEvent.m_eventType == WorkerEvent::DONE)
446	{
447	for(TList::compatibility_iterator it = m_threadWorkers.GetFirst(); it ; it = it->GetNext()) {
448	if (it->GetData() == pEvent.m_sender) {
449	m_threadWorkers.DeleteNode(it);
450	break;
451	}
452	}
453	for(EList::compatibility_iterator it = m_eventWorkers.GetFirst(); it ; it = it->GetNext())
454	{
455	if (it->GetData() == pEvent.m_sender) {
456	delete it->GetData();
457	m_eventWorkers.DeleteNode(it);
458	break;
459	}
460	}
461	if ((m_threadWorkers.GetCount() == 0) && (m_eventWorkers.GetCount() == 0))
462	{
463	mTimer.Stop();
464	dumpStatistics();
465	wxSleep(2);
466	ExitMainLoop();
467	}
468	else
469	{
470	mTimer.Start(timeout_val,true);
471	}
472	}
473	}
474
475	void
476	Client::RemoveEventWorker(EventWorker* p_worker) {
477	for(EList::compatibility_iterator it = m_eventWorkers.GetFirst(); it ; it = it->GetNext()) {
478	if (it->GetData() == p_worker) {
479	//wxLogDebug(wxT("Deleting event worker"));
480	delete it->GetData();
481	m_eventWorkers.DeleteNode(it);
482	return;
483	}
484	}
485	}
486
487	void
488	Client::dumpStatistics() {
489	wxString msg(
490	wxString::Format(_("Connecting:\t%d\nSending\t\t%d\nReceiving\t%d\nDisconnecting:\t%d\nDone:\t\t%d\nFailed:\t\t%d\n"),
491	m_statConnecting,
492	m_statSending,
493	m_statReceiving,
494	m_statDisconnecting,
495	m_statDone,
496	m_statFailed
497	));
498
499	wxLogMessage(wxT("Current status:\n%s\n"),msg.c_str());
500	}
501
502	void
503	Client::OnTimerEvent(wxTimerEvent&) {
504	dumpStatistics();
505	}
506
507	BEGIN_EVENT_TABLE(Client,wxEvtHandler)
508	EVT_WORKER(Client::OnWorkerEvent)
509	EVT_TIMER(wxID_ANY,Client::OnTimerEvent)
510	END_EVENT_TABLE()
511
512
513
514	EventWorker::EventWorker(const wxString& p_host, char* p_buf, int p_size)
515	: m_host(p_host),
516	m_outbuf(p_buf),
517	m_outsize(p_size),
518	m_written(0),
519	m_readed(0)
520	{
521	m_clientSocket = new wxSocketClient(wxSOCKET_NOWAIT);
522	m_clientSocket->SetEventHandler(*this);
523	m_insize = m_outsize - 2;
524	m_inbuf = new char[m_insize];
525	}
526
527	void
528	EventWorker::Run() {
529	wxIPV4address ca;
530	ca.Hostname(m_host);
531	ca.Service(3000);
532	m_clientSocket->SetNotify(wxSOCKET_CONNECTION_FLAG\|wxSOCKET_LOST_FLAG\|wxSOCKET_OUTPUT_FLAG\|wxSOCKET_INPUT_FLAG);
533	m_clientSocket->Notify(true);
534	m_currentType = WorkerEvent::CONNECTING;
535	m_doneSent = false;
536	//wxLogMessage(wxT("EventWorker: Connecting....."));
537	m_clientSocket->Connect(ca,false);
538	}
539
540	void
541	EventWorker::OnSocketEvent(wxSocketEvent& pEvent) {
542	switch(pEvent.GetSocketEvent()) {
543	case wxSOCKET_INPUT:
544	//wxLogDebug(wxT("EventWorker: INPUT"));
545	do {
546	if (m_readed == m_insize)
547	return; //event already posted
548	m_clientSocket->Read(m_inbuf + m_readed, m_insize - m_readed);
549	if (m_clientSocket->Error())
550	{
551	if (m_clientSocket->LastError() != wxSOCKET_WOULDBLOCK)
552	{
553	wxLogError(wxT("%s: read error"),CreateIdent(m_localaddr).c_str());
554	SendEvent(true);
555	}
556	}
557
558	m_readed += m_clientSocket->LastCount();
559	//wxLogDebug(wxT("EventWorker: readed %d bytes, %d bytes to do"),m_clientSocket->LastCount(), m_insize - m_readed);
560	if (m_readed == m_insize)
561	{
562	if (!memcmp(m_inbuf,m_outbuf,m_insize)) {
563	wxLogError(wxT("%s: data mismatch"),CreateIdent(m_localaddr).c_str());
564	SendEvent(true);
565	}
566	m_currentType = WorkerEvent::DISCONNECTING;
567	wxLogDebug(wxT("%s: DISCONNECTING"),CreateIdent(m_localaddr).c_str());
568	SendEvent(false);
569
570	//wxLogDebug(wxT("EventWorker %p closing"),this);
571	m_clientSocket->Close();
572
573	m_currentType = WorkerEvent::DONE;
574	wxLogDebug(wxT("%s: DONE"),CreateIdent(m_localaddr).c_str());
575	SendEvent(false);
576	}
577	} while (!m_clientSocket->Error());
578	break;
579	case wxSOCKET_OUTPUT:
580	//wxLogDebug(wxT("EventWorker: OUTPUT"));
581	do {
582	if (m_written == m_outsize)
583	return;
584	if (m_written == 0)
585	{
586	m_currentType = WorkerEvent::SENDING;
587	wxLogDebug(wxT("%s: SENDING"),CreateIdent(m_localaddr).c_str());
588	}
589	m_clientSocket->Write(m_outbuf + m_written, m_outsize - m_written);
590	if (m_clientSocket->Error())
591	{
592	if (m_clientSocket->LastError() != wxSOCKET_WOULDBLOCK) {
593	wxLogError(wxT("%s: Write error"),CreateIdent(m_localaddr).c_str());
594	SendEvent(true);
595	}
596	}
597	m_written += m_clientSocket->LastCount();
598	if (m_written != m_outsize)
599	{
600	//wxLogDebug(wxT("EventWorker: written %d bytes, %d bytes to do"),m_clientSocket->LastCount(),m_outsize - m_written);
601	}
602	else
603	{
604	//wxLogDebug(wxT("EventWorker %p SENDING->RECEIVING"),this);
605	m_currentType = WorkerEvent::RECEIVING;
606	wxLogDebug(wxT("%s: RECEIVING"),CreateIdent(m_localaddr).c_str());
607	SendEvent(false);
608	}
609	} while(!m_clientSocket->Error());
610	break;
611	case wxSOCKET_CONNECTION:
612	{
613	//wxLogMessage(wxT("EventWorker: got connection"));
614	wxLogMessage(wxT("%s: starting writing message (2 bytes for signature and %d bytes of data to write)"),CreateIdent(m_localaddr).c_str(),m_outsize-2);
615	if (!m_clientSocket->GetLocal(m_localaddr))
616	wxLogError(_("Cannot get peer data for socket %p"),m_clientSocket);
617	m_currentType = WorkerEvent::SENDING;
618	wxLogDebug(wxT("%s: CONNECTING"),CreateIdent(m_localaddr).c_str());
619	SendEvent(false);
620	}
621	break;
622	case wxSOCKET_LOST:
623	{
624	wxLogError(_("%s: connection lost"),CreateIdent(m_localaddr).c_str());
625	SendEvent(true);
626	}
627	break;
628	}
629	}
630
631	void
632	EventWorker::SendEvent(bool failed) {
633	if (m_doneSent)
634	return;
635	WorkerEvent e(this,m_currentType);
636	if (failed) e.setFailed();
637	wxGetApp().AddPendingEvent(e);
638	m_doneSent = failed \|\| m_currentType == WorkerEvent::DONE;
639	};
640
641	EventWorker::~EventWorker() {
642	m_clientSocket->Destroy();
643	delete [] m_outbuf;
644	delete [] m_inbuf;
645	}
646
647	BEGIN_EVENT_TABLE(EventWorker,wxEvtHandler)
648	EVT_SOCKET(wxID_ANY,EventWorker::OnSocketEvent)
649	END_EVENT_TABLE()
650
651
652	ThreadWorker::ThreadWorker(const wxString& p_host, char* p_buf, int p_size)
653	: wxThread(wxTHREAD_DETACHED),
654	m_host(p_host),
655	m_outbuf(p_buf),
656	m_outsize(p_size)
657	{
658	m_clientSocket = new wxSocketClient(wxSOCKET_BLOCK\|wxSOCKET_WAITALL);
659	m_insize = m_outsize - 2;
660	m_inbuf = new char[m_insize];
661	}
662
663	wxThread::ExitCode ThreadWorker::Entry()
664	{
665	wxIPV4address ca;
666	ca.Hostname(m_host);
667	ca.Service(5678);
668	//wxLogDebug(wxT("ThreadWorker: Connecting....."));
669	m_clientSocket->SetTimeout(60);
670	bool failed = false;
671	WorkerEvent::evt_type etype = WorkerEvent::CONNECTING;
672	if (!m_clientSocket->Connect(ca)) {
673	wxLogError(wxT("Cannot connect to %s:%d"),ca.IPAddress().c_str(), ca.Service());
674	failed = true;
675	} else {
676	//wxLogMessage(wxT("ThreadWorker: Connected. Sending %d bytes of data"),m_outsize);
677	etype = WorkerEvent::SENDING;
678	WorkerEvent e(this,etype);
679	wxGetApp().AddPendingEvent(e);
680	int to_process = m_outsize;
681	do {
682	m_clientSocket->Write(m_outbuf,m_outsize);
683	if (m_clientSocket->Error()) {
684	wxLogError(wxT("ThreadWorker: Write error"));
685	failed = true;
686	}
687	to_process -= m_clientSocket->LastCount();
688	//wxLogDebug(wxT("EventWorker: written %d bytes, %d bytes to do"),m_clientSocket->LastCount(),to_process);
689	} while(!m_clientSocket->Error() && to_process != 0);
690
691	if (!failed) {
692	etype = WorkerEvent::RECEIVING;
693	WorkerEvent e(this,etype);
694	wxGetApp().AddPendingEvent(e);
695	to_process = m_insize;
696	do {
697	m_clientSocket->Read(m_inbuf,m_insize);
698	if (m_clientSocket->Error()) {
699	wxLogError(wxT("ThreadWorker: Read error"));
700	failed = true;
701	break;
702	}
703	to_process -= m_clientSocket->LastCount();
704	//wxLogDebug(wxT("EventWorker: readed %d bytes, %d bytes to do"),m_clientSocket->LastCount(),to_process);
705	} while(!m_clientSocket->Error() && to_process != 0);
706	}
707
708	char* outdat = (char*)m_outbuf+2;
709	if (!failed && (memcmp(m_inbuf,outdat,m_insize) != 0))
710	{
711	wxLogError(wxT("Data mismatch"));
712	failed = true;
713	}
714	}
715	//wxLogDebug(wxT("ThreadWorker: Finished"));
716	if (!failed) {
717	etype = WorkerEvent::DISCONNECTING;
718	WorkerEvent e(this,etype);
719	wxGetApp().AddPendingEvent(e);
720	};
721	m_clientSocket->Close();
722	m_clientSocket->Destroy();
723	m_clientSocket = NULL;
724	delete [] m_outbuf;
725	delete [] m_inbuf;
726	if (!failed)
727	etype = WorkerEvent::DONE;
728	WorkerEvent e(this,etype);
729	if (failed) e.setFailed();
730	wxGetApp().AddPendingEvent(e);
731	return 0;
732	}
733