sco/pokerth
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Added helper class for receiving data. Modified sender thread to support sending partial packets. Prepared abstractions for server state machine.

Browse Source
This commit is contained in:
lotodore
2007-03-13 02:23:12 +00:00
parent 29b29ec42e
commit f5289a6fcf
24 changed files with 508 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/net/common/clientexception.cpp
+26
View File
@@ -0,0 +1,26 @@
/***************************************************************************
* Copyright (C) 2007 by Lothar May *
* *
* This program is free software; you can redistribute it and/or modify *
* it under the terms of the GNU General Public License as published by *
* the Free Software Foundation; either version 2 of the License, or *
* (at your option) any later version. *
* *
* This program is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
* GNU General Public License for more details. *
* *
* You should have received a copy of the GNU General Public License *
* along with this program; if not, write to the *
* Free Software Foundation, Inc., *
* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
***************************************************************************/
#include <net/clientexception.h>
ClientException::~ClientException()
{
}
src/net/common/clientstate.cpp
+10 -17
View File
@@ -21,6 +21,7 @@
#include <net/clientthread.h>
#include <net/clientdata.h>
#include <net/senderthread.h>
#include <net/receiverhelper.h>
#include <net/netpacket.h>
#include <net/resolverthread.h>
#include <net/clientexception.h>
@@ -322,6 +323,7 @@ ClientStateStartSession::~ClientStateStartSession()
int
ClientStateStartSession::Process(ClientThread &client)
{
client.GetReceiver().Init(client.GetData().sockfd);
client.GetSender().Init(client.GetData().sockfd);
client.GetSender().Run();
@@ -356,29 +358,20 @@ ClientStateWaitSession::Process(ClientThread &client)
int retVal;
ClientData &data = client.GetData();
// TODO: use receiver thread.
fd_set readSet;
struct timeval timeout;
// delegate to receiver helper class
FD_ZERO(&readSet);
FD_SET(data.sockfd, &readSet);
boost::shared_ptr<NetPacket> tmpPacket = client.GetReceiver().Recv();
timeout.tv_sec = 0;
timeout.tv_usec = CLIENT_WAIT_TIMEOUT_MSEC * 1000;
int selectResult = select(data.sockfd + 1, &readSet, NULL, NULL, &timeout);
if (selectResult > 0) // recv is possible
if (tmpPacket.get())
{
char buf[128];
if (recv(data.sockfd, buf, sizeof(buf), 0) > 0)
{
client.SetState(ClientStateFinal::Instance());
retVal = MSG_SOCK_SESSION_DONE;
}
else
throw ClientException(ERR_SOCK_RECV_FAILED, 0);
client.SetState(ClientStateFinal::Instance());
retVal = MSG_SOCK_SESSION_DONE;
}
else
{
retVal = MSG_SOCK_INTERNAL_PENDING;
Thread::Msleep(CLIENT_WAIT_TIMEOUT_MSEC);
}
return retVal;
}
src/net/common/clientthread.cpp
+13 -4
View File
@@ -21,6 +21,7 @@
#include <net/clientstate.h>
#include <net/clientdata.h>
#include <net/senderthread.h>
#include <net/receiverhelper.h>
#include <net/clientcallback.h>
#include <net/clientexception.h>
#include <net/socket_msg.h>
@@ -58,18 +59,19 @@ ClientThread::Init(const string &serverAddress, unsigned serverPort, bool ipv6,
void
ClientThread::Main()
{
m_sender.reset(new SenderThread);
m_sender.reset(new SenderThread(m_callback));
m_receiver.reset(new ReceiverHelper);
SetState(CLIENT_INITIAL_STATE::Instance());
try {
while (!ShouldTerminate())
{
int msg = GetState().Process(*this);
if (msg != MSG_SOCK_INTERNAL_PENDING)
m_callback.SignalNetClientSuccess(msg);
m_callback.SignalNetSuccess(msg);
}
} catch (const ClientException &e)
} catch (const NetException &e)
{
m_callback.SignalNetClientError(e.GetErrorId(), e.GetOsErrorCode());
m_callback.SignalNetError(e.GetErrorId(), e.GetOsErrorCode());
}
GetSender().SignalTermination();
GetSender().Join(100);
@@ -109,3 +111,10 @@ ClientThread::GetSender()
return *m_sender;
}
ReceiverHelper &
ClientThread::GetReceiver()
{
assert(m_receiver.get());
return *m_receiver;
}
src/net/common/netcallback.cpp
+26
View File
@@ -0,0 +1,26 @@
/***************************************************************************
* Copyright (C) 2007 by Lothar May *
* *
* This program is free software; you can redistribute it and/or modify *
* it under the terms of the GNU General Public License as published by *
* the Free Software Foundation; either version 2 of the License, or *
* (at your option) any later version. *
* *
* This program is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
* GNU General Public License for more details. *
* *
* You should have received a copy of the GNU General Public License *
* along with this program; if not, write to the *
* Free Software Foundation, Inc., *
* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
***************************************************************************/
#include <net/netcallback.h>
NetCallback::~NetCallback()
{
}
src/net/common/netexception.cpp
+26
View File
@@ -0,0 +1,26 @@
/***************************************************************************
* Copyright (C) 2007 by Lothar May *
* *
* This program is free software; you can redistribute it and/or modify *
* it under the terms of the GNU General Public License as published by *
* the Free Software Foundation; either version 2 of the License, or *
* (at your option) any later version. *
* *
* This program is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
* GNU General Public License for more details. *
* *
* You should have received a copy of the GNU General Public License *
* along with this program; if not, write to the *
* Free Software Foundation, Inc., *
* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
***************************************************************************/
#include <net/netexception.h>
NetException::~NetException()
{
}
src/net/common/netpacket.cpp
+22 -3
View File
@@ -18,6 +18,8 @@
***************************************************************************/
#include <net/netpacket.h>
#include <net/netexception.h>
#include <net/socket_msg.h>
NetPacket::~NetPacket()
{
@@ -25,11 +27,15 @@ NetPacket::~NetPacket()
//-----------------------------------------------------------------------------
TestNetPacket::TestNetPacket()
{
}
TestNetPacket::TestNetPacket(u_int32_t value)
{
m_data.head.type = 0;
m_data.head.length = sizeof(m_data);
m_data.test = value;
m_data.head.type = htons(NET_TYPE_TEST);
m_data.head.length = htons(sizeof(m_data));
m_data.test = htonl(value);
}
TestNetPacket::~TestNetPacket()
@@ -42,3 +48,16 @@ TestNetPacket::GetData()
return (NetPacketHeader *)&m_data;
}
void
TestNetPacket::SetData(const NetPacketHeader *p)
{
u_int16_t tmpLen = ntohs(p->length);
if (tmpLen != sizeof(m_data)
|| ntohs(p->type) != NET_TYPE_TEST)
{
throw NetException(ERR_SOCK_INTERNAL, 0);
}
memcpy(&m_data, p, tmpLen);
}
src/net/common/receiverhelper.cpp
+142
View File
@@ -0,0 +1,142 @@
/***************************************************************************
* Copyright (C) 2007 by Lothar May *
* *
* This program is free software; you can redistribute it and/or modify *
* it under the terms of the GNU General Public License as published by *
* the Free Software Foundation; either version 2 of the License, or *
* (at your option) any later version. *
* *
* This program is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
* GNU General Public License for more details. *
* *
* You should have received a copy of the GNU General Public License *
* along with this program; if not, write to the *
* Free Software Foundation, Inc., *
* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
***************************************************************************/
#include <net/receiverhelper.h>
#include <net/socket_msg.h>
#include <net/netexception.h>
#include <cstring>
using namespace std;
#define RECV_TIMEOUT_MSEC 50
ReceiverHelper::ReceiverHelper()
: m_socket(INVALID_SOCKET), m_tmpInBufSize(0)
{
}
ReceiverHelper::~ReceiverHelper()
{
}
void
ReceiverHelper::Init(SOCKET socket)
{
if (!IS_VALID_SOCKET(socket))
return; // TODO: throw exception
m_socket = socket;
}
boost::shared_ptr<NetPacket>
ReceiverHelper::Recv()
{
boost::shared_ptr<NetPacket> tmpPacket(InternalGetPacket());
if (!tmpPacket.get())
{
unsigned bufSize = RECV_BUF_SIZE - m_tmpInBufSize;
if (bufSize) // check if there is room in the input buffer
{
fd_set readSet;
struct timeval timeout;
FD_ZERO(&readSet);
FD_SET(m_socket, &readSet);
timeout.tv_sec = 0;
timeout.tv_usec = RECV_TIMEOUT_MSEC * 1000;
int selectResult = select(m_socket + 1, &readSet, NULL, NULL, &timeout);
if (!IS_VALID_SELECT(selectResult))
{
throw NetException(ERR_SOCK_SELECT_FAILED, SOCKET_ERRNO());
}
if (selectResult > 0) // recv is possible
{
int bytesRecvd = recv(m_socket, m_tmpInBuf + m_tmpInBufSize, bufSize, 0);
if (!IS_VALID_RECV(bytesRecvd))
{
throw NetException(ERR_SOCK_RECV_FAILED, SOCKET_ERRNO());
}
else if (bytesRecvd == 0)
{
throw NetException(ERR_SOCK_CONN_RESET, 0);
}
else
{
m_tmpInBufSize += bytesRecvd;
tmpPacket = InternalGetPacket();
}
}
}
}
return tmpPacket;
}
boost::shared_ptr<NetPacket>
ReceiverHelper::InternalGetPacket()
{
boost::shared_ptr<NetPacket> tmpPacket;
// This is necessary, because we use TCP.
// Packets may be received in multiple chunks or
// several packets may be received at once.
if (m_tmpInBufSize >= sizeof(NetPacketHeader))
{
NetPacketHeader *tmpHeader = (NetPacketHeader *)m_tmpInBuf;
u_int16_t tmpLen = ntohs(tmpHeader->length);
if (tmpLen < sizeof(NetPacketHeader)
|| tmpLen > MAX_PACKET_SIZE)
{
// Invalid packet - reset input buffer.
m_tmpInBufSize = 0;
}
else if (m_tmpInBufSize >= tmpLen)
{
tmpPacket = InternalCreateNetPacket(tmpHeader);
m_tmpInBufSize -= tmpLen;
}
}
return tmpPacket;
}
boost::shared_ptr<NetPacket>
ReceiverHelper::InternalCreateNetPacket(const NetPacketHeader *p)
{
boost::shared_ptr<NetPacket> tmpPacket;
switch(ntohs(p->type))
{
case NET_TYPE_TEST:
try
{
tmpPacket = boost::shared_ptr<NetPacket>(new TestNetPacket);
tmpPacket->SetData(p);
} catch (const NetException &)
{
tmpPacket.reset();
}
break;
}
return tmpPacket;
}
src/net/common/senderthread.cpp
+52 -11
View File
@@ -18,11 +18,16 @@
***************************************************************************/
#include <net/senderthread.h>
#include <net/netpacket.h>
#include <net/netcallback.h>
#include <net/socket_msg.h>
#include <cstring>
using namespace std;
#define SEND_TIMEOUT_MSEC 50
SenderThread::SenderThread()
SenderThread::SenderThread(NetCallback &cb)
: m_tmpOutBufSize(0), m_callback(cb)
{
}
@@ -49,19 +54,33 @@ SenderThread::Send(boost::shared_ptr<NetPacket> packet)
void
SenderThread::Main()
{
boost::shared_ptr<NetPacket> tmpPacket;
while (!ShouldTerminate())
{
if (!tmpPacket.get())
// Send remaining bytes of output buffer OR
// copy ONE packet to output buffer.
// For reasons of simplicity, only one packet is sent at a time.
if (!m_tmpOutBufSize)
{
boost::mutex::scoped_lock lock(m_outBufMutex);
if (!m_outBuf.empty())
boost::shared_ptr<NetPacket> tmpPacket;
{
tmpPacket = m_outBuf.front();
m_outBuf.pop_front();
boost::mutex::scoped_lock lock(m_outBufMutex);
if (!m_outBuf.empty())
{
tmpPacket = m_outBuf.front();
m_outBuf.pop_front();
}
}
if (tmpPacket.get())
{
u_int16_t tmpLen = ntohs(tmpPacket->GetData()->length);
if (tmpLen <= MAX_PACKET_SIZE)
{
m_tmpOutBufSize = tmpLen;
memcpy(m_tmpOutBuf, tmpPacket->GetData(), m_tmpOutBufSize);
}
}
}
if (tmpPacket.get())
if (m_tmpOutBufSize)
{
fd_set writeSet;
struct timeval timeout;
@@ -72,10 +91,32 @@ SenderThread::Main()
timeout.tv_sec = 0;
timeout.tv_usec = SEND_TIMEOUT_MSEC * 1000;
int selectResult = select(m_socket + 1, NULL, &writeSet, NULL, &timeout);
if (!IS_VALID_SELECT(selectResult))
{
m_callback.SignalNetError(ERR_SOCK_SELECT_FAILED, SOCKET_ERRNO());
// Assume that this is a fatal error, terminate thread.
return;
}
if (selectResult > 0) // send is possible
{
send(m_socket, (const char *)tmpPacket->GetData(), tmpPacket->GetData()->length, 0);
tmpPacket.reset();
// send next chunk of data
int bytesSent = send(m_socket, m_tmpOutBuf, m_tmpOutBufSize, 0);
if (!IS_VALID_SEND(bytesSent))
{
m_callback.SignalNetError(ERR_SOCK_SEND_FAILED, SOCKET_ERRNO());
// Assume that this is a fatal error, terminate thread.
return;
}
else if ((unsigned)bytesSent < m_tmpOutBufSize)
{
m_tmpOutBufSize = m_tmpOutBufSize - (unsigned)bytesSent;
memmove(m_tmpOutBuf, m_tmpOutBuf + bytesSent, m_tmpOutBufSize);
}
else
{
m_tmpOutBufSize = 0;
}
}
}
else