[wxWidgets.git] / samples / sockets / baseserver.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@user.sourceforge.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/datetime.h"
#include "wx/timer.h"
#include "wx/thread.h"

const char *GetSocketErrorMsg(int pSockError)
{
    switch(pSockError)
    {
        case wxSOCKET_NOERROR:
            return "wxSOCKET_NOERROR";

        case wxSOCKET_INVOP:
            return "wxSOCKET_INVOP";

        case wxSOCKET_IOERR:
            return "wxSOCKET_IOERR";

        case wxSOCKET_INVADDR:
            return "wxSOCKET_INVADDR";

        case wxSOCKET_NOHOST:
            return "wxSOCKET_NOHOST";

        case wxSOCKET_INVPORT:
            return "wxSOCKET_INVPORT";

        case wxSOCKET_WOULDBLOCK:
            return "wxSOCKET_WOULDBLOCK";

        case wxSOCKET_TIMEDOUT:
            return "wxSOCKET_TIMEDOUT";

        case wxSOCKET_MEMERR:
            return "wxSOCKET_MEMERR";

        default:
            return "Unknown";
    }
}

//event sent by workers to server class
//after client is served
const wxEventType wxEVT_WORKER = wxNewEventType();
#define EVT_WORKER(func) DECLARE_EVENT_TABLE_ENTRY( wxEVT_WORKER, -1, -1, (wxObjectEventFunction) (wxEventFunction) (WorkerEventFunction) & func, (wxObject *) NULL ),

class WorkerEvent : public wxEvent
{
public:
    WorkerEvent(void* pSender)
    {
        SetId(-1);
        SetEventType(wxEVT_WORKER);
        m_sender = pSender;
        m_exit = false;
        m_workerFailed = false;
    }

    virtual wxEvent* Clone() const
    {
        return new WorkerEvent(*this);
    }

    void* m_sender;
    bool m_exit;
    bool m_workerFailed;
};

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

class ThreadWorker;
class EventWorker;

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

//main server class contains listening socket
//and list of two type worker classes that serve clients
class Server : public wxApp
{
    DECLARE_EVENT_TABLE()
public:
    Server() : m_maxConnections(-1) {}
    ~Server() {}
private:
    enum WorkMode
    {
      MIXED,
      THREADS,
      EVENTS
    };

    virtual bool OnInit();
    virtual int OnExit();

    void OnInitCmdLine(wxCmdLineParser& pParser);
    bool OnCmdLineParsed(wxCmdLineParser& pParser);

    void OnSocketEvent(wxSocketEvent& pEvent);
    void OnWorkerEvent(WorkerEvent& pEvent);
    void OnTimerEvent(wxTimerEvent& pEvent);
    void DumpStatistics();

    TList m_threadWorkers;
    EList m_eventWorkers;
    WorkMode m_workMode;
    wxSocketServer* m_listeningSocket;

    // statistics
    unsigned m_threadWorkersCreated;
    unsigned m_threadWorkersDone;
    unsigned m_threadWorkersFailed;
    unsigned m_maxThreadWorkers;

    unsigned m_eventWorkersCreated;
    unsigned m_eventWorkersDone;
    unsigned m_eventWorkersFailed;
    unsigned m_maxEventWorkers;

    long int m_maxConnections;

    long m_port;

    wxTimer mTimer;
};

DECLARE_APP(Server);

// just some common things shared between ThreadWorker and EventWorker
class WorkerBase
{
protected:
    // outputs log message with IP and TCP port number prepended
    void LogWorker(const wxString& msg, wxLogLevel level = wxLOG_Info)
    {
        wxLogGeneric(level,
                     "%s:%d %s", m_peer.IPAddress(), m_peer.Service(), msg);
    }

    wxIPV4address m_peer;
};

//thread based worker reads signature and all data first from connected client
//and resends data to client after reading
class ThreadWorker : public wxThread, private WorkerBase
{
public:
    ThreadWorker(wxSocketBase* pSocket);
    virtual ExitCode Entry();

private:
    wxSocketBase* m_socket;
};

//event based worker reads signature and creates buffer for incoming data.
//When part of data arrives this worker resends it as soon as possible.
class EventWorker : public wxEvtHandler, private WorkerBase
{
public:
    EventWorker(wxSocketBase* pSock);
    virtual ~EventWorker();

private:
    wxSocketBase* m_socket;

    unsigned char m_signature[2];
    char* m_inbuf;
    int m_infill;
    int m_size;
    char* m_outbuf;
    int m_outfill;
    int m_written;

    void OnSocketEvent(wxSocketEvent& pEvent);
    void DoWrite();
    void DoRead();

    DECLARE_EVENT_TABLE()
};

/******************* Implementation ******************/
IMPLEMENT_APP_CONSOLE(Server)

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


void
Server::OnInitCmdLine(wxCmdLineParser& pParser)
{
    wxApp::OnInitCmdLine(pParser);
    pParser.AddSwitch("t","threads","Use thread based workers only");
    pParser.AddSwitch("e","events","Use event based workers only");
    pParser.AddOption("m","max","Exit after <n> connections",
                      wxCMD_LINE_VAL_NUMBER);
    pParser.AddOption("p","port","listen on given port (default 3000)",
                      wxCMD_LINE_VAL_NUMBER);
}

void
Server::DumpStatistics()
{
    wxString mode;
    switch(m_workMode)
    {
        case EVENTS:
            mode = "Event based workers";
            break;
        case THREADS:
            mode = "Thread based workers";
            break;
        case MIXED:
            mode = "Event and thread based workers";
            break;
    }
    wxLogMessage("Server mode: %s",mode);
    wxLogMessage("\t\t\t\tThreads\tEvents\tTotal");
    wxLogMessage("Workers created:\t\t%d\t%d\t%d",
                 m_threadWorkersCreated,
                 m_eventWorkersCreated,
                 m_threadWorkersCreated + m_eventWorkersCreated);
    wxLogMessage("Max concurrent workers:\t%d\t%d\t%d",
                 m_maxThreadWorkers,
                 m_maxEventWorkers,
                 m_maxThreadWorkers + m_maxEventWorkers);
    wxLogMessage("Workers failed:\t\t%d\t%d\t%d",
                 m_threadWorkersFailed,
                 m_eventWorkersFailed,
                 m_threadWorkersFailed + m_eventWorkersFailed);
    wxLogMessage("Workers done:\t\t%d\t%d\t%d",
                 m_threadWorkersDone,
                 m_eventWorkersDone,
                 m_threadWorkersDone + m_eventWorkersDone);

    if ((int)(m_threadWorkersDone+m_eventWorkersDone) == m_maxConnections)
    {
        wxLogMessage("%d connection(s) served, exiting",m_maxConnections);
        ExitMainLoop();
    }
}


bool
Server::OnCmdLineParsed(wxCmdLineParser& pParser)
{
    if (pParser.Found("verbose"))
    {
        wxLog::AddTraceMask("wxSocket");
        wxLog::AddTraceMask("epolldispatcher");
        wxLog::AddTraceMask("selectdispatcher");
        wxLog::AddTraceMask("thread");
        wxLog::AddTraceMask("events");
        wxLog::AddTraceMask("timer");
    }

    if (pParser.Found("m",&m_maxConnections))
    {
        wxLogMessage("%d connection(s) to exit",m_maxConnections);
    }

    if (pParser.Found("p",&m_port))
    {
        wxLogMessage("%d connection(s) to exit",m_maxConnections);
    }

    if (pParser.Found("t"))
        m_workMode = THREADS;
    else if (pParser.Found("e"))
        m_workMode = EVENTS;
    else
        m_workMode = MIXED;

    return wxApp::OnCmdLineParsed(pParser);
}

bool Server::OnInit()
{
    wxLog* logger = new wxLogStderr();
    wxLog::SetActiveTarget(logger);

    m_port = 3000;

    //send interesting things to console
    if (!wxApp::OnInit())
        return false;

    //setup listening socket
    wxIPV4address la;
    la.Service(m_port);
    m_listeningSocket = new wxSocketServer(la,wxSOCKET_NOWAIT|wxSOCKET_REUSEADDR);
    m_listeningSocket->SetEventHandler(*this);
    m_listeningSocket->SetNotify(wxSOCKET_CONNECTION_FLAG);
    m_listeningSocket->Notify(true);
    if (!m_listeningSocket->Ok())
    {
        wxLogError("Cannot bind listening socket");
        return false;
    }

    m_threadWorkersCreated = 0;
    m_threadWorkersDone = 0;
    m_threadWorkersFailed = 0;
    m_maxThreadWorkers = 0;

    m_eventWorkersCreated = 0;
    m_eventWorkersDone = 0;
    m_eventWorkersFailed = 0;
    m_maxEventWorkers = 0;

    wxLogMessage("Server listening at port %d, waiting for connections", m_port);
    return true;
}


int Server::OnExit()
{
    for ( TList::compatibility_iterator it = m_threadWorkers.GetFirst();
          it;
          it = it->GetNext() )
    {
        it->GetData()->Wait();
        delete it->GetData();
    }

    for ( EList::compatibility_iterator it2 = m_eventWorkers.GetFirst();
          it2;
          it2->GetNext() )
    {
        delete it2->GetData();
    }

    m_threadWorkers.Clear();
    m_eventWorkers.Clear();
    m_listeningSocket->Destroy();
    return 0;
}

void Server::OnSocketEvent(wxSocketEvent& pEvent)
{
    switch(pEvent.GetSocketEvent())
    {
        case wxSOCKET_INPUT:
            wxLogError("Unexpected wxSOCKET_INPUT in wxSocketServer");
            break;
        case wxSOCKET_OUTPUT:
            wxLogError("Unexpected wxSOCKET_OUTPUT in wxSocketServer");
        break;
        case wxSOCKET_CONNECTION:
        {
            wxSocketBase* sock = m_listeningSocket->Accept();
            wxIPV4address addr;
            if (!sock->GetPeer(addr))
            {
                wxLogError("Server: cannot get peer info");
            } else {
                wxLogMessage("Got connection from %s:%d",addr.IPAddress().c_str(), addr.Service());
            }
            bool createThread;

            if (m_workMode != MIXED)
                createThread = m_workMode == THREADS;
            else
                createThread = (wxDateTime::Now().GetSecond())%2 == 0;

            if (createThread)
            {
                ThreadWorker* c = new ThreadWorker(sock);
                if (c->Create() == wxTHREAD_NO_ERROR)
                {
                    m_threadWorkers.Append(c);
                    if (m_threadWorkers.GetCount() > m_maxThreadWorkers)
                    m_maxThreadWorkers++;
                    m_threadWorkersCreated++;
                    c->Run();
                }
                else
                {
                    wxLogError("Server: cannot create next thread (current threads: %d", m_threadWorkers.size());
                };
            }
            else
            {
                EventWorker* w = new EventWorker(sock);
                m_eventWorkers.Append(w);
                if (m_eventWorkers.GetCount() > m_maxEventWorkers)
                m_maxEventWorkers++;
                m_eventWorkersCreated++;
            }
        }
        break;
        case wxSOCKET_LOST:
            wxLogError("Unexpected wxSOCKET_LOST in wxSocketServer");
        break;
    }
}

void  Server::OnWorkerEvent(WorkerEvent& pEvent)
{
    //wxLogMessage("Got worker event");
    for(TList::compatibility_iterator it = m_threadWorkers.GetFirst(); it ; it = it->GetNext())
    {
        if (it->GetData() == pEvent.m_sender)
        {
            wxLogVerbose("Deleting thread worker (%d left)",
                         m_threadWorkers.GetCount());
            it->GetData()->Wait();
            delete it->GetData();
            m_threadWorkers.DeleteNode(it);
            if (!pEvent.m_workerFailed)
                m_threadWorkersDone++;
            else
                m_threadWorkersFailed++;
            break;
        }
    }
    for(EList::compatibility_iterator it2 = m_eventWorkers.GetFirst(); it2 ; it2 = it2->GetNext())
    {
        if (it2->GetData() == pEvent.m_sender)
        {
            wxLogVerbose("Deleting event worker (%d left)",
                         m_eventWorkers.GetCount());
            delete it2->GetData();
            m_eventWorkers.DeleteNode(it2);
            if (!pEvent.m_workerFailed)
                m_eventWorkersDone++;
            else
                m_eventWorkersFailed++;
            break;
        }
    }

    if (m_eventWorkers.GetCount() == 0 && m_threadWorkers.GetCount() == 0)
    {
        mTimer.Start(1000,true);
    }
}

void Server::OnTimerEvent(wxTimerEvent&)
{
  DumpStatistics();
}


BEGIN_EVENT_TABLE(Server,wxEvtHandler)
  EVT_SOCKET(wxID_ANY,Server::OnSocketEvent)
  EVT_WORKER(Server::OnWorkerEvent)
  EVT_TIMER(wxID_ANY,Server::OnTimerEvent)
END_EVENT_TABLE()


ThreadWorker::ThreadWorker(wxSocketBase* pSocket) : wxThread(wxTHREAD_JOINABLE)
{
    m_socket = pSocket;
    //Notify() cannot be called in thread context. We have to detach from main loop
    //before switching thread contexts.
    m_socket->Notify(false);
    m_socket->SetFlags(wxSOCKET_WAITALL|wxSOCKET_BLOCK);
    pSocket->GetPeer(m_peer);
}

wxThread::ExitCode ThreadWorker::Entry()
{
    WorkerEvent e(this);
    if (!m_socket->IsConnected())
    {
        LogWorker("ThreadWorker: not connected",wxLOG_Error);
        return 0;
    }
    int to_process = -1;
    if (m_socket->IsConnected())
    {
        unsigned char signature[2];
        LogWorker("ThreadWorker: reading for data");
        to_process = 2;
        do
        {
            m_socket->Read(&signature,to_process);
            if (m_socket->Error())
            {
                LogWorker("ThreadWorker: Read error",wxLOG_Error);
                wxGetApp().AddPendingEvent(e);
                return 0;
            }
            to_process -= m_socket->LastCount();
            LogWorker(wxString::Format("to_process: %d",to_process));

        }
        while (!m_socket->Error() && to_process != 0);

        if (signature[0] == 0)
        {
            e.m_exit = true;
            return 0;
        }

        if (signature[0] == 0xCE)
        {
            LogWorker("This server does not support test2 from GUI client",wxLOG_Error);
            e.m_workerFailed = true;
            e.m_exit = true;
            return 0;
        }
        int size = signature[1] * (signature[0] == 0xBE ? 1 : 1024);
        char* buf = new char[size];
        LogWorker(wxString::Format("Message signature: chunks: %d, kilobytes: %d, size: %d (bytes)",signature[0],signature[1],size));

        to_process = size;
        LogWorker(wxString::Format("ThreadWorker: reading %d bytes of data",to_process));

        do
        {
            m_socket->Read(buf,to_process);
            if (m_socket->Error())
            {
                LogWorker("ThreadWorker: Read error",wxLOG_Error);
                wxGetApp().AddPendingEvent(e);
                return 0;
            }
            to_process -= m_socket->LastCount();
            LogWorker(wxString::Format("ThreadWorker: %d bytes readed, %d todo",m_socket->LastCount(),to_process));

        }
        while(!m_socket->Error() && to_process != 0);

        to_process = size;

        do
        {
            m_socket->Write(buf,to_process);
            if (m_socket->Error())
            {
                LogWorker("ThreadWorker: Write error",wxLOG_Error);
                break;
            }
           to_process -= m_socket->LastCount();
           LogWorker(wxString::Format("ThreadWorker: %d bytes written, %d todo",m_socket->LastCount(),to_process));
        }
        while(!m_socket->Error() && to_process != 0);
    }

    LogWorker("ThreadWorker: done");
    e.m_workerFailed = to_process != 0;
    m_socket->Destroy();
    wxGetApp().AddPendingEvent(e);
    return 0;
}

EventWorker::EventWorker(wxSocketBase* pSock)
  : m_socket(pSock),
    m_inbuf(NULL),
    m_infill(0),
    m_outbuf(NULL),
    m_outfill(0)
{
    m_socket->SetNotify(wxSOCKET_LOST_FLAG|wxSOCKET_INPUT_FLAG|wxSOCKET_OUTPUT_FLAG);
    m_socket->Notify(true);
    m_socket->SetEventHandler(*this);
    m_socket->SetFlags(wxSOCKET_NOWAIT);
    m_socket->GetPeer(m_peer);
}

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

void
EventWorker::DoRead()
{
    if (m_inbuf == NULL)
    {
        //read message header
        do
        {
            m_socket->Read(m_signature + m_infill, 2 - m_infill);
            if (m_socket->Error())
            {
                if (m_socket->LastError() != wxSOCKET_WOULDBLOCK)
                {
                    LogWorker(wxString::Format("Read error (%d): %s",m_socket->LastError(),GetSocketErrorMsg(m_socket->LastError())),wxLOG_Error);
                    m_socket->Close();
                }
            }
            else
            {
                m_infill += m_socket->LastCount();
                if (m_infill == 2)
                {
                    unsigned char chunks = m_signature[1];
                    unsigned char type = m_signature[0];
                    if (type == 0xCE)
                    {
                        LogWorker("This server does not support test2 from GUI client",wxLOG_Error);
                        m_written = -1; //wxSOCKET_LOST will interpret this as failure
                        m_socket->Close();
                    }
                    else if (type == 0xBE || type == 0xDE)
                    {
                        m_size = chunks * (type == 0xBE ? 1 : 1024);
                        m_inbuf = new char[m_size];
                        m_outbuf = new char[m_size];
                        m_infill = 0;
                        m_outfill = 0;
                        m_written = 0;
                        LogWorker(wxString::Format("Message signature: len: %d, type: %s, size: %d (bytes)",chunks,type == 0xBE ? "b" : "kB",m_size));
                        break;
                    }
                    else
                    {
                        LogWorker(wxString::Format("Unknown test type %x",type));
                        m_socket->Close();
                    }
                }
            }
        }
        while(!m_socket->Error() && (2 - m_infill != 0));
    }

    if (m_inbuf == NULL)
        return;
    //read message data
    do
    {
        if (m_size == m_infill)
        {
            m_signature[0] = m_signature[1] = 0x0;
            delete [] m_inbuf;
            m_inbuf = NULL;
            m_infill = 0;
            return;
        }
        m_socket->Read(m_inbuf + m_infill,m_size - m_infill);
        if (m_socket->Error())
        {
            if (m_socket->LastError() != wxSOCKET_WOULDBLOCK)
            {
                LogWorker(wxString::Format("Read error (%d): %s",
                          m_socket->LastError(),
                          GetSocketErrorMsg(m_socket->LastError())),
                          wxLOG_Error);

                m_socket->Close();
            }
        }
        else
        {
            memcpy(m_outbuf+m_outfill,m_inbuf+m_infill,m_socket->LastCount());
            m_infill += m_socket->LastCount();
            m_outfill += m_socket->LastCount();
            DoWrite();
        }
    }
    while(!m_socket->Error());
};

void EventWorker::OnSocketEvent(wxSocketEvent& pEvent)
{
    switch(pEvent.GetSocketEvent())
    {
        case wxSOCKET_INPUT:
            DoRead();
            break;

        case wxSOCKET_OUTPUT:
            if ( m_outbuf )
                DoWrite();
            break;

        case wxSOCKET_CONNECTION:
            LogWorker("Unexpected wxSOCKET_CONNECTION in EventWorker", wxLOG_Error);
            break;

        case wxSOCKET_LOST:
            {
                LogWorker("Connection lost");
                WorkerEvent e(this);
                e.m_workerFailed = m_written != m_size;
                wxGetApp().AddPendingEvent(e);
            }
            break;
    }
}

void  EventWorker::DoWrite()
{
    do
    {
        if (m_written == m_size)
        {
            delete [] m_outbuf;
            m_outbuf = NULL;
            m_outfill = 0;
            LogWorker( "All data written");
            return;
        }
        if (m_outfill - m_written == 0)
        {
            return;
        }
        m_socket->Write(m_outbuf + m_written,m_outfill - m_written);
        if (m_socket->Error())
        {
            if (m_socket->LastError() != wxSOCKET_WOULDBLOCK)
            {
                LogWorker(
                            wxString::Format("Write error (%d): %s",
                                            m_socket->LastError(),
                                            GetSocketErrorMsg(m_socket->LastError())
                                            )
                            ,wxLOG_Error
                            );
                m_socket->Close();
            }
            else
            {
                LogWorker("Write would block, waiting for OUTPUT event");
            }
        }
        else
        {
            memmove(m_outbuf,m_outbuf+m_socket->LastCount(),m_outfill-m_socket->LastCount());
            m_written += m_socket->LastCount();
        }
        LogWorker(wxString::Format("Written %d of %d bytes, todo %d",
                  m_socket->LastCount(),m_size,m_size - m_written));
    }
    while (!m_socket->Error());
}

BEGIN_EVENT_TABLE(EventWorker,wxEvtHandler)
    EVT_SOCKET(wxID_ANY,EventWorker::OnSocketEvent)
END_EVENT_TABLE()
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@user.sourceforge.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/datetime.h"
	26	#include "wx/timer.h"
	27	#include "wx/thread.h"
	28
e2bcc781	29	const char *GetSocketErrorMsg(int pSockError)
2804f77d VZ	30	{
	31	switch(pSockError)
	32	{
	33	case wxSOCKET_NOERROR:
e2bcc781 VZ	34	return "wxSOCKET_NOERROR";
e2bcc781 VZ	35
2804f77d	36	case wxSOCKET_INVOP:
e2bcc781 VZ	37	return "wxSOCKET_INVOP";
e2bcc781 VZ	38
2804f77d	39	case wxSOCKET_IOERR:
e2bcc781 VZ	40	return "wxSOCKET_IOERR";
e2bcc781 VZ	41
2804f77d	42	case wxSOCKET_INVADDR:
e2bcc781 VZ	43	return "wxSOCKET_INVADDR";
e2bcc781 VZ	44
2804f77d	45	case wxSOCKET_NOHOST:
e2bcc781 VZ	46	return "wxSOCKET_NOHOST";
e2bcc781 VZ	47
2804f77d	48	case wxSOCKET_INVPORT:
e2bcc781 VZ	49	return "wxSOCKET_INVPORT";
e2bcc781 VZ	50
2804f77d	51	case wxSOCKET_WOULDBLOCK:
e2bcc781 VZ	52	return "wxSOCKET_WOULDBLOCK";
e2bcc781 VZ	53
2804f77d	54	case wxSOCKET_TIMEDOUT:
e2bcc781 VZ	55	return "wxSOCKET_TIMEDOUT";
e2bcc781 VZ	56
2804f77d	57	case wxSOCKET_MEMERR:
e2bcc781 VZ	58	return "wxSOCKET_MEMERR";
e2bcc781 VZ	59
2804f77d	60	default:
e2bcc781	61	return "Unknown";
2804f77d VZ	62	}
	63	}
	64
2804f77d VZ	65	//event sent by workers to server class
	66	//after client is served
	67	const wxEventType wxEVT_WORKER = wxNewEventType();
	68	#define EVT_WORKER(func) DECLARE_EVENT_TABLE_ENTRY( wxEVT_WORKER, -1, -1, (wxObjectEventFunction) (wxEventFunction) (WorkerEventFunction) & func, (wxObject *) NULL ),
	69
e2bcc781 VZ	70	class WorkerEvent : public wxEvent
e2bcc781 VZ	71	{
2804f77d VZ	72	public:
	73	WorkerEvent(void* pSender)
	74	{
	75	SetId(-1);
	76	SetEventType(wxEVT_WORKER);
	77	m_sender = pSender;
	78	m_exit = false;
	79	m_workerFailed = false;
	80	}
	81
	82	virtual wxEvent* Clone() const
	83	{
	84	return new WorkerEvent(*this);
	85	}
	86
	87	void* m_sender;
	88	bool m_exit;
	89	bool m_workerFailed;
	90	};
	91
	92	typedef void (wxEvtHandler::*WorkerEventFunction)(WorkerEvent&);
	93
	94	class ThreadWorker;
	95	class EventWorker;
	96
	97	WX_DECLARE_LIST(ThreadWorker, TList);
	98	WX_DECLARE_LIST(EventWorker, EList);
	99
	100	//main server class contains listening socket
	101	//and list of two type worker classes that serve clients
	102	class Server : public wxApp
	103	{
66034aba	104	DECLARE_EVENT_TABLE()
2804f77d VZ	105	public:
	106	Server() : m_maxConnections(-1) {}
	107	~Server() {}
	108	private:
	109	enum WorkMode
	110	{
	111	MIXED,
	112	THREADS,
	113	EVENTS
	114	};
	115
	116	virtual bool OnInit();
	117	virtual int OnExit();
	118
	119	void OnInitCmdLine(wxCmdLineParser& pParser);
	120	bool OnCmdLineParsed(wxCmdLineParser& pParser);
	121
	122	void OnSocketEvent(wxSocketEvent& pEvent);
	123	void OnWorkerEvent(WorkerEvent& pEvent);
	124	void OnTimerEvent(wxTimerEvent& pEvent);
	125	void DumpStatistics();
	126
	127	TList m_threadWorkers;
	128	EList m_eventWorkers;
	129	WorkMode m_workMode;
	130	wxSocketServer* m_listeningSocket;
	131
	132	// statistics
	133	unsigned m_threadWorkersCreated;
	134	unsigned m_threadWorkersDone;
	135	unsigned m_threadWorkersFailed;
	136	unsigned m_maxThreadWorkers;
	137
	138	unsigned m_eventWorkersCreated;
	139	unsigned m_eventWorkersDone;
	140	unsigned m_eventWorkersFailed;
	141	unsigned m_maxEventWorkers;
	142
	143	long int m_maxConnections;
	144
	145	long m_port;
	146
	147	wxTimer mTimer;
	148	};
	149
	150	DECLARE_APP(Server);
	151
c3bcb48f VZ	152	// just some common things shared between ThreadWorker and EventWorker
	153	class WorkerBase
	154	{
	155	protected:
	156	// outputs log message with IP and TCP port number prepended
	157	void LogWorker(const wxString& msg, wxLogLevel level = wxLOG_Info)
	158	{
	159	wxLogGeneric(level,
	160	"%s:%d %s", m_peer.IPAddress(), m_peer.Service(), msg);
	161	}
	162
	163	wxIPV4address m_peer;
	164	};
	165
2804f77d VZ	166	//thread based worker reads signature and all data first from connected client
2804f77d VZ	167	//and resends data to client after reading
c3bcb48f	168	class ThreadWorker : public wxThread, private WorkerBase
2804f77d VZ	169	{
	170	public:
	171	ThreadWorker(wxSocketBase* pSocket);
	172	virtual ExitCode Entry();
c3bcb48f	173
2804f77d VZ	174	private:
2804f77d VZ	175	wxSocketBase* m_socket;
2804f77d VZ	176	};
	177
	178	//event based worker reads signature and creates buffer for incoming data.
	179	//When part of data arrives this worker resends it as soon as possible.
c3bcb48f	180	class EventWorker : public wxEvtHandler, private WorkerBase
2804f77d	181	{
2804f77d VZ	182	public:
2804f77d VZ	183	EventWorker(wxSocketBase* pSock);
c3bcb48f VZ	184	virtual ~EventWorker();
c3bcb48f VZ	185
2804f77d VZ	186	private:
2804f77d VZ	187	wxSocketBase* m_socket;
2804f77d VZ	188
	189	unsigned char m_signature[2];
	190	char* m_inbuf;
	191	int m_infill;
	192	int m_size;
	193	char* m_outbuf;
	194	int m_outfill;
	195	int m_written;
	196
	197	void OnSocketEvent(wxSocketEvent& pEvent);
	198	void DoWrite();
	199	void DoRead();
c3bcb48f	200
66034aba	201	DECLARE_EVENT_TABLE()
2804f77d VZ	202	};
	203
	204	/***************** Implementation ****************/
	205	IMPLEMENT_APP_CONSOLE(Server)
	206
	207	#include <wx/listimpl.cpp>
	208	WX_DEFINE_LIST(TList);
	209	WX_DEFINE_LIST(EList);
	210
	211
	212	void
	213	Server::OnInitCmdLine(wxCmdLineParser& pParser)
	214	{
	215	wxApp::OnInitCmdLine(pParser);
c3bcb48f VZ	216	pParser.AddSwitch("t","threads","Use thread based workers only");
	217	pParser.AddSwitch("e","events","Use event based workers only");
	218	pParser.AddOption("m","max","Exit after <n> connections",
	219	wxCMD_LINE_VAL_NUMBER);
	220	pParser.AddOption("p","port","listen on given port (default 3000)",
	221	wxCMD_LINE_VAL_NUMBER);
2804f77d VZ	222	}
	223
	224	void
	225	Server::DumpStatistics()
	226	{
	227	wxString mode;
	228	switch(m_workMode)
	229	{
	230	case EVENTS:
c3bcb48f	231	mode = "Event based workers";
2804f77d VZ	232	break;
2804f77d VZ	233	case THREADS:
c3bcb48f	234	mode = "Thread based workers";
2804f77d VZ	235	break;
2804f77d VZ	236	case MIXED:
c3bcb48f	237	mode = "Event and thread based workers";
2804f77d VZ	238	break;
2804f77d VZ	239	}
e2bcc781 VZ	240	wxLogMessage("Server mode: %s",mode);
	241	wxLogMessage("\t\t\t\tThreads\tEvents\tTotal");
	242	wxLogMessage("Workers created:\t\t%d\t%d\t%d",
	243	m_threadWorkersCreated,
	244	m_eventWorkersCreated,
	245	m_threadWorkersCreated + m_eventWorkersCreated);
	246	wxLogMessage("Max concurrent workers:\t%d\t%d\t%d",
	247	m_maxThreadWorkers,
	248	m_maxEventWorkers,
	249	m_maxThreadWorkers + m_maxEventWorkers);
	250	wxLogMessage("Workers failed:\t\t%d\t%d\t%d",
	251	m_threadWorkersFailed,
	252	m_eventWorkersFailed,
	253	m_threadWorkersFailed + m_eventWorkersFailed);
	254	wxLogMessage("Workers done:\t\t%d\t%d\t%d",
	255	m_threadWorkersDone,
	256	m_eventWorkersDone,
	257	m_threadWorkersDone + m_eventWorkersDone);
2804f77d VZ	258
	259	if ((int)(m_threadWorkersDone+m_eventWorkersDone) == m_maxConnections)
	260	{
e2bcc781	261	wxLogMessage("%d connection(s) served, exiting",m_maxConnections);
2804f77d VZ	262	ExitMainLoop();
	263	}
	264	}
	265
	266
	267	bool
	268	Server::OnCmdLineParsed(wxCmdLineParser& pParser)
	269	{
c3bcb48f	270	if (pParser.Found("verbose"))
2804f77d	271	{
e2bcc781 VZ	272	wxLog::AddTraceMask("wxSocket");
	273	wxLog::AddTraceMask("epolldispatcher");
	274	wxLog::AddTraceMask("selectdispatcher");
	275	wxLog::AddTraceMask("thread");
	276	wxLog::AddTraceMask("events");
	277	wxLog::AddTraceMask("timer");
2804f77d VZ	278	}
2804f77d VZ	279
e2bcc781	280	if (pParser.Found("m",&m_maxConnections))
2804f77d	281	{
e2bcc781	282	wxLogMessage("%d connection(s) to exit",m_maxConnections);
2804f77d VZ	283	}
2804f77d VZ	284
e2bcc781	285	if (pParser.Found("p",&m_port))
2804f77d	286	{
e2bcc781	287	wxLogMessage("%d connection(s) to exit",m_maxConnections);
2804f77d VZ	288	}
2804f77d VZ	289
e2bcc781	290	if (pParser.Found("t"))
2804f77d	291	m_workMode = THREADS;
e2bcc781	292	else if (pParser.Found("e"))
2804f77d VZ	293	m_workMode = EVENTS;
	294	else
	295	m_workMode = MIXED;
	296
	297	return wxApp::OnCmdLineParsed(pParser);
	298	}
	299
	300	bool Server::OnInit()
	301	{
	302	wxLog* logger = new wxLogStderr();
	303	wxLog::SetActiveTarget(logger);
	304
	305	m_port = 3000;
	306
	307	//send interesting things to console
	308	if (!wxApp::OnInit())
	309	return false;
	310
	311	//setup listening socket
	312	wxIPV4address la;
	313	la.Service(m_port);
	314	m_listeningSocket = new wxSocketServer(la,wxSOCKET_NOWAIT\|wxSOCKET_REUSEADDR);
	315	m_listeningSocket->SetEventHandler(*this);
	316	m_listeningSocket->SetNotify(wxSOCKET_CONNECTION_FLAG);
	317	m_listeningSocket->Notify(true);
	318	if (!m_listeningSocket->Ok())
	319	{
e2bcc781	320	wxLogError("Cannot bind listening socket");
2804f77d VZ	321	return false;
	322	}
	323
	324	m_threadWorkersCreated = 0;
	325	m_threadWorkersDone = 0;
	326	m_threadWorkersFailed = 0;
	327	m_maxThreadWorkers = 0;
	328
	329	m_eventWorkersCreated = 0;
	330	m_eventWorkersDone = 0;
	331	m_eventWorkersFailed = 0;
	332	m_maxEventWorkers = 0;
	333
e2bcc781	334	wxLogMessage("Server listening at port %d, waiting for connections", m_port);
2804f77d VZ	335	return true;
	336	}
	337
	338
	339	int Server::OnExit()
	340	{
e2bcc781 VZ	341	for ( TList::compatibility_iterator it = m_threadWorkers.GetFirst();
	342	it;
	343	it = it->GetNext() )
	344	{
2804f77d VZ	345	it->GetData()->Wait();
	346	delete it->GetData();
	347	}
	348
e2bcc781 VZ	349	for ( EList::compatibility_iterator it2 = m_eventWorkers.GetFirst();
	350	it2;
	351	it2->GetNext() )
	352	{
45a50a2e	353	delete it2->GetData();
2804f77d VZ	354	}
	355
	356	m_threadWorkers.Clear();
	357	m_eventWorkers.Clear();
	358	m_listeningSocket->Destroy();
	359	return 0;
	360	}
	361
	362	void Server::OnSocketEvent(wxSocketEvent& pEvent)
	363	{
	364	switch(pEvent.GetSocketEvent())
	365	{
	366	case wxSOCKET_INPUT:
e2bcc781 VZ	367	wxLogError("Unexpected wxSOCKET_INPUT in wxSocketServer");
e2bcc781 VZ	368	break;
2804f77d	369	case wxSOCKET_OUTPUT:
e2bcc781	370	wxLogError("Unexpected wxSOCKET_OUTPUT in wxSocketServer");
2804f77d VZ	371	break;
	372	case wxSOCKET_CONNECTION:
	373	{
	374	wxSocketBase* sock = m_listeningSocket->Accept();
	375	wxIPV4address addr;
	376	if (!sock->GetPeer(addr))
	377	{
e2bcc781	378	wxLogError("Server: cannot get peer info");
2804f77d	379	} else {
e2bcc781	380	wxLogMessage("Got connection from %s:%d",addr.IPAddress().c_str(), addr.Service());
2804f77d VZ	381	}
	382	bool createThread;
	383
	384	if (m_workMode != MIXED)
	385	createThread = m_workMode == THREADS;
	386	else
	387	createThread = (wxDateTime::Now().GetSecond())%2 == 0;
	388
	389	if (createThread)
	390	{
	391	ThreadWorker* c = new ThreadWorker(sock);
	392	if (c->Create() == wxTHREAD_NO_ERROR)
	393	{
	394	m_threadWorkers.Append(c);
	395	if (m_threadWorkers.GetCount() > m_maxThreadWorkers)
	396	m_maxThreadWorkers++;
	397	m_threadWorkersCreated++;
	398	c->Run();
	399	}
	400	else
	401	{
e2bcc781	402	wxLogError("Server: cannot create next thread (current threads: %d", m_threadWorkers.size());
2804f77d VZ	403	};
	404	}
	405	else
	406	{
	407	EventWorker* w = new EventWorker(sock);
	408	m_eventWorkers.Append(w);
	409	if (m_eventWorkers.GetCount() > m_maxEventWorkers)
	410	m_maxEventWorkers++;
	411	m_eventWorkersCreated++;
	412	}
	413	}
	414	break;
	415	case wxSOCKET_LOST:
e2bcc781	416	wxLogError("Unexpected wxSOCKET_LOST in wxSocketServer");
2804f77d VZ	417	break;
	418	}
	419	}
	420
	421	void Server::OnWorkerEvent(WorkerEvent& pEvent)
	422	{
e2bcc781	423	//wxLogMessage("Got worker event");
2804f77d VZ	424	for(TList::compatibility_iterator it = m_threadWorkers.GetFirst(); it ; it = it->GetNext())
	425	{
	426	if (it->GetData() == pEvent.m_sender)
	427	{
e2bcc781 VZ	428	wxLogVerbose("Deleting thread worker (%d left)",
e2bcc781 VZ	429	m_threadWorkers.GetCount());
2804f77d VZ	430	it->GetData()->Wait();
	431	delete it->GetData();
	432	m_threadWorkers.DeleteNode(it);
	433	if (!pEvent.m_workerFailed)
	434	m_threadWorkersDone++;
	435	else
	436	m_threadWorkersFailed++;
	437	break;
	438	}
	439	}
45a50a2e	440	for(EList::compatibility_iterator it2 = m_eventWorkers.GetFirst(); it2 ; it2 = it2->GetNext())
2804f77d	441	{
45a50a2e	442	if (it2->GetData() == pEvent.m_sender)
2804f77d	443	{
e2bcc781 VZ	444	wxLogVerbose("Deleting event worker (%d left)",
e2bcc781 VZ	445	m_eventWorkers.GetCount());
45a50a2e VZ	446	delete it2->GetData();
45a50a2e VZ	447	m_eventWorkers.DeleteNode(it2);
2804f77d VZ	448	if (!pEvent.m_workerFailed)
	449	m_eventWorkersDone++;
	450	else
	451	m_eventWorkersFailed++;
	452	break;
	453	}
	454	}
	455
	456	if (m_eventWorkers.GetCount() == 0 && m_threadWorkers.GetCount() == 0)
	457	{
	458	mTimer.Start(1000,true);
	459	}
	460	}
	461
	462	void Server::OnTimerEvent(wxTimerEvent&)
	463	{
	464	DumpStatistics();
	465	}
	466
	467
	468	BEGIN_EVENT_TABLE(Server,wxEvtHandler)
	469	EVT_SOCKET(wxID_ANY,Server::OnSocketEvent)
	470	EVT_WORKER(Server::OnWorkerEvent)
	471	EVT_TIMER(wxID_ANY,Server::OnTimerEvent)
	472	END_EVENT_TABLE()
	473
	474
	475	ThreadWorker::ThreadWorker(wxSocketBase* pSocket) : wxThread(wxTHREAD_JOINABLE)
	476	{
	477	m_socket = pSocket;
	478	//Notify() cannot be called in thread context. We have to detach from main loop
	479	//before switching thread contexts.
	480	m_socket->Notify(false);
	481	m_socket->SetFlags(wxSOCKET_WAITALL\|wxSOCKET_BLOCK);
	482	pSocket->GetPeer(m_peer);
	483	}
	484
	485	wxThread::ExitCode ThreadWorker::Entry()
	486	{
	487	WorkerEvent e(this);
	488	if (!m_socket->IsConnected())
	489	{
c3bcb48f	490	LogWorker("ThreadWorker: not connected",wxLOG_Error);
2804f77d VZ	491	return 0;
	492	}
	493	int to_process = -1;
	494	if (m_socket->IsConnected())
	495	{
	496	unsigned char signature[2];
c3bcb48f	497	LogWorker("ThreadWorker: reading for data");
2804f77d VZ	498	to_process = 2;
	499	do
	500	{
	501	m_socket->Read(&signature,to_process);
	502	if (m_socket->Error())
	503	{
c3bcb48f	504	LogWorker("ThreadWorker: Read error",wxLOG_Error);
2804f77d VZ	505	wxGetApp().AddPendingEvent(e);
	506	return 0;
	507	}
	508	to_process -= m_socket->LastCount();
c3bcb48f	509	LogWorker(wxString::Format("to_process: %d",to_process));
2804f77d VZ	510
	511	}
	512	while (!m_socket->Error() && to_process != 0);
	513
	514	if (signature[0] == 0)
	515	{
	516	e.m_exit = true;
	517	return 0;
	518	}
	519
	520	if (signature[0] == 0xCE)
	521	{
c3bcb48f	522	LogWorker("This server does not support test2 from GUI client",wxLOG_Error);
2804f77d VZ	523	e.m_workerFailed = true;
	524	e.m_exit = true;
	525	return 0;
	526	}
	527	int size = signature[1] * (signature[0] == 0xBE ? 1 : 1024);
	528	char* buf = new char[size];
c3bcb48f	529	LogWorker(wxString::Format("Message signature: chunks: %d, kilobytes: %d, size: %d (bytes)",signature[0],signature[1],size));
2804f77d VZ	530
2804f77d VZ	531	to_process = size;
c3bcb48f	532	LogWorker(wxString::Format("ThreadWorker: reading %d bytes of data",to_process));
2804f77d VZ	533
	534	do
	535	{
	536	m_socket->Read(buf,to_process);
	537	if (m_socket->Error())
	538	{
c3bcb48f	539	LogWorker("ThreadWorker: Read error",wxLOG_Error);
2804f77d VZ	540	wxGetApp().AddPendingEvent(e);
	541	return 0;
	542	}
	543	to_process -= m_socket->LastCount();
c3bcb48f	544	LogWorker(wxString::Format("ThreadWorker: %d bytes readed, %d todo",m_socket->LastCount(),to_process));
2804f77d VZ	545
	546	}
	547	while(!m_socket->Error() && to_process != 0);
	548
	549	to_process = size;
	550
	551	do
	552	{
	553	m_socket->Write(buf,to_process);
e2bcc781 VZ	554	if (m_socket->Error())
e2bcc781 VZ	555	{
c3bcb48f	556	LogWorker("ThreadWorker: Write error",wxLOG_Error);
2804f77d	557	break;
e2bcc781	558	}
2804f77d	559	to_process -= m_socket->LastCount();
c3bcb48f	560	LogWorker(wxString::Format("ThreadWorker: %d bytes written, %d todo",m_socket->LastCount(),to_process));
2804f77d VZ	561	}
	562	while(!m_socket->Error() && to_process != 0);
	563	}
	564
c3bcb48f	565	LogWorker("ThreadWorker: done");
2804f77d VZ	566	e.m_workerFailed = to_process != 0;
	567	m_socket->Destroy();
	568	wxGetApp().AddPendingEvent(e);
	569	return 0;
	570	}
	571
	572	EventWorker::EventWorker(wxSocketBase* pSock)
	573	: m_socket(pSock),
	574	m_inbuf(NULL),
	575	m_infill(0),
	576	m_outbuf(NULL),
	577	m_outfill(0)
	578	{
	579	m_socket->SetNotify(wxSOCKET_LOST_FLAG\|wxSOCKET_INPUT_FLAG\|wxSOCKET_OUTPUT_FLAG);
	580	m_socket->Notify(true);
	581	m_socket->SetEventHandler(*this);
	582	m_socket->SetFlags(wxSOCKET_NOWAIT);
	583	m_socket->GetPeer(m_peer);
	584	}
	585
e2bcc781 VZ	586	EventWorker::~EventWorker()
e2bcc781 VZ	587	{
2804f77d VZ	588	m_socket->Destroy();
	589	delete [] m_inbuf;
	590	delete [] m_outbuf;
	591	}
	592
	593	void
e2bcc781	594	EventWorker::DoRead()
2804f77d VZ	595	{
	596	if (m_inbuf == NULL)
	597	{
	598	//read message header
	599	do
	600	{
0f359e5c	601	m_socket->Read(m_signature + m_infill, 2 - m_infill);
e2bcc781 VZ	602	if (m_socket->Error())
e2bcc781 VZ	603	{
2804f77d VZ	604	if (m_socket->LastError() != wxSOCKET_WOULDBLOCK)
2804f77d VZ	605	{
c3bcb48f	606	LogWorker(wxString::Format("Read error (%d): %s",m_socket->LastError(),GetSocketErrorMsg(m_socket->LastError())),wxLOG_Error);
2804f77d VZ	607	m_socket->Close();
	608	}
	609	}
	610	else
	611	{
	612	m_infill += m_socket->LastCount();
e2bcc781 VZ	613	if (m_infill == 2)
e2bcc781 VZ	614	{
2804f77d VZ	615	unsigned char chunks = m_signature[1];
	616	unsigned char type = m_signature[0];
	617	if (type == 0xCE)
	618	{
c3bcb48f	619	LogWorker("This server does not support test2 from GUI client",wxLOG_Error);
2804f77d VZ	620	m_written = -1; //wxSOCKET_LOST will interpret this as failure
	621	m_socket->Close();
	622	}
	623	else if (type == 0xBE \|\| type == 0xDE)
	624	{
	625	m_size = chunks * (type == 0xBE ? 1 : 1024);
	626	m_inbuf = new char[m_size];
	627	m_outbuf = new char[m_size];
	628	m_infill = 0;
	629	m_outfill = 0;
	630	m_written = 0;
c3bcb48f	631	LogWorker(wxString::Format("Message signature: len: %d, type: %s, size: %d (bytes)",chunks,type == 0xBE ? "b" : "kB",m_size));
2804f77d	632	break;
0f359e5c VZ	633	}
0f359e5c VZ	634	else
2804f77d	635	{
c3bcb48f	636	LogWorker(wxString::Format("Unknown test type %x",type));
2804f77d VZ	637	m_socket->Close();
	638	}
	639	}
	640	}
	641	}
	642	while(!m_socket->Error() && (2 - m_infill != 0));
	643	}
	644
	645	if (m_inbuf == NULL)
	646	return;
	647	//read message data
e2bcc781 VZ	648	do
	649	{
	650	if (m_size == m_infill)
	651	{
2804f77d VZ	652	m_signature[0] = m_signature[1] = 0x0;
	653	delete [] m_inbuf;
	654	m_inbuf = NULL;
	655	m_infill = 0;
	656	return;
	657	}
	658	m_socket->Read(m_inbuf + m_infill,m_size - m_infill);
e2bcc781 VZ	659	if (m_socket->Error())
e2bcc781 VZ	660	{
2804f77d VZ	661	if (m_socket->LastError() != wxSOCKET_WOULDBLOCK)
2804f77d VZ	662	{
c3bcb48f VZ	663	LogWorker(wxString::Format("Read error (%d): %s",
	664	m_socket->LastError(),
	665	GetSocketErrorMsg(m_socket->LastError())),
	666	wxLOG_Error);
2804f77d VZ	667
	668	m_socket->Close();
	669	}
	670	}
	671	else
	672	{
	673	memcpy(m_outbuf+m_outfill,m_inbuf+m_infill,m_socket->LastCount());
	674	m_infill += m_socket->LastCount();
	675	m_outfill += m_socket->LastCount();
	676	DoWrite();
	677	}
	678	}
	679	while(!m_socket->Error());
	680	};
	681
	682	void EventWorker::OnSocketEvent(wxSocketEvent& pEvent)
	683	{
	684	switch(pEvent.GetSocketEvent())
	685	{
	686	case wxSOCKET_INPUT:
	687	DoRead();
e2bcc781 VZ	688	break;
e2bcc781 VZ	689
2804f77d	690	case wxSOCKET_OUTPUT:
7ea70b6f	691	if ( m_outbuf )
2804f77d	692	DoWrite();
e2bcc781 VZ	693	break;
e2bcc781 VZ	694
2804f77d	695	case wxSOCKET_CONNECTION:
7ea70b6f	696	LogWorker("Unexpected wxSOCKET_CONNECTION in EventWorker", wxLOG_Error);
e2bcc781 VZ	697	break;
e2bcc781 VZ	698
2804f77d	699	case wxSOCKET_LOST:
e2bcc781	700	{
7ea70b6f	701	LogWorker("Connection lost");
e2bcc781 VZ	702	WorkerEvent e(this);
	703	e.m_workerFailed = m_written != m_size;
	704	wxGetApp().AddPendingEvent(e);
	705	}
	706	break;
2804f77d VZ	707	}
	708	}
	709
e2bcc781 VZ	710	void EventWorker::DoWrite()
	711	{
	712	do
	713	{
2804f77d VZ	714	if (m_written == m_size)
2804f77d VZ	715	{
e2bcc781 VZ	716	delete [] m_outbuf;
	717	m_outbuf = NULL;
	718	m_outfill = 0;
c3bcb48f	719	LogWorker( "All data written");
e2bcc781	720	return;
2804f77d VZ	721	}
	722	if (m_outfill - m_written == 0)
	723	{
	724	return;
	725	}
	726	m_socket->Write(m_outbuf + m_written,m_outfill - m_written);
	727	if (m_socket->Error())
	728	{
	729	if (m_socket->LastError() != wxSOCKET_WOULDBLOCK)
	730	{
c3bcb48f	731	LogWorker(
e2bcc781	732	wxString::Format("Write error (%d): %s",
2804f77d VZ	733	m_socket->LastError(),
	734	GetSocketErrorMsg(m_socket->LastError())
	735	)
e2bcc781	736	,wxLOG_Error
2804f77d VZ	737	);
	738	m_socket->Close();
	739	}
	740	else
	741	{
c3bcb48f	742	LogWorker("Write would block, waiting for OUTPUT event");
2804f77d VZ	743	}
	744	}
	745	else
	746	{
	747	memmove(m_outbuf,m_outbuf+m_socket->LastCount(),m_outfill-m_socket->LastCount());
	748	m_written += m_socket->LastCount();
	749	}
c3bcb48f VZ	750	LogWorker(wxString::Format("Written %d of %d bytes, todo %d",
c3bcb48f VZ	751	m_socket->LastCount(),m_size,m_size - m_written));
2804f77d VZ	752	}
	753	while (!m_socket->Error());
	754	}
	755
	756	BEGIN_EVENT_TABLE(EventWorker,wxEvtHandler)
e2bcc781	757	EVT_SOCKET(wxID_ANY,EventWorker::OnSocketEvent)
2804f77d	758	END_EVENT_TABLE()