samples/sockets/baseserver.cpp

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