Network - Send/Recv both ways.

Impl: SetBlockingMode() : Connection.
Impl: Resize() : OysterByte.
Remade Clear() : OysterByte.
Added bool return if FetchMessage() succeded or failed : PostBox.
Fixed a few errors with mutex on ThreadedClient.
Added missing mutex unlock in ThreadSafeQueue.
This commit is contained in:
Pontus Fransson 2013-12-03 23:12:48 +01:00
parent e837b80e43
commit df1470eb2c
11 changed files with 188 additions and 142 deletions

View File

@ -132,6 +132,7 @@ namespace Oyster
this->front = NULL;
this->back = NULL;
}
mutex.UnlockMutex();
return item;
}
@ -146,6 +147,7 @@ namespace Oyster
mutex.LockMutex();
if(front != NULL)
{
mutex.UnlockMutex();
return this->front->item;
}
mutex.UnlockMutex();

View File

@ -96,7 +96,7 @@ int Connection::Recieve(OysterByte& bytes)
{
int nBytes;
bytes.Clear(1000);
bytes.Resize(1000);
nBytes = recv(this->socket, bytes, 500, 0);
if(nBytes == SOCKET_ERROR)
{
@ -109,8 +109,6 @@ int Connection::Recieve(OysterByte& bytes)
std::cout << "Size of the recieved data: " << nBytes << " bytes" << std::endl;
//bytes.byteArray[nBytes] = '\0';
return 0;
}
@ -139,15 +137,24 @@ int Connection::InitiateSocket()
return 0;
}
void Connection::SetBlockingMode(bool blocking)
int Connection::SetBlockingMode(bool blocking)
{
//TODO: Implement this function. Setting the socket to blocking or non-blocking.
DWORD nonBlocking;
if(blocking)
{
//fcntl(this->socket, F_SETFL, O_NONBLOCK);
nonBlocking = 0;
}
else
{
nonBlocking = 1;
}
int result = ioctlsocket(this->socket, FIONBIO, &nonBlocking);
if(result != 0)
{
return WSAGetLastError();
}
return 0;
}

View File

@ -20,7 +20,6 @@ namespace Oyster
Connection( int socket ) { this->socket = socket; };
virtual ~Connection();
virtual int InitiateServer( unsigned short port );
virtual int InitiateClient();
@ -32,9 +31,11 @@ namespace Oyster
virtual int Listen();
//Setting the socket to blocking/non-blocking mode.
int SetBlockingMode( bool blocking );
private:
int InitiateSocket();
void SetBlockingMode( bool blocking );
int socket;

View File

@ -15,7 +15,7 @@ namespace Oyster
public:
virtual ~IPostBox() {}
virtual void PostMessage(T& message) = 0;
virtual void FetchMessage(T& message) = 0;
virtual bool FetchMessage(T& message) = 0;
virtual bool IsFull() = 0;
};

View File

@ -21,11 +21,18 @@ OysterByte::~OysterByte()
delete[] byteArray;
}
void OysterByte::Clear(unsigned int cap)
void OysterByte::Clear()
{
size = 0;
}
void OysterByte::Resize(unsigned int cap)
{
if(capacity < cap)
{
delete[] byteArray;
byteArray = new unsigned char[cap];
size = 0;
}
}
int OysterByte::GetSize()

View File

@ -18,7 +18,8 @@ namespace Oyster
OysterByte(int cap);
virtual ~OysterByte();
void Clear(unsigned int cap);
void Clear(); //Resets size to 0
void Resize(unsigned int cap); //Resizes the array with, it does not keep anything in it.
int GetSize();
unsigned char* GetByteArray();

View File

@ -23,7 +23,7 @@ namespace Oyster
virtual ~PostBox();
virtual void PostMessage(T& message);
virtual void FetchMessage(T& message);
virtual bool FetchMessage(T& message);
virtual bool IsFull();
private:
@ -49,13 +49,15 @@ namespace Oyster
}
template <class T>
void PostBox<T>::FetchMessage(T& message)
bool PostBox<T>::FetchMessage(T& message)
{
if(IsFull())
{
message = messages.Front();
messages.Pop();
return true;
}
return false;
}
template <class T>

View File

@ -4,7 +4,6 @@
using namespace Oyster::Network;
using namespace Oyster::Thread;
ThreadedClient::ThreadedClient()
{
this->connection = new Connection();
@ -17,6 +16,21 @@ ThreadedClient::ThreadedClient(unsigned int socket)
this->connection = new Connection(socket);
this->sendPostBox = new PostBox<OysterByte*>();
this->recvPostBox = NULL;
connection->SetBlockingMode(false);
thread.Create(this, true);
}
ThreadedClient::ThreadedClient(IPostBox<OysterByte*>* postBox, unsigned int socket)
{
this->connection = new Connection(socket);
this->sendPostBox = new PostBox<OysterByte*>;
this->recvPostBox = postBox;
connection->SetBlockingMode(false);
thread.Create(this, true);
}
ThreadedClient::~ThreadedClient()
@ -35,7 +49,9 @@ ThreadedClient::~ThreadedClient()
int ThreadedClient::Send(OysterByte* byte)
{
mutex.LockMutex();
this->sendPostBox->PostMessage(byte);
mutex.UnlockMutex();
return 0;
}
@ -43,13 +59,13 @@ int ThreadedClient::Send()
{
int errorCode = 0;
mutex.LockMutex();
if(!sendPostBox->IsFull())
if(sendPostBox->IsFull())
{
OysterByte *temp = NULL;
sendPostBox->FetchMessage(temp);
errorCode = this->connection->Send(*temp);
mutex.UnlockMutex();
}
mutex.UnlockMutex();
return errorCode;
}
@ -57,42 +73,48 @@ int ThreadedClient::Send()
int ThreadedClient::Recv()
{
int errorCode = 0;
mutex.LockMutex();
if(!recvPostBox->IsFull())
{
OysterByte *temp = NULL;
OysterByte *temp = new OysterByte();
errorCode = this->connection->Recieve(*temp);
if(errorCode == 0)
{
mutex.LockMutex();
recvPostBox->PostMessage(temp);
mutex.UnlockMutex();
}
else
{
delete temp;
}
return errorCode;
}
void ThreadedClient::ThreadEntry()
{
std::cout<< "Thread started" << std::endl;
std::cout<< "Client Thread started" << std::endl;
}
void ThreadedClient::ThreadExit()
{
std::cout << "Thread exit" << std::endl;
std::cout << "Client Thread exit" << std::endl;
}
bool ThreadedClient::DoWork()
{
int errorCode;
errorCode = Send();
if(errorCode != 0)
/*if(errorCode != 0)
{
return false;
}
}*/
errorCode = Recv();
if(errorCode != 0)
/*if(errorCode != 0)
{
return false;
}
}*/
return true;
}

View File

@ -20,6 +20,7 @@ namespace Oyster
public:
ThreadedClient();
ThreadedClient(unsigned int socket);
ThreadedClient(IPostBox<OysterByte*>* postBox, unsigned int socket);
virtual ~ThreadedClient();
int Send(OysterByte* byte);
@ -27,8 +28,8 @@ namespace Oyster
int Connect(unsigned short port, const char serverName[]);
void setRecvPostBox(IPostBox<OysterByte*>* postBox);
private:
private:
virtual int Send();
virtual int Recv();
@ -36,9 +37,7 @@ namespace Oyster
virtual void ThreadExit();
virtual bool DoWork();
private:
Connection* connection;
IPostBox<OysterByte*>* sendPostBox;
IPostBox<OysterByte*>* recvPostBox;

View File

@ -7,14 +7,18 @@
#include "../NetworkDependencies/OysterByte.h"
#include "../../Misc/ThreadSafeQueue.h"
#include "../NetworkDependencies/ThreadedClient.h"
#include "../../Misc/WinTimer.h"
#include "../../Misc/Utilities.h"
#pragma comment(lib, "ws2_32.lib")
using namespace std;
using namespace Oyster::Network::Protocols;
using namespace Oyster::Network;
using namespace Utility;
void chat(ThreadedClient &client);
void PrintOutMessage(ProtocolSet* set);
int main()
{
@ -27,10 +31,10 @@ int main()
cout << "Client" << endl;
//Create Client
ThreadedClient client;
ThreadedClient* client = new ThreadedClient;
//Connect to server
errorCode = client.Connect(9876, "localhost");
errorCode = client->Connect(9876, "localhost");
if(errorCode != 0)
{
@ -38,7 +42,8 @@ int main()
wcout << "errorMessage: " << errorTest << endl;
}
//chat(client);
chat(*client);
delete client;
ShutdownWinSock();
@ -49,29 +54,57 @@ int main()
void chat(ThreadedClient &client)
{
Oyster::Network::Translator *t = new Oyster::Network::Translator();
IPostBox<OysterByte*>* postBox = new PostBox<OysterByte*>;
Oyster::Network::OysterByte msgRecv;
string msgSend = "";
client.setRecvPostBox(postBox);
Oyster::Network::OysterByte* msgRecv = NULL;
Oyster::Network::OysterByte* msgSend = new OysterByte();
ProtocolSet* set = new ProtocolSet;
ProtocolPlayerPos test;
test.ID = 5;
test.matrix = new float[16];
test.nrOfFloats = 5;
test.matrix = new float[test.nrOfFloats];
float temp = 10;
for(int i = 0; i < 5; i++)
for(int i = 0; i < (int)test.nrOfFloats; i++)
{
test.matrix[i] = temp;
temp++;
}
t->Pack(test, *msgSend);
bool chatDone = false;
/*
while(!chatDone)
WinTimer timer;
while(1)
{
client.Recv(msgRecv);
//Fetch new messages from the postbox
if(postBox->FetchMessage(msgRecv))
{
t->Unpack(set, *msgRecv);
delete msgRecv;
t->Unpack(set, msgRecv);
PrintOutMessage(set);
set->Release();
}
//Send message to server each second
if(timer.getElapsedSeconds() > 1)
{
cout << "Sending to server." << endl;
timer.reset();
client.Send(msgSend);
}
}
delete msgSend;
delete postBox;
delete t;
delete set;
}
void PrintOutMessage(ProtocolSet* set)
{
switch(set->type)
{
case PackageType_header:
@ -84,47 +117,14 @@ void chat(ThreadedClient &client)
}
cout << endl;
break;
case PackageType_player_pos:
cout << "Server: ID " << set->Protocol.pPlayerPos->ID << endl;
for(int i = 0; i < set->Protocol.pPlayerPos->nrOfFloats; i++)
cout << "ID " << set->Protocol.pPlayerPos->ID << endl;
for(int i = 0; i < (int)set->Protocol.pPlayerPos->nrOfFloats; i++)
{
cout << set->Protocol.pPlayerPos->matrix[i] << ' ';
}
cout << endl;
break;
}
set->Release();
msgRecv.Clear(1000);
/*std::getline(std::cin, msgSend);
if( msgSend != "exit")
{
if(msgSend.length() < 1)
{
msgSend = "ERROR!";
}
test.textMessage = msgSend;
t->Pack(test, msgRecv);
client.Send(msgRecv);
}
else
{
chatDone = true;
}
cin.clear();*/
/*
}
delete t;
delete set;
*/
}

View File

@ -18,10 +18,15 @@ using namespace Oyster::Network;
using namespace ::Protocols;
using namespace Utility;
void PrintOutMessage(ProtocolSet* set);
int main()
{
OysterByte recvBuffer;
OysterByte sendBuffer;
OysterByte* recvBuffer = NULL;
ProtocolSet* set = new ProtocolSet;
IPostBox<int>* postBox = new PostBox<int>();
IPostBox<OysterByte*>* recvPostBox = new PostBox<OysterByte*>();
cout << "Server" << endl;
Translator t;
@ -43,7 +48,7 @@ int main()
ProtocolPlayerPos test;
test.clientID = 0;
test.ID = 5;
test.nrOfFloats = 16;
test.nrOfFloats = 10;
test.matrix = new float[test.nrOfFloats];
for(int i = 0; i < (int)test.nrOfFloats; i++)
@ -51,7 +56,7 @@ int main()
test.matrix[i] = (float)i;
}
t.Pack(test, recvBuffer);
t.Pack(test, sendBuffer);
WinTimer timer;
@ -59,14 +64,13 @@ int main()
int client = -1;
while(1)
{
client = -1;
postBox->FetchMessage(client);
if(client != -1)
//Fetch new clients from the postbox
if(postBox->FetchMessage(client))
{
cout << "Client connected: " << client << endl;
clients.push_back(new ThreadedClient(client));
clients.push_back(new ThreadedClient(recvPostBox, client));
clients.at(clients.size()-1)->Send(&recvBuffer);
clients.at(clients.size()-1)->Send(&sendBuffer);
}
//Send a message every 1 secounds to all clients.
@ -76,51 +80,52 @@ int main()
timer.reset();
for(int i = 0; i < (int)clients.size(); i++)
{
clients.at(i)->Send(&recvBuffer);
clients.at(i)->Send(&sendBuffer);
}
}
Sleep(100);
//Fetch messages
if(recvPostBox->FetchMessage(recvBuffer))
{
t.Unpack(set, *recvBuffer);
delete recvBuffer;
PrintOutMessage(set);
set->Release();
}
Sleep(1);
}
listener.Shutdown();
/*
ProtocolSet* set = new ProtocolSet;
client1.Send(recvBuffer);
while(1)
{
client1.Recv(recvBuffer);
t.Unpack(set, recvBuffer);
cout << set->Protocol.pTest->clientID << ' ' << set->Protocol.pTest->packageType << ' ' << set->Protocol.pTest->size << endl;
cout << "Client1: " << set->Protocol.pTest->textMessage << endl;
for(int i = 0; i < (int)set->Protocol.pTest->numOfFloats; i++)
{
cout << set->Protocol.pTest->f[i] << ' ';
}
cout << endl;
set->Release();
client2.Send(recvBuffer);
client2.Recv(recvBuffer);
t.Unpack(set, recvBuffer);
cout << set->Protocol.pTest->clientID << ' ' << set->Protocol.pTest->packageType << ' ' << set->Protocol.pTest->size << endl;
cout << "Client2: " << set->Protocol.pTest->textMessage << endl;
for(int i = 0; i < (int)set->Protocol.pTest->numOfFloats; i++)
{
cout << set->Protocol.pTest->f[i] << ' ';
}
cout << endl;
set->Release();
client1.Send(recvBuffer);
}
ShutdownWinSock();
delete set;
*/
system("pause");
delete postBox;
return 0;
}
void PrintOutMessage(ProtocolSet* set)
{
switch(set->type)
{
case PackageType_header:
break;
case PackageType_test:
cout <<"Client 2: " << set->Protocol.pTest->textMessage <<endl;
for(int i = 0; i < (int)set->Protocol.pTest->numOfFloats; i++)
{
cout << set->Protocol.pTest->f[i] << ' ' ;
}
cout << endl;
break;
case PackageType_player_pos:
cout << "ID " << set->Protocol.pPlayerPos->ID << endl;
for(int i = 0; i < (int)set->Protocol.pPlayerPos->nrOfFloats; i++)
{
cout << set->Protocol.pPlayerPos->matrix[i] << ' ';
}
cout << endl;
break;
}
}