Amebis/wxWidgets
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use main thread for the socket poller if possible

Browse Source 
On windows and systems where wxUSE_EVENTLOOP_SOURCE is 1, it is
possible to monitor socket descriptors for activity from the main
thread. The SocketPoller class used with wxWebSessionCURL is modified
to use an implementation class that does so.

On windows, the implementation uses the winsock1 function
WSAAsyncSelect to send events to wxWebSessionCURL when activity is
detected. When wxUSE_EVENTLOOP_SOURCE is 1, a wxEventLoopSource is used
to monitor for socket activity. The event loop source is given a custom
wxEventLoopSourceHandler object to send the necessary event.
This commit is contained in:
New Pagodi
2021-02-07 20:55:24 -06:00
parent 88dca37b3f
commit 252a920bd7
1 changed files with 363 additions and 415 deletions
Download Patch File Download Diff File Expand all files Collapse all files

778
src/common/webrequest_curl.cpp
View File

@@ -24,19 +24,17 @@
#include "wx/uri.h"
#include "wx/socket.h"
#include "wx/msgqueue.h"
#include "wx/hashset.h"
#include "wx/hashmap.h"
#include "wx/evtloop.h"
#ifdef __WINDOWS__
#include <wx/msw/wrapwin.h>
#include "wx/hashset.h"
#include "wx/msw/wrapwin.h"
#else
#include <sys/socket.h>
#include <sys/select.h>
#include "wx/evtloopsrc.h"
#include "wx/evtloop.h"
#endif
// Define symbols that might be missing from older libcurl headers
#ifndef CURL_AT_LEAST_VERSION
#define CURL_VERSION_BITS(x,y,z) ((x)<<16|(y)<<8|z)
@@ -422,503 +420,451 @@ void wxWebAuthChallengeCURL::SetCredentials(const wxWebCredentials& cred)
m_request.StartRequest();
}
//
// SocketPoller - a helper class for wxWebSessionCURL
//
wxDECLARE_EVENT(wxSocketAction,wxThreadEvent);
wxDECLARE_EVENT(wxEVT_SOCKET_POLLER_RESULT, wxThreadEvent);
class SocketPoller: public wxThreadHelper
class SocketPollerImpl;
class SocketPoller
{
public:
enum PollAction
{
InvalidAction = 0,
PollForRead = 1,
PollForWrite = 2,
PollForError = 4
INVALID_ACTION = 0x00,
POLL_FOR_READ = 0x01,
POLL_FOR_WRITE = 0x02
};
enum Result
{
InvalidResult = 0,
ReadyForRead = 1,
ReadyForWrite = 2,
HasError = 4
INVALID_RESULT = 0x00,
READY_FOR_READ = 0x01,
READY_FOR_WRITE = 0x02,
HAS_ERROR = 0x04
};
SocketPoller();
SocketPoller(wxEvtHandler*);
~SocketPoller();
bool StartPolling(wxSOCKET_T, int, wxEvtHandler*);
bool StartPolling(wxSOCKET_T, int);
void StopPolling(wxSOCKET_T);
void ResumePolling(wxSOCKET_T);
private:
static const wxSOCKET_T SOCKET_POLLER_INVALID_SOCKET;
class Message
{
public:
enum MessageId
{
AddSocketAction,
DeleteSocketAction,
ResumePolling,
Quit,
LastMessageId
};
explicit Message(MessageId i = LastMessageId,
wxSOCKET_T s = SOCKET_POLLER_INVALID_SOCKET,
int f = 0,
wxEvtHandler* h = NULL)
{
m_messageId = i;
m_socket = s;
m_flags = f;
m_evtHandler = h;
}
MessageId GetMessageId() const { return m_messageId; }
wxSOCKET_T GetSocket() const { return m_socket; }
int GetFlags() const { return m_flags; }
wxEvtHandler* GetEvtHandler() const { return m_evtHandler; }
private:
wxSOCKET_T m_socket;
int m_flags;
MessageId m_messageId;
wxEvtHandler* m_evtHandler;
};
struct SocketData
{
int m_pollAction;
void* m_event;
wxEvtHandler* m_handler;
};
WX_DECLARE_HASH_SET(wxSOCKET_T, wxIntegerHash, wxIntegerEqual, SocketSet );
WX_DECLARE_HASH_MAP(wxSOCKET_T, SocketData, wxIntegerHash, wxIntegerEqual, \
SocketDataMap );
// Items used from main thread.
void CreateSocketPair();
void CloseSocket(wxSOCKET_T);
void PostAndSignal(const Message&);
SocketSet m_polledSockets;
wxSOCKET_T m_writeEnd;
// Items used in both threads.
wxMessageQueue<Message> m_msgQueue;
wxSOCKET_T m_readEnd;
// Items used from worker thread.
virtual wxThread::ExitCode Entry() wxOVERRIDE;
void ThreadRemoveFromDL(wxSOCKET_T);
void ThreadSetSocketAction(wxSOCKET_T, int,wxEvtHandler*);
void ThreadDeleteSocketAction(wxSOCKET_T);
void ThreadCheckSockets();
SocketDataMap m_socketData;
SocketSet m_disabledList;
SocketPollerImpl* m_impl;
};
wxDEFINE_EVENT(wxSocketAction,wxThreadEvent);
wxDEFINE_EVENT(wxEVT_SOCKET_POLLER_RESULT, wxThreadEvent);
#ifdef INVALID_SOCKET
const wxSOCKET_T SocketPoller::SOCKET_POLLER_INVALID_SOCKET =
INVALID_SOCKET;
#else
const wxSOCKET_T SocketPoller::SOCKET_POLLER_INVALID_SOCKET =
static_cast<wxSOCKET_T>(~0);
#endif
SocketPoller::SocketPoller()
class SocketPollerImpl
{
m_writeEnd = static_cast<wxSOCKET_T>(SocketPoller::SOCKET_POLLER_INVALID_SOCKET);
m_readEnd = SocketPoller::SOCKET_POLLER_INVALID_SOCKET;
CreateSocketPair();
public:
virtual ~SocketPollerImpl(){};
virtual bool StartPolling(wxSOCKET_T, int) = 0;
virtual void StopPolling(wxSOCKET_T) = 0;
virtual void ResumePolling(wxSOCKET_T) = 0;
if (CreateThread(wxTHREAD_JOINABLE) != wxTHREAD_NO_ERROR)
{
wxLogDebug("Could not create socket poller worker thread!");
return;
}
static SocketPollerImpl* Create(wxEvtHandler*);
};
if (GetThread()->Run() != wxTHREAD_NO_ERROR)
{
wxLogDebug("Could not run socket poller worker thread!");
return;
}
SocketPoller::SocketPoller(wxEvtHandler* hndlr)
{
m_impl = SocketPollerImpl::Create(hndlr);
}
SocketPoller::~SocketPoller()
{
SocketPoller::Message msg(Message::Quit);
PostAndSignal(msg);
GetThread()->Wait();
CloseSocket(m_writeEnd);
CloseSocket(m_readEnd);
#ifdef __WINDOWS__
WSACleanup();
#endif
delete m_impl;
}
void SocketPoller::CreateSocketPair()
bool SocketPoller::StartPolling(wxSOCKET_T sock, int pollAction)
{
return m_impl->StartPolling(sock, pollAction);
}
void SocketPoller::StopPolling(wxSOCKET_T sock)
{
m_impl->StopPolling(sock);
}
void SocketPoller::ResumePolling(wxSOCKET_T sock)
{
m_impl->ResumePolling(sock);
}
#ifdef __WINDOWS__
class WinSock1SocketPoller: public SocketPollerImpl
{
public:
WinSock1SocketPoller(wxEvtHandler*);
virtual ~WinSock1SocketPoller();
virtual bool StartPolling(wxSOCKET_T, int) wxOVERRIDE;
virtual void StopPolling(wxSOCKET_T) wxOVERRIDE;
virtual void ResumePolling(wxSOCKET_T) wxOVERRIDE;
private:
static LRESULT CALLBACK MsgProc(HWND hwnd, WXUINT uMsg, WXWPARAM wParam,
WXLPARAM lParam);
static const WXUINT SOCKET_MESSAGE;
WX_DECLARE_HASH_SET(wxSOCKET_T, wxIntegerHash, wxIntegerEqual, SocketSet);
SocketSet m_polledSockets;
WXHWND m_hwnd;
};
const WXUINT WinSock1SocketPoller::SOCKET_MESSAGE = WM_USER + 1;
WinSock1SocketPoller::WinSock1SocketPoller(wxEvtHandler* hndlr)
{
// Initialize winsock in case it's not already done.
WORD wVersionRequested = MAKEWORD(1,1);
WSADATA wsaData;
WSAStartup(wVersionRequested, &wsaData);
m_writeEnd = socket(AF_INET, SOCK_DGRAM, 0);
// Create a dummy message only window.
m_hwnd = CreateWindowEx(
0, //DWORD dwExStyle,
TEXT("STATIC"), //LPCSTR lpClassName,
NULL, //LPCSTR lpWindowName,
0, //DWORD dwStyle,
0, //int X,
0, //int Y,
0, //int nWidth,
0, //int nHeight,
HWND_MESSAGE, //HWND hWndParent,
NULL, //HMENU hMenu,
NULL, //HINSTANCE hInstance,
NULL //LPVOID lpParam
);
if ( m_writeEnd == INVALID_SOCKET )
if ( m_hwnd == NULL )
{
wxLogDebug("Unable to create write end of socket pair.");
m_writeEnd = SocketPoller::SOCKET_POLLER_INVALID_SOCKET;
wxLogError("Unable to create message window for WinSock1SocketPoller");
return;
}
m_readEnd = socket(AF_INET, SOCK_DGRAM, 0);
if ( m_readEnd == INVALID_SOCKET )
{
wxLogDebug("Unable to create read end of socket pair.");
CloseSocket(m_writeEnd);
m_writeEnd = SocketPoller::SOCKET_POLLER_INVALID_SOCKET;
m_readEnd = SocketPoller::SOCKET_POLLER_INVALID_SOCKET;
return;
}
// Bind the read end. This will assign it to an unused port.
struct sockaddr_in serverAddr;
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = htons(0);
serverAddr.sin_addr.s_addr = inet_addr("127.0.0.1");
int retcode = bind(m_readEnd, (struct sockaddr *)&serverAddr,
sizeof(serverAddr));
if ( retcode < 0 )
{
wxLogDebug("Unable bind socket to port.");
CloseSocket(m_writeEnd);
m_writeEnd = SocketPoller::SOCKET_POLLER_INVALID_SOCKET;
CloseSocket(m_readEnd);
m_readEnd = SocketPoller::SOCKET_POLLER_INVALID_SOCKET;
return;
}
// Get the ip address and port of the read end.
struct sockaddr_in readEndAddr;
ZeroMemory(&readEndAddr, sizeof(readEndAddr));
int len = sizeof(readEndAddr);
getsockname(m_readEnd, (struct sockaddr *) &readEndAddr, &len);
// Unlike with stream sockets, this is a synchronous operation and just
// sets location for the send function.
connect(m_writeEnd, (struct sockaddr *)&readEndAddr, sizeof(readEndAddr));
#else
int fd[2];
socketpair(AF_UNIX, SOCK_STREAM, 0, fd);
m_writeEnd = fd[0];
m_readEnd = fd[1];
#endif
// Set the event handler to be the message window's user data. Also set the
// message window to use our MsgProc to process messages it receives.
SetWindowLongPtr(m_hwnd, GWLP_USERDATA, reinterpret_cast<LONG_PTR>(hndlr));
SetWindowLongPtr(m_hwnd, GWLP_WNDPROC,
reinterpret_cast<LONG_PTR>(WinSock1SocketPoller::MsgProc));
}
void SocketPoller::CloseSocket(wxSOCKET_T s)
WinSock1SocketPoller::~WinSock1SocketPoller()
{
#ifdef __WINDOWS__
closesocket(s);
#else
close(s);
#endif
// Stop monitoring any leftover sockets.
for ( SocketSet::iterator it = m_polledSockets.begin() ;
it != m_polledSockets.end() ; ++it )
{
WSAAsyncSelect(*it, m_hwnd, 0, 0);
}
// Close the message window.
if ( m_hwnd )
{
CloseWindow(m_hwnd);
}
// Cleanup winsock.
WSACleanup();
}
bool SocketPoller::StartPolling(wxSOCKET_T s, int flags, wxEvtHandler* hndlr)
bool WinSock1SocketPoller::StartPolling(wxSOCKET_T sock, int pollAction)
{
SocketSet::iterator it = m_polledSockets.find(s);
StopPolling(sock);
if ( m_polledSockets.size() > 1023 )
// Convert pollAction to a flag that can be used by winsock.
int winActions = 0;
if ( pollAction & SocketPoller::POLL_FOR_READ )
{
if ( it == m_polledSockets.end() )
{
return false;
}
winActions |= FD_READ;
}
if ( it == m_polledSockets.end() )
if ( pollAction & SocketPoller::POLL_FOR_WRITE )
{
m_polledSockets.insert(s);
winActions |= FD_WRITE;
}
SocketPoller::Message msg(Message::AddSocketAction, s, flags, hndlr);
PostAndSignal(msg);
// Have winsock send a message to our window whenever activity is
// detected on the socket.
WSAAsyncSelect(sock, m_hwnd, SOCKET_MESSAGE, winActions);
m_polledSockets.insert(sock);
return true;
}
void SocketPoller::StopPolling(wxSOCKET_T s)
void WinSock1SocketPoller::StopPolling(wxSOCKET_T sock)
{
SocketPoller::Message msg(Message::DeleteSocketAction, s);
PostAndSignal(msg);
SocketSet::iterator it = m_polledSockets.find(s);
SocketSet::iterator it = m_polledSockets.find(sock);
if ( it != m_polledSockets.end() )
{
// Stop sending messages when there is activity on the socket.
WSAAsyncSelect(sock, m_hwnd, 0, 0);
m_polledSockets.erase(it);
}
}
void SocketPoller::ResumePolling(wxSOCKET_T s)
void WinSock1SocketPoller::ResumePolling(wxSOCKET_T WXUNUSED(sock))
{
SocketPoller::Message msg(Message::ResumePolling, s);
PostAndSignal(msg);
}
void SocketPoller::PostAndSignal(const Message& msg)
LRESULT CALLBACK WinSock1SocketPoller::MsgProc(WXHWND hwnd, WXUINT uMsg,
WXWPARAM wParam, WXLPARAM lParam)
{
m_msgQueue.Post(msg);
// We only handle 1 message - the message we told winsock to send when
// it notices activity on sockets we are monitoring.
if ( m_writeEnd != SocketPoller::SOCKET_POLLER_INVALID_SOCKET )
if ( uMsg == SOCKET_MESSAGE )
{
char c = 32;
send(m_writeEnd, &c, 1, 0);
// Extract the result any any errors from lParam.
int winResult = LOWORD(lParam);
int error = HIWORD(lParam);
// Convert the result/errors to a SocketPoller::Result flag.
int pollResult = 0;
if ( winResult & FD_READ )
{
pollResult |= SocketPoller::READY_FOR_READ;
}
if ( winResult & FD_WRITE )
{
pollResult |= SocketPoller::READY_FOR_WRITE;
}
if ( error != 0 )
{
pollResult |= SocketPoller::HAS_ERROR;
}
// If there is a significant result, send an event.
if ( pollResult != 0 )
{
// The event handler is stored in the window's user data and the
// socket with activity is given by wParam.
LONG_PTR userData = GetWindowLongPtr(hwnd, GWLP_USERDATA);
wxEvtHandler* hndlr = reinterpret_cast<wxEvtHandler*>(userData);
wxSOCKET_T sock = wParam;
wxThreadEvent* event =
new wxThreadEvent(wxEVT_SOCKET_POLLER_RESULT);
event->SetPayload<wxSOCKET_T>(sock);
event->SetInt(pollResult);
if ( wxThread::IsMain() )
{
hndlr->ProcessEvent(*event);
delete event;
}
else
{
wxQueueEvent(hndlr, event);
}
}
return 0;
}
else
{
return DefWindowProc(hwnd, uMsg, wParam, lParam);
}
}
wxThread::ExitCode SocketPoller::Entry()
SocketPollerImpl* SocketPollerImpl::Create(wxEvtHandler* hndlr)
{
wxMessageQueueError er;
SocketPoller::Message msg;
bool done = false;
size_t socketsUnderConsideration = 0;
int timeOut = 50;
Message::MessageId id;
while ( !done )
{
if ( GetThread()->TestDestroy() )
{
break;
}
// Process all messages in the message queue. If we currently have no
// sockets, wait a little for a message to come in.
timeOut = (socketsUnderConsideration == 0) ? 50 : 0;
er = wxMSGQUEUE_NO_ERROR;
while ( er == wxMSGQUEUE_NO_ERROR )
{
er = m_msgQueue.ReceiveTimeout(timeOut, msg);
id = msg.GetMessageId();
if ( er == wxMSGQUEUE_TIMEOUT )
{
break;
}
else if ( er == wxMSGQUEUE_MISC_ERROR )
{
wxLogDebug("Error with socket poller message queue.");
done = true;
break;
}
else if ( id == Message::Quit )
{
done = true;
break;
}
else if ( id == Message::AddSocketAction )
{
wxSOCKET_T s = msg.GetSocket();
int f = msg.GetFlags();
wxEvtHandler* hndlr = msg.GetEvtHandler();
ThreadSetSocketAction(s, f, hndlr);
}
else if ( id == Message::DeleteSocketAction )
{
wxSOCKET_T s = msg.GetSocket();
ThreadDeleteSocketAction(s);
}
else if ( id == Message::ResumePolling )
{
wxSOCKET_T s = msg.GetSocket();
ThreadRemoveFromDL(s);
}
timeOut = 0;
}
socketsUnderConsideration = m_socketData.size() - m_disabledList.size();
if ( socketsUnderConsideration > 0 && !done )
{
ThreadCheckSockets();
}
}
return static_cast<wxThread::ExitCode>(0);
return new WinSock1SocketPoller(hndlr);
}
void SocketPoller::ThreadRemoveFromDL(wxSOCKET_T s)
{
SocketSet::iterator it = m_disabledList.find(s);
#else
if ( it != m_disabledList.end() )
// SocketPollerSourceHandler - a source handler used by the SocketPoller class.
class SocketPollerSourceHandler: public wxEventLoopSourceHandler
{
public:
SocketPollerSourceHandler(wxSOCKET_T, wxEvtHandler*);
void OnReadWaiting() wxOVERRIDE;
void OnWriteWaiting() wxOVERRIDE;
void OnExceptionWaiting() wxOVERRIDE;
~SocketPollerSourceHandler(){}
private:
void SendEvent(int);
wxSOCKET_T m_socket;
wxEvtHandler* m_handler;
};
SocketPollerSourceHandler::SocketPollerSourceHandler(wxSOCKET_T sock,
wxEvtHandler* hndlr)
{
m_socket = sock;
m_handler = hndlr;
}
void SocketPollerSourceHandler::OnReadWaiting()
{
SendEvent(SocketPoller::READY_FOR_READ);
}
void SocketPollerSourceHandler::OnWriteWaiting()
{
SendEvent(SocketPoller::READY_FOR_WRITE);
}
void SocketPollerSourceHandler::OnExceptionWaiting()
{
SendEvent(SocketPoller::HAS_ERROR);
}
void SocketPollerSourceHandler::SendEvent(int result)
{
wxThreadEvent event(wxEVT_SOCKET_POLLER_RESULT);
event.SetPayload<wxSOCKET_T>(m_socket);
event.SetInt(result);
m_handler->ProcessEvent(event);
}
// SourceSocketPoller - a SocketPollerImpl based on event loop sources.
class SourceSocketPoller: public SocketPollerImpl
{
public:
SourceSocketPoller(wxEvtHandler*);
~SourceSocketPoller();
bool StartPolling(wxSOCKET_T, int) wxOVERRIDE;
void StopPolling(wxSOCKET_T) wxOVERRIDE;
void ResumePolling(wxSOCKET_T) wxOVERRIDE;
private:
WX_DECLARE_HASH_MAP(wxSOCKET_T, wxEventLoopSource*, wxIntegerHash,\
wxIntegerEqual, SocketDataMap);
void CleanUpSocketSource(wxEventLoopSource*);
SocketDataMap m_socketData;
wxEvtHandler* m_handler;
};
SourceSocketPoller::SourceSocketPoller(wxEvtHandler* hndlr)
{
m_handler = hndlr;
}
SourceSocketPoller::~SourceSocketPoller()
{
// Clean up any leftover socket data.
for ( SocketDataMap::iterator it = m_socketData.begin() ;
it != m_socketData.end() ; ++it )
{
m_disabledList.erase(it);
CleanUpSocketSource(it->second);
}
}
void SocketPoller::ThreadDeleteSocketAction(wxSOCKET_T sock)
static int SocketPoller2EventSource(int pollAction)
{
ThreadRemoveFromDL(sock);
// Convert the SocketPoller::PollAction value to a flag that can be used
// by wxEventLoopSource.
// Always check for errors.
int eventSourceFlag = wxEVENT_SOURCE_EXCEPTION;
if ( pollAction & SocketPoller::POLL_FOR_READ )
{
eventSourceFlag |= wxEVENT_SOURCE_INPUT;
}
if ( pollAction & SocketPoller::POLL_FOR_WRITE )
{
eventSourceFlag |= wxEVENT_SOURCE_OUTPUT;
}
return eventSourceFlag;
}
bool SourceSocketPoller::StartPolling(wxSOCKET_T sock, int pollAction)
{
SocketDataMap::iterator it = m_socketData.find(sock);
wxEventLoopSourceHandler* srcHandler = NULL;
if ( it != m_socketData.end() )
{
// If this socket is already being polled, reuse the old handler. Also
// delete the old source object to stop the old polling operations.
wxEventLoopSource* oldSrc = it->second;
srcHandler = oldSrc->GetHandler();
delete oldSrc;
}
else
{
// Otherwise create a new source handler.
srcHandler =
new SocketPollerSourceHandler(sock, m_handler);
}
// Get a new source object for these polling checks.
bool socketIsPolled = true;
int eventSourceFlag = SocketPoller2EventSource(pollAction);
wxEventLoopSource* newSrc =
wxEventLoopBase::AddSourceForFD(sock, srcHandler, eventSourceFlag);
if ( newSrc == NULL )
{
// We were not able to add a source for this socket.
wxLogDebug(wxString::Format(
"Unable to create event loop source for %d",
static_cast<int>(sock)));
delete srcHandler;
socketIsPolled = false;
if ( it != m_socketData.end() )
{
m_socketData.erase(it);
}
}
else
{
m_socketData[sock] = newSrc;
}
return socketIsPolled;
}
void SourceSocketPoller::StopPolling(wxSOCKET_T sock)
{
SocketDataMap::iterator it = m_socketData.find(sock);
if ( it != m_socketData.end() )
{
CleanUpSocketSource(it->second);
m_socketData.erase(it);
}
}
void SocketPoller::ThreadSetSocketAction(wxSOCKET_T sock, int flags,
wxEvtHandler* hndlr)
void SourceSocketPoller::ResumePolling(wxSOCKET_T WXUNUSED(sock))
{
ThreadDeleteSocketAction(sock);
SocketData data;
data.m_pollAction = flags;
data.m_event = NULL;
data.m_handler = hndlr;
m_socketData[sock] = data;
}
void SocketPoller::ThreadCheckSockets()
void SourceSocketPoller::CleanUpSocketSource(wxEventLoopSource* source)
{
fd_set readFds, writeFds, errorFds;
FD_ZERO(&readFds);
FD_ZERO(&writeFds);
FD_ZERO(&errorFds);
wxSOCKET_T maxSd = 0;
if ( m_readEnd != SocketPoller::SOCKET_POLLER_INVALID_SOCKET )
{
FD_SET(m_readEnd, &readFds);
maxSd = m_readEnd;
}
for ( SocketDataMap::iterator it = m_socketData.begin() ;
it != m_socketData.end() ; ++it )
{
wxSOCKET_T sock = it->first;
if ( m_disabledList.find(sock) != m_disabledList.end() )
{
continue;
}
int checkAction = it->second.m_pollAction;
if ( checkAction & PollForRead )
{
FD_SET(sock, &readFds);
}
if ( checkAction & PollForWrite )
{
FD_SET(sock, &writeFds);
}
FD_SET(sock, &errorFds);
if ( sock > maxSd )
{
maxSd = sock;
}
}
struct timeval timeout;
timeout.tv_sec = 0; // 1s timeout
timeout.tv_usec = 50*1000;
int selectStatus = select(maxSd+1, &readFds, &writeFds, &errorFds,&timeout);
if ( selectStatus < 0 )
{
// Massive error: do something
}
else if ( selectStatus == 0 )
{
;// select timed out. There is no need to do anything.
}
else
{
if ( (m_readEnd != SocketPoller::SOCKET_POLLER_INVALID_SOCKET)
&& FD_ISSET(m_readEnd, &readFds) )
{
char c;
recv(m_readEnd, &c, 1, 0);
}
for ( SocketDataMap::iterator it = m_socketData.begin() ;
it != m_socketData.end() ; ++it )
{
wxSOCKET_T sock = it->first;
if ( m_disabledList.find(sock) != m_disabledList.end() )
{
continue;
}
int checkActions = it->second.m_pollAction;
wxEvtHandler* hndlr = it->second.m_handler;
int result = InvalidResult;
if ( checkActions & PollForRead )
{
if ( FD_ISSET(sock, &readFds) )
{
result |= ReadyForRead;
}
}
if ( checkActions & ReadyForWrite )
{
if ( FD_ISSET(sock, &writeFds) )
{
result |= ReadyForWrite;
}
}
if ( FD_ISSET(sock, &errorFds) )
{
result |= HasError;
}
if ( result != InvalidResult && hndlr != NULL )
{
wxThreadEvent* event = new wxThreadEvent(wxSocketAction);
event->SetPayload<wxSOCKET_T>(sock);
event->SetInt(result);
wxQueueEvent(hndlr,event);
m_disabledList.insert(sock);
}
}
}
wxEventLoopSourceHandler* srcHandler = source->GetHandler();
delete source;
delete srcHandler;
}
SocketPollerImpl* SocketPollerImpl::Create(wxEvtHandler* hndlr)
{
return new SourceSocketPoller(hndlr);
}
#endif
//
// wxWebSessionCURL
@@ -946,10 +892,11 @@ wxWebSessionCURL::wxWebSessionCURL() :
ms_activeSessions++;
m_socketPoller = new SocketPoller();
m_socketPoller = new SocketPoller(this);
m_timeoutTimer.SetOwner(this);
Bind(wxEVT_TIMER, &wxWebSessionCURL::TimeoutNotification, this);
Bind(wxSocketAction, &wxWebSessionCURL::ProcessSocketPollerResult, this);
Bind(wxEVT_SOCKET_POLLER_RESULT,
&wxWebSessionCURL::ProcessSocketPollerResult, this);
}
wxWebSessionCURL::~wxWebSessionCURL()
@@ -1139,19 +1086,20 @@ void wxWebSessionCURL::ProcessTimeoutNotification()
static int CurlPoll2SocketPoller(int what)
{
int pollAction = SocketPoller::InvalidAction;
int pollAction = SocketPoller::INVALID_ACTION;
if ( what == CURL_POLL_IN )
{
pollAction = SocketPoller::PollForRead ;
pollAction = SocketPoller::POLL_FOR_READ ;
}
else if ( what == CURL_POLL_OUT )
{
pollAction = SocketPoller::PollForWrite;
pollAction = SocketPoller::POLL_FOR_WRITE;
}
else if ( what == CURL_POLL_INOUT )
{
pollAction = SocketPoller::PollForRead | SocketPoller::PollForWrite;
pollAction =
SocketPoller::POLL_FOR_READ | SocketPoller::POLL_FOR_WRITE;
}
return pollAction;
@@ -1169,7 +1117,7 @@ void wxWebSessionCURL::ProcessSocketCallback(curl_socket_t s, int what)
case CURL_POLL_OUT:
wxFALLTHROUGH;
case CURL_POLL_INOUT:
m_socketPoller->StartPolling(s, CurlPoll2SocketPoller(what), this);
m_socketPoller->StartPolling(s, CurlPoll2SocketPoller(what));
break;
case CURL_POLL_REMOVE:
m_socketPoller->StopPolling(s);
@@ -1184,17 +1132,17 @@ static int SocketPollerResult2CurlSelect(int socketEventFlag)
{
int curlSelect = 0;
if ( socketEventFlag & SocketPoller::ReadyForRead )
if ( socketEventFlag & SocketPoller::READY_FOR_READ )
{
curlSelect |= CURL_CSELECT_IN;
}
if ( socketEventFlag & SocketPoller::ReadyForWrite )
if ( socketEventFlag & SocketPoller::READY_FOR_WRITE )
{
curlSelect |= CURL_CSELECT_OUT;
}
if ( socketEventFlag & SocketPoller::HasError )
if ( socketEventFlag & SocketPoller::HAS_ERROR )
{
curlSelect |= CURL_CSELECT_ERR;
}