result queue of class Connection replaced with notification functionality of class NetMessage.

This commit is contained in:
Roland Jax
2014-11-19 18:55:52 +01:00
parent a5896e53ca
commit 5b14d6dbec
4 changed files with 64 additions and 30 deletions
+2 -4
View File
@@ -85,10 +85,8 @@ void BaseLoop::start()
// send result to client // send result to client
result += '\n'; result += '\n';
Connection* connection = message->getConnection(); message->setResult(result);
connection->addResult(NetMessage(result)); message->sendSignal();
delete message;
// stop daemon // stop daemon
if (strcasecmp(data.c_str(), "STOP") == 0) if (strcasecmp(data.c_str(), "STOP") == 0)
+4 -11
View File
@@ -26,12 +26,6 @@ extern Logger& L;
int Connection::m_sum = 0; int Connection::m_sum = 0;
void Connection::addResult(NetMessage message)
{
NetMessage* tmp = new NetMessage(NetMessage(message));
m_netQueueResult.add(tmp);
}
void* Connection::run() void* Connection::run()
{ {
m_running = true; m_running = true;
@@ -80,22 +74,21 @@ void* Connection::run()
// send data // send data
data[datalen] = '\0'; data[datalen] = '\0';
m_netQueueData->add(new NetMessage(data, this)); NetMessage message(data);
m_netQueue->add(&message);
// wait for result // wait for result
L.log(net, debug, "[%05d] wait for result", getID()); L.log(net, debug, "[%05d] wait for result", getID());
NetMessage* message = m_netQueueResult.remove(); message.waitSignal();
L.log(net, debug, "[%05d] result added", getID()); L.log(net, debug, "[%05d] result added", getID());
std::string result(message->getData()); std::string result = message.getResult();
if (m_socket->isValid() == true) if (m_socket->isValid() == true)
m_socket->send(result.c_str(), result.size()); m_socket->send(result.c_str(), result.size());
else else
break; break;
delete message;
} }
} }
+2 -5
View File
@@ -31,9 +31,7 @@ class Connection : public Thread
public: public:
Connection(TCPSocket* socket, WQueue<NetMessage*>* netQueue) Connection(TCPSocket* socket, WQueue<NetMessage*>* netQueue)
: m_socket(socket), m_netQueueData(netQueue), m_running(false) { m_sum++; m_id = m_sum;} : m_socket(socket), m_netQueue(netQueue), m_running(false) { m_sum++; m_id = m_sum;}
void addResult(NetMessage message);
void* run(); void* run();
void stop() const { m_notify.notify(); } void stop() const { m_notify.notify(); }
@@ -43,8 +41,7 @@ public:
private: private:
TCPSocket* m_socket; TCPSocket* m_socket;
WQueue<NetMessage*>* m_netQueueData; WQueue<NetMessage*>* m_netQueue;
WQueue<NetMessage*> m_netQueueResult;
Notify m_notify; Notify m_notify;
bool m_running; bool m_running;
int m_id; int m_id;
+56 -10
View File
@@ -35,16 +35,27 @@ public:
/** /**
* @brief constructs a new instance with message and source client address. * @brief constructs a new instance with message and source client address.
* @param data from client. * @param data from client.
* @param connection to return result to correct client.
*/ */
NetMessage(const std::string data, Connection* connection=NULL) NetMessage(const std::string data) : m_data(data)
: m_data(data), m_connection(connection) {} {
pthread_mutex_init(&m_mutex, NULL);
pthread_cond_init(&m_cond, NULL);
}
/**
* @brief destructor.
*/
~NetMessage()
{
pthread_mutex_destroy(&m_mutex);
pthread_cond_destroy(&m_cond);
}
/** /**
* @brief copy constructor. * @brief copy constructor.
* @param src message object for copy. * @param src message object for copy.
*/ */
NetMessage(const NetMessage& src) : m_data(src.m_data), m_connection(src.m_connection) {} NetMessage(const NetMessage& src) : m_data(src.m_data) {}
/** /**
* @brief get the data string. * @brief get the data string.
@@ -53,17 +64,52 @@ public:
std::string getData() const { return m_data; } std::string getData() const { return m_data; }
/** /**
* @brief original connection. * @brief get the result string.
* @return pointer to connection. * @return the result string.
*/ */
Connection* getConnection() const { return m_connection; } std::string getResult() const { return m_result; }
/**
* @brief set the result string.
* @return the result string.
*/
void setResult(const std::string result) { m_result = result; }
/**
* @brief wait on notification.
*/
void waitSignal()
{
pthread_mutex_lock(&m_mutex);
while (m_result.size() == 0)
pthread_cond_wait(&m_cond, &m_mutex);
pthread_mutex_unlock(&m_mutex);
}
/**
* @brief send notification.
*/
void sendSignal()
{
pthread_mutex_lock(&m_mutex);
pthread_cond_signal(&m_cond);
pthread_mutex_unlock(&m_mutex);
}
private: private:
/** the data/message string */ /** the data string */
std::string m_data; std::string m_data;
/** the source connection */ /** the result string */
Connection* m_connection; std::string m_result;
/** mutex variable for exclusive lock */
pthread_mutex_t m_mutex;
/** condition variable for exclusive lock */
pthread_cond_t m_cond;
}; };