[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
// Copyright:   (c) 2005 Lukasz Michalski <lmichalski@sf.net>
// Licence:     wxWindows licence
/////////////////////////////////////////////////////////////////////////////

// ============================================================================
// 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()
{
    int i;
    switch(m_sendType)
    {
        case STRESS_TEST:
            switch(m_workMode)
            {
                case THREADS:
                    for (i = 0; i < m_stressWorkers; i++) {
                        if (m_message.empty())
                            StartWorker(THREADS);
                        else
                            StartWorker(THREADS, m_message);
                    }
                    break;
                case EVENTS:
                    for (i = 0; i < m_stressWorkers; i++) {
                        if (m_message.empty())
                            StartWorker(EVENTS);
                        else
                            StartWorker(EVENTS, m_message);
                    }
                    break;
                default:
                    for (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] = (unsigned char)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] = (unsigned char)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 = wxStrdup(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 it2 = m_eventWorkers.GetFirst(); it2 ; it2 = it2->GetNext())
        {
            if (it2->GetData() == pEvent.m_sender) {
                delete it2->GetData();
                m_eventWorkers.DeleteNode(it2);
                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
2804f77d	7	// Copyright: (c) 2005 Lukasz Michalski <lmichalski@sf.net>
526954c5	8	// Licence: wxWindows licence
2804f77d VZ	9	/////////////////////////////////////////////////////////////////////////////
	10
	11	// ============================================================================
	12	// declarations
	13	// ============================================================================
	14
	15	// ----------------------------------------------------------------------------
	16	// headers
	17	// ----------------------------------------------------------------------------
	18
	19	#include "wx/wx.h"
	20	#include "wx/socket.h"
	21	#include "wx/event.h"
	22	#include "wx/list.h"
	23	#include "wx/cmdline.h"
	24	#include "wx/ffile.h"
	25	#include "wx/datetime.h"
	26	#include "wx/timer.h"
	27	#include "wx/thread.h"
	28
	29	const wxEventType wxEVT_WORKER = wxNewEventType();
	30	#define EVT_WORKER(func) DECLARE_EVENT_TABLE_ENTRY( wxEVT_WORKER, -1, -1, (wxObjectEventFunction) (wxEventFunction) (WorkerEventFunction) & func, (wxObject *) NULL ),
	31
	32	const int timeout_val = 1000;
	33
	34	class WorkerEvent : public wxEvent {
	35	public:
	36	typedef enum {
	37	CONNECTING,
	38	SENDING,
	39	RECEIVING,
	40	DISCONNECTING,
	41	DONE
	42	} evt_type;
	43	WorkerEvent(void* pSender, evt_type type)
	44	{
	45	SetId(-1);
	46	SetEventType(wxEVT_WORKER);
	47	m_sender = pSender;
	48	m_eventType = type;
	49	m_isFailed = false;
	50	}
	51
	52	void setFailed() { m_isFailed = true; }
	53	bool isFailed() const { return m_isFailed; }
	54
	55	virtual wxEvent* Clone() const
	56	{
	57	return new WorkerEvent(*this);
	58	}
	59	void* m_sender;
	60	bool m_isFailed;
	61	wxString m_workerIdent;
	62	evt_type m_eventType;
	63	};
	64
	65	typedef void (wxEvtHandler::*WorkerEventFunction)(WorkerEvent&);
	66
	67	class ThreadWorker;
	68	class EventWorker;
	69
	70	WX_DECLARE_LIST(ThreadWorker, TList);
	71	WX_DECLARE_LIST(EventWorker, EList);
	72
73	class Client : public wxApp {
66034aba	74	DECLARE_EVENT_TABLE()
2804f77d VZ	75	public:
	76	void RemoveEventWorker(EventWorker* p_worker);
	77	private:
	78	typedef enum
	79	{
	80	THREADS,
	81	EVENTS
	82	} workMode;
	83
	84	typedef enum
	85	{
	86	SEND_RANDOM,
	87	SEND_MESSAGE,
	88	STRESS_TEST
	89	} sendType;
	90
	91	workMode m_workMode;
	92	sendType m_sendType;
	93	wxString m_message;
	94	wxString m_host;
	95	long m_stressWorkers;
	96
	97	virtual bool OnInit();
	98	virtual int OnRun();
	99	virtual int OnExit();
	100	void OnInitCmdLine(wxCmdLineParser& pParser);
	101	bool OnCmdLineParsed(wxCmdLineParser& pParser);
	102	void OnWorkerEvent(WorkerEvent& pEvent);
	103	void OnTimerEvent(wxTimerEvent& pEvent);
	104
	105	void StartWorker(workMode pMode, const wxString& pMessage);
	106	void StartWorker(workMode pMode);
	107	char* CreateBuffer(int *msgsize);
	108
	109	void dumpStatistics();
	110
	111	TList m_threadWorkers;
	112	EList m_eventWorkers;
	113
	114	unsigned m_statConnecting;
	115	unsigned m_statSending;
	116	unsigned m_statReceiving;
	117	unsigned m_statDisconnecting;
	118	unsigned m_statDone;
	119	unsigned m_statFailed;
	120
	121	wxTimer mTimer;
	122	};
	123
	124	DECLARE_APP(Client);
	125
	126	class ThreadWorker : public wxThread
	127	{
	128	public:
	129	ThreadWorker(const wxString& p_host, char* p_buf, int p_size);
	130	virtual ExitCode Entry();
	131	private:
	132	wxString m_host;
	133	wxSocketClient* m_clientSocket;
	134	char* m_inbuf;
	135	char* m_outbuf;
	136	int m_outsize;
	137	int m_insize;
	138	wxString m_workerIdent;
139	};
140
141	class EventWorker : public wxEvtHandler
142	{
66034aba	143	DECLARE_EVENT_TABLE()
2804f77d VZ	144	public:
	145	EventWorker(const wxString& p_host, char* p_buf, int p_size);
	146	void Run();
	147	virtual ~EventWorker();
	148	private:
	149	wxString m_host;
	150	wxSocketClient* m_clientSocket;
	151	char* m_inbuf;
	152	char* m_outbuf;
	153	int m_outsize;
	154	int m_written;
	155	int m_insize;
	156	int m_readed;
	157
	158	WorkerEvent::evt_type m_currentType;
	159	bool m_doneSent;
	160	wxIPV4address m_localaddr;
	161
	162	void OnSocketEvent(wxSocketEvent& pEvent);
	163	void SendEvent(bool failed);
	164	};
	165
	166	/***************** Implementation ****************/
	167	IMPLEMENT_APP_CONSOLE(Client);
	168
	169	#include <wx/listimpl.cpp>
	170	WX_DEFINE_LIST(TList);
	171	WX_DEFINE_LIST(EList);
	172
	173	wxString
	174	CreateIdent(const wxIPV4address& addr)
	175	{
	176	return wxString::Format(wxT("%s:%d"),addr.IPAddress().c_str(),addr.Service());
	177	}
	178
	179	void
	180	Client::OnInitCmdLine(wxCmdLineParser& pParser)
	181	{
	182	wxApp::OnInitCmdLine(pParser);
	183	pParser.AddSwitch(wxT("e"),wxT("event"),_("Use event based worker (default)"),wxCMD_LINE_PARAM_OPTIONAL);
	184	pParser.AddSwitch(wxT("t"),wxT("thread"),_("Use thread based worker"),wxCMD_LINE_PARAM_OPTIONAL);
	185	pParser.AddSwitch(wxT("r"),wxT("random"),_("Send radnom data (default)"),wxCMD_LINE_PARAM_OPTIONAL);
	186	pParser.AddOption(wxT("m"),wxT("message"),_("Send message from <str>"),wxCMD_LINE_VAL_STRING,wxCMD_LINE_PARAM_OPTIONAL);
	187	pParser.AddOption(wxT("f"),wxT("file"),_("Send contents of <file>"),wxCMD_LINE_VAL_STRING,wxCMD_LINE_PARAM_OPTIONAL);
	188	pParser.AddOption(wxT("H"),wxT("hostname"),_("IP or name of host to connect to"),wxCMD_LINE_VAL_STRING,wxCMD_LINE_PARAM_OPTIONAL);
	189	pParser.AddOption(wxT("s"),wxT("stress"),_("stress test with <num> concurrent connections"),wxCMD_LINE_VAL_NUMBER,wxCMD_LINE_PARAM_OPTIONAL);
	190	}
	191
	192
	193	bool
	194	Client::OnCmdLineParsed(wxCmdLineParser& pParser)
	195	{
	196	wxString fname;
	197	m_workMode = EVENTS;
	198	m_stressWorkers = 50;
	199
	200	if (pParser.Found(_("verbose")))
	201	{
	202	wxLog::AddTraceMask(wxT("wxSocket"));
	203	wxLog::AddTraceMask(wxT("epolldispatcher"));
	204	wxLog::AddTraceMask(wxT("selectdispatcher"));
	205	wxLog::AddTraceMask(wxT("thread"));
	206	wxLog::AddTraceMask(wxT("events"));
	207	}
208
209	if (pParser.Found(wxT("t")))
210	m_workMode = THREADS;
211	m_sendType = SEND_RANDOM;
212
213	if (pParser.Found(wxT("m"),&m_message))
214	m_sendType = SEND_MESSAGE;
215	else if (pParser.Found(wxT("f"),&fname))
216	{
217	wxFFile file(fname);
218	if (!file.IsOpened()) {
219	wxLogError(wxT("Cannot open file %s"),fname.c_str());
220	return false;
221	};
222	if (!file.ReadAll(&m_message)) {
223	wxLogError(wxT("Cannot read conten of file %s"),fname.c_str());
224	return false;
225	};
226	m_sendType = SEND_MESSAGE;
227	};
228
229	if (pParser.Found(wxT("s"),&m_stressWorkers))
230	m_sendType = STRESS_TEST;
231
232	m_host = wxT("127.0.0.1");
233	pParser.Found(wxT("H"),&m_host);
234	return wxApp::OnCmdLineParsed(pParser);
235	};
236
237	bool
238	Client::OnInit()
239	{
240	if (!wxApp::OnInit())
241	return false;
242	srand(wxDateTime::Now().GetTicks());
243	mTimer.SetOwner(this);
244	m_statConnecting = 0;
245	m_statSending = 0;
246	m_statReceiving = 0;
247	m_statDisconnecting = 0;
248	m_statDone = 0;
249	m_statFailed = 0;
250	return true;
251	}
252
253	int
254	Client::OnRun()
255	{
45a50a2e	256	int i;
2804f77d VZ	257	switch(m_sendType)
	258	{
	259	case STRESS_TEST:
	260	switch(m_workMode)
	261	{
	262	case THREADS:
45a50a2e	263	for (i = 0; i < m_stressWorkers; i++) {
2804f77d VZ	264	if (m_message.empty())
	265	StartWorker(THREADS);
	266	else
	267	StartWorker(THREADS, m_message);
	268	}
	269	break;
	270	case EVENTS:
45a50a2e	271	for (i = 0; i < m_stressWorkers; i++) {
2804f77d VZ	272	if (m_message.empty())
	273	StartWorker(EVENTS);
	274	else
	275	StartWorker(EVENTS, m_message);
	276	}
	277	break;
	278	default:
45a50a2e	279	for (i = 0; i < m_stressWorkers; i++) {
2804f77d VZ	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
ce00f59b	308	// Create buffer to be sent by client. Buffer contains test indicator
2804f77d	309	// message size and place for data
ce00f59b	310	// msgsize parameter contains size of data in bytes and
2804f77d VZ	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
ce00f59b	320	if ((*msgsize) > 250)
2804f77d VZ	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];
6a8a90fe	327	buf[0] = (unsigned char)0xDE; //second byte contains size in kilobytes
2804f77d VZ	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];
6a8a90fe	336	buf[0] = (unsigned char)0xBE; //second byte contains size in bytes
2804f77d VZ	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) {
3bcd8f97	370	char* tmpbuf = wxStrdup(pMessage.mb_str());
2804f77d VZ	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	}
45a50a2e	453	for(EList::compatibility_iterator it2 = m_eventWorkers.GetFirst(); it2 ; it2 = it2->GetNext())
2804f77d	454	{
45a50a2e VZ	455	if (it2->GetData() == pEvent.m_sender) {
	456	delete it2->GetData();
	457	m_eventWorkers.DeleteNode(it2);
2804f77d VZ	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))
43b2d5e7	616	{
2804f77d	617	wxLogError(_("Cannot get peer data for socket %p"),m_clientSocket);
43b2d5e7	618	}
2804f77d VZ	619	m_currentType = WorkerEvent::SENDING;
	620	wxLogDebug(wxT("%s: CONNECTING"),CreateIdent(m_localaddr).c_str());
	621	SendEvent(false);
	622	}
	623	break;
	624	case wxSOCKET_LOST:
	625	{
	626	wxLogError(_("%s: connection lost"),CreateIdent(m_localaddr).c_str());
	627	SendEvent(true);
	628	}
	629	break;
	630	}
	631	}
	632
	633	void
	634	EventWorker::SendEvent(bool failed) {
	635	if (m_doneSent)
	636	return;
	637	WorkerEvent e(this,m_currentType);
	638	if (failed) e.setFailed();
	639	wxGetApp().AddPendingEvent(e);
	640	m_doneSent = failed \|\| m_currentType == WorkerEvent::DONE;
	641	};
	642
	643	EventWorker::~EventWorker() {
	644	m_clientSocket->Destroy();
	645	delete [] m_outbuf;
	646	delete [] m_inbuf;
	647	}
	648
	649	BEGIN_EVENT_TABLE(EventWorker,wxEvtHandler)
	650	EVT_SOCKET(wxID_ANY,EventWorker::OnSocketEvent)
	651	END_EVENT_TABLE()
	652
	653
	654	ThreadWorker::ThreadWorker(const wxString& p_host, char* p_buf, int p_size)
	655	: wxThread(wxTHREAD_DETACHED),
	656	m_host(p_host),
	657	m_outbuf(p_buf),
	658	m_outsize(p_size)
	659	{
	660	m_clientSocket = new wxSocketClient(wxSOCKET_BLOCK\|wxSOCKET_WAITALL);
	661	m_insize = m_outsize - 2;
	662	m_inbuf = new char[m_insize];
	663	}
	664
	665	wxThread::ExitCode ThreadWorker::Entry()
	666	{
	667	wxIPV4address ca;
	668	ca.Hostname(m_host);
	669	ca.Service(5678);
	670	//wxLogDebug(wxT("ThreadWorker: Connecting....."));
	671	m_clientSocket->SetTimeout(60);
	672	bool failed = false;
	673	WorkerEvent::evt_type etype = WorkerEvent::CONNECTING;
	674	if (!m_clientSocket->Connect(ca)) {
	675	wxLogError(wxT("Cannot connect to %s:%d"),ca.IPAddress().c_str(), ca.Service());
	676	failed = true;
	677	} else {
	678	//wxLogMessage(wxT("ThreadWorker: Connected. Sending %d bytes of data"),m_outsize);
	679	etype = WorkerEvent::SENDING;
	680	WorkerEvent e(this,etype);
	681	wxGetApp().AddPendingEvent(e);
	682	int to_process = m_outsize;
683	do {
684	m_clientSocket->Write(m_outbuf,m_outsize);
685	if (m_clientSocket->Error()) {
686	wxLogError(wxT("ThreadWorker: Write error"));
687	failed = true;
688	}
689	to_process -= m_clientSocket->LastCount();
690	//wxLogDebug(wxT("EventWorker: written %d bytes, %d bytes to do"),m_clientSocket->LastCount(),to_process);
691	} while(!m_clientSocket->Error() && to_process != 0);
692
693	if (!failed) {
694	etype = WorkerEvent::RECEIVING;
695	WorkerEvent e(this,etype);
696	wxGetApp().AddPendingEvent(e);
697	to_process = m_insize;
698	do {
699	m_clientSocket->Read(m_inbuf,m_insize);
700	if (m_clientSocket->Error()) {
701	wxLogError(wxT("ThreadWorker: Read error"));
702	failed = true;
703	break;
704	}
705	to_process -= m_clientSocket->LastCount();
706	//wxLogDebug(wxT("EventWorker: readed %d bytes, %d bytes to do"),m_clientSocket->LastCount(),to_process);
707	} while(!m_clientSocket->Error() && to_process != 0);
708	}
709
710	char* outdat = (char*)m_outbuf+2;
711	if (!failed && (memcmp(m_inbuf,outdat,m_insize) != 0))
712	{
713	wxLogError(wxT("Data mismatch"));
714	failed = true;
715	}
716	}
717	//wxLogDebug(wxT("ThreadWorker: Finished"));
718	if (!failed) {
719	etype = WorkerEvent::DISCONNECTING;
720	WorkerEvent e(this,etype);
721	wxGetApp().AddPendingEvent(e);
722	};
723	m_clientSocket->Close();
724	m_clientSocket->Destroy();
725	m_clientSocket = NULL;
726	delete [] m_outbuf;
727	delete [] m_inbuf;
728	if (!failed)
729	etype = WorkerEvent::DONE;
730	WorkerEvent e(this,etype);
731	if (failed) e.setFailed();
732	wxGetApp().AddPendingEvent(e);
733	return 0;
734	}
735