diff --git a/src/ebusd/Makefile.am b/src/ebusd/Makefile.am index 804fd608..c1be62df 100644 --- a/src/ebusd/Makefile.am +++ b/src/ebusd/Makefile.am @@ -6,15 +6,12 @@ AM_CXXFLAGS = -fpic \ bin_PROGRAMS = ebusd -ebusd_SOURCES = connection.cpp \ - connection.h \ - network.cpp \ +ebusd_SOURCES = network.cpp \ network.h \ ebusloop.cpp \ ebusloop.h \ baseloop.cpp \ baseloop.h \ - netmessage.h \ ebusd.cpp ebusd_LDADD = $(top_srcdir)/src/lib/utils/libutils.a \ diff --git a/src/ebusd/connection.cpp b/src/ebusd/connection.cpp deleted file mode 100644 index 239bb0a1..00000000 --- a/src/ebusd/connection.cpp +++ /dev/null @@ -1,102 +0,0 @@ -/* - * Copyright (C) Roland Jax 2012-2014 - * - * This file is part of ebusd. - * - * ebusd 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 3 of the License, or - * (at your option) any later version. - * - * ebusd 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 ebusd. If not, see http://www.gnu.org/licenses/. - */ - -#include "connection.h" -#include "logger.h" -#include -#include - -extern Logger& L; - -int Connection::m_sum = 0; - -void* Connection::run() -{ - m_running = true; - - int ret, nfds = 2; - struct pollfd fds[nfds]; - struct timespec tdiff; - - // set select timeout 10 secs - tdiff.tv_sec = 10; - tdiff.tv_nsec = 0; - - memset(fds, 0, sizeof(fds)); - - fds[0].fd = m_notify.notifyFD(); - fds[0].events = POLLIN; - - fds[1].fd = m_socket->getFD(); - fds[1].events = POLLIN; - - for (;;) { - // wait for new fd event - ret = ppoll(fds, nfds, &tdiff, NULL); - - if (ret == 0) { - continue; - } - - // new data from notify - if (fds[0].revents & POLLIN) - break; - - // new data from socket - if (fds[1].revents & POLLIN) { - char data[256]; - size_t datalen; - - if (m_socket->isValid() == true) - datalen = m_socket->recv(data, sizeof(data)-1); - else - break; - - // removed closed socket - if (datalen <= 0 || strcasecmp(data, "QUIT") == 0) - break; - - // send data - data[datalen] = '\0'; - NetMessage message(data); - m_netQueue->add(&message); - - // wait for result - L.log(net, debug, "[%05d] wait for result", getID()); - message.waitSignal(); - - L.log(net, debug, "[%05d] result added", getID()); - std::string result = message.getResult(); - - if (m_socket->isValid() == true) - m_socket->send(result.c_str(), result.size()); - else - break; - - } - - } - - delete m_socket; - m_running = false; - L.log(net, trace, "[%05d] connection closed", getID()); - - return NULL; -} - diff --git a/src/ebusd/connection.h b/src/ebusd/connection.h deleted file mode 100644 index e860bc8c..00000000 --- a/src/ebusd/connection.h +++ /dev/null @@ -1,53 +0,0 @@ -/* - * Copyright (C) Roland Jax 2012-2014 - * - * This file is part of ebusd. - * - * ebusd 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 3 of the License, or - * (at your option) any later version. - * - * ebusd 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 ebusd. If not, see http://www.gnu.org/licenses/. - */ - -#ifndef CONNECTION_H_ -#define CONNECTION_H_ - -#include "tcpsocket.h" -#include "wqueue.h" -#include "notify.h" -#include "thread.h" -#include "netmessage.h" - -class Connection : public Thread -{ - -public: - Connection(TCPSocket* socket, WQueue* netQueue) - : m_socket(socket), m_netQueue(netQueue), m_running(false) { m_sum++; m_id = m_sum;} - - void* run(); - void stop() const { m_notify.notify(); } - bool isRunning() const { return m_running; } - - int getID() { return m_id; } - -private: - TCPSocket* m_socket; - WQueue* m_netQueue; - Notify m_notify; - bool m_running; - int m_id; - - static int m_sum; - -}; - -#endif // CONNECTION_H_ diff --git a/src/ebusd/netmessage.h b/src/ebusd/netmessage.h deleted file mode 100644 index 5c3449ae..00000000 --- a/src/ebusd/netmessage.h +++ /dev/null @@ -1,117 +0,0 @@ -/* - * Copyright (C) Roland Jax 2012-2014 - * - * This file is part of ebusd. - * - * ebusd 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 3 of the License, or - * (at your option) any later version. - * - * ebusd 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 ebusd. If not, see http://www.gnu.org/licenses/. - */ - -#ifndef NETMESSAGE_H_ -#define NETMESSAGE_H_ - -#include - -/** forward declaration for class Connection */ -class Connection; - -/** - * @brief class for data/message transfer between connection and baseloop. - */ -class NetMessage -{ - -public: - /** - * @brief constructs a new instance with message and source client address. - * @param data from client. - */ - NetMessage(const std::string data) : m_data(data) - { - 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. - * @param src message object for copy. - */ - NetMessage(const NetMessage& src) : m_data(src.m_data) {} - - /** - * @brief get the data string. - * @return the data string. - */ - std::string getData() const { return m_data; } - - /** - * @brief get the result string. - * @return the result string. - */ - 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: - /** the data string */ - std::string m_data; - - /** the result string */ - std::string m_result; - - /** mutex variable for exclusive lock */ - pthread_mutex_t m_mutex; - - /** condition variable for exclusive lock */ - pthread_cond_t m_cond; - -}; - - -#endif // NETMESSAGE_H_ diff --git a/src/ebusd/network.cpp b/src/ebusd/network.cpp index 92b18877..99dd0092 100644 --- a/src/ebusd/network.cpp +++ b/src/ebusd/network.cpp @@ -20,16 +20,93 @@ #include "network.h" #include "logger.h" #include "appl.h" +#include #include extern Logger& L; extern Appl& A; +int Connection::m_ids = 0; -Network::Network(const bool localhost, WQueue* netQueue) +void* Connection::run() +{ + m_running = true; + + int ret, nfds = 2; + struct pollfd fds[nfds]; + struct timespec tdiff; + + // set select timeout 10 secs + tdiff.tv_sec = 10; + tdiff.tv_nsec = 0; + + memset(fds, 0, sizeof(fds)); + + fds[0].fd = m_notify.notifyFD(); + fds[0].events = POLLIN; + + fds[1].fd = m_socket->getFD(); + fds[1].events = POLLIN; + + for (;;) { + // wait for new fd event + ret = ppoll(fds, nfds, &tdiff, NULL); + + if (ret == 0) { + continue; + } + + // new data from notify + if (fds[0].revents & POLLIN) + break; + + // new data from socket + if (fds[1].revents & POLLIN) { + char data[256]; + size_t datalen; + + if (m_socket->isValid() == true) + datalen = m_socket->recv(data, sizeof(data)-1); + else + break; + + // removed closed socket + if (datalen <= 0 || strcasecmp(data, "QUIT") == 0) + break; + + // send data + data[datalen] = '\0'; + NetMessage message(data); + m_netQueue->add(&message); + + // wait for result + L.log(net, debug, "[%05d] wait for result", getID()); + message.waitSignal(); + + L.log(net, debug, "[%05d] result added", getID()); + std::string result = message.getResult(); + + if (m_socket->isValid() == true) + m_socket->send(result.c_str(), result.size()); + else + break; + + } + + } + + delete m_socket; + m_running = false; + L.log(net, trace, "[%05d] connection closed", getID()); + + return NULL; +} + + +Network::Network(const bool local, WQueue* netQueue) : m_netQueue(netQueue), m_listening(false), m_running(false) { - if (localhost == true) + if (local == true) m_tcpServer = new TCPServer(A.getOptVal("port"), "127.0.0.1"); else m_tcpServer = new TCPServer(A.getOptVal("port"), "0.0.0.0"); diff --git a/src/ebusd/network.h b/src/ebusd/network.h index 2805ec95..2f25b46e 100644 --- a/src/ebusd/network.h +++ b/src/ebusd/network.h @@ -20,28 +20,218 @@ #ifndef NETWORK_H_ #define NETWORK_H_ -#include "connection.h" +#include "tcpsocket.h" +#include "wqueue.h" +#include "notify.h" +#include "thread.h" +#include +/** forward declaration for class connection */ +class Connection; + +/** + * @brief class for data/message transfer between connection and baseloop. + */ +class NetMessage +{ + +public: + /** + * @brief constructs a new instance with message and source client address. + * @param data from client. + */ + NetMessage(const std::string data) : m_data(data) + { + 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. + * @param src message object for copy. + */ + NetMessage(const NetMessage& src) : m_data(src.m_data) {} + + /** + * @brief get the data string. + * @return the data string. + */ + std::string getData() const { return m_data; } + + /** + * @brief get the result string. + * @return the result string. + */ + 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: + /** the data string */ + std::string m_data; + + /** the result string */ + std::string m_result; + + /** mutex variable for exclusive lock */ + pthread_mutex_t m_mutex; + + /** condition variable for exclusive lock */ + pthread_cond_t m_cond; + +}; + +/** + * @brief class connection which handle client and baseloop communication. + */ +class Connection : public Thread +{ + +public: + /** + * @brief create a new connection instance. + * @param socket the tcp socket for communication. + * @param netQueue the remote queue for network messages. + */ + Connection(TCPSocket* socket, WQueue* netQueue) + : m_socket(socket), m_netQueue(netQueue), m_running(false) + { m_id = ++m_ids; } + + /** + * @brief endless loop for connection instance. + * @return void pointer. + */ + void* run(); + + /** + * @brief closs active connection. + */ + void stop() const { m_notify.notify(); } + + /** + * @brief status of connection instance. + * @return true if connection is running. + */ + bool isRunning() const { return m_running; } + + /** + * @brief return own connection id. + * @return id of current connection. + */ + int getID() { return m_id; } + +private: + /** the tcp socket instance */ + TCPSocket* m_socket; + + /** remote queue for network messages */ + WQueue* m_netQueue; + + /** notification object for shutdown procedure */ + Notify m_notify; + + /** true if this instance is running */ + bool m_running; + + /** id of current connection*/ + int m_id; + + /** sumary for opened connections */ + static int m_ids; + +}; + +/** + * @brief class network which listening on tcp socket for incoming connections. + */ class Network : public Thread { public: - Network(const bool localhost, WQueue* netQueue); + /** + * @brief create a new network instance and listening for incoming connections. + * @param local true to accept connections only for local host. + * @param netQueue the remote queue for network messages. + */ + Network(const bool local, WQueue* netQueue); + + /** + * @brief destructor. + */ ~Network(); + /** + * @brief endless loop for network instance. + * @return void pointer. + */ void* run(); + + /** + * @brief shutdown network subsystem. + */ void stop() const { m_notify.notify(); usleep(100000); } private: + /** container for active connections */ std::list m_connections; + + /** remote queue for network messages */ WQueue* m_netQueue; + + /** the tcp server instance */ TCPServer* m_tcpServer; + + /** notification object for shutdown procedure */ Notify m_notify; + + /** true if this instance is listening */ bool m_listening; + + /** true if this instance is running */ bool m_running; + /** + * @brief clean inactive connections from container. + */ void cleanConnections(); }; #endif // NETWORK_H_ +