From a630d132c5be3b8732feea061749e25ee38019f0 Mon Sep 17 00:00:00 2001 From: Roland Jax Date: Sun, 4 Jan 2015 17:11:17 +0100 Subject: [PATCH] new command 'listen' added. --- src/ebusd/baseloop.cpp | 73 ++++++++++++++++++++++++++++++++++-------- src/ebusd/baseloop.h | 19 ++++++++--- src/ebusd/network.cpp | 66 +++++++++++++++++++++----------------- src/ebusd/network.h | 32 +++++++++++++++--- 4 files changed, 139 insertions(+), 51 deletions(-) diff --git a/src/ebusd/baseloop.cpp b/src/ebusd/baseloop.cpp index 27bc2b84..e0979d60 100644 --- a/src/ebusd/baseloop.cpp +++ b/src/ebusd/baseloop.cpp @@ -129,22 +129,37 @@ void BaseLoop::start() NetMessage* message = m_netQueue.remove(); string data = message->getData(); - data.erase(remove(data.begin(), data.end(), '\r'), data.end()); - data.erase(remove(data.begin(), data.end(), '\n'), data.end()); + time_t since, now; + time(&now); - L.log(bas, event, ">>> %s", data.c_str()); + bool listening = message->isListening(since); - // decode message - if (strcasecmp(data.c_str(), "STOP") != 0) - result = decodeMessage(data); - else - result = "done"; + if (data.length() == 0) + result = getUpdates(since, now); + else { + data.erase(remove(data.begin(), data.end(), '\r'), data.end()); + data.erase(remove(data.begin(), data.end(), '\n'), data.end()); - L.log(bas, event, "<<< %s", result.c_str()); + L.log(bas, event, ">>> %s", data.c_str()); + + if (data.length() == 0 && listening == true) + data = "listen"; + + // decode message + if (strcasecmp(data.c_str(), "STOP") != 0) + result = decodeMessage(data, listening); + else + result = "done"; + + L.log(bas, event, "<<< %s", result.c_str()); + result += "\n\n"; + } + + // add help sign for Connection::waitSignal() + result += "\r"; // send result to client - result += "\n\n"; - message->setResult(result); + message->setResult(result, listening, now); message->sendSignal(); // stop daemon @@ -160,7 +175,7 @@ void BaseLoop::logRaw(const unsigned char byte, bool received) { L.log(bus, event, ">%02x", byte); } -string BaseLoop::decodeMessage(const string& data) +string BaseLoop::decodeMessage(const string& data, bool& listening) { ostringstream result; @@ -482,6 +497,16 @@ string BaseLoop::decodeMessage(const string& data) result << "no message found"; break; } + case ct_listen: { + if (args.size() != argPos) { + result << "usage: 'listen'"; + break; + } + + bool enabled = !listening; + listening = enabled; + return (enabled ? "listen started" : "listen stopped"); + } case ct_scan: { if (args.size() == argPos) { result_t ret = m_busHandler->startScan(); @@ -578,7 +603,8 @@ string BaseLoop::decodeMessage(const string& data) result << "commands:" << endl << " read - read ebus values 'read [-v] [-f] [-m seconds] [-c class] name [field]'" << endl << " write - write ebus values 'write class name value[;value]*' or 'write -h ZZPBSBNNDx'" << endl - << " find - find ebus values 'find [-v] [-r] [-w] [-p] [-d] [-c class] [name]'" << endl << endl + << " find - find ebus values 'find [-v] [-r] [-w] [-p] [-d] [-c class] [name]'" << endl + << " listen - listen for updates 'listen'" << endl << endl << " scan - scan ebus known addresses 'scan'" << endl << " - scan ebus all addresses 'scan full'" << endl << " - show scan results 'scan result'" << endl << endl @@ -597,3 +623,24 @@ string BaseLoop::decodeMessage(const string& data) return result.str(); } +string BaseLoop::getUpdates(time_t since, time_t until) +{ + ostringstream result; + + deque messages; + messages = m_messages->findAll("", "", -1, false, true, true, true); + + for (deque::iterator it = messages.begin(); it < messages.end();) { + Message* message = *it++; + unsigned char dstAddress = message->getDstAddress(); + if (dstAddress == SYN) + continue; + time_t lastchg = message->getLastChangeTime(); + if (lastchg < since || lastchg >= until) + continue; + result << message->getClass() << " " << message->getName() << " = "; + result << message->getLastValue() << endl; + } + + return result.str(); +} diff --git a/src/ebusd/baseloop.h b/src/ebusd/baseloop.h index db95a2b3..057688d4 100644 --- a/src/ebusd/baseloop.h +++ b/src/ebusd/baseloop.h @@ -33,6 +33,7 @@ enum CommandType { ct_read, //!< read ebus values ct_write, //!< write ebus values ct_find, //!< find values + ct_listen, //!< listen for updates to values ct_scan, //!< scan ebus ct_log, //!< logger settings ct_raw, //!< toggle log raw data @@ -120,6 +121,7 @@ private: if (strcasecmp(str, "R") == 0 || strcasecmp(str, "READ") == 0) return ct_read; if (strcasecmp(str, "W") == 0 || strcasecmp(str, "WRITE") == 0) return ct_write; if (strcasecmp(str, "F") == 0 || strcasecmp(str, "FIND") == 0) return ct_find; + if (strcasecmp(str, "L") == 0 || strcasecmp(str, "LISTEN") == 0) return ct_listen; if (strcasecmp(str, "SCAN") == 0) return ct_scan; if (strcasecmp(str, "LOG") == 0) return ct_log; if (strcasecmp(str, "RAW") == 0) return ct_raw; @@ -131,11 +133,20 @@ private: } /** - * @brief decode and execute client message - * @param data the data string to decode - * @return result string to send back to client + * @brief Decode and execute client message. + * @param data the data string to decode (may be empty). + * @param listening true if client is in listening mode. + * @return result string to send back to client. */ - string decodeMessage(const string& data); + string decodeMessage(const string& data, bool& listening); + + /** + * @brief Get the updates received since the specified time. + * @param since the time from which to add the updates. + * @param until the time from which to add the updates. + * @return result string to send back to client. + */ + string getUpdates(time_t since, time_t until); }; diff --git a/src/ebusd/network.cpp b/src/ebusd/network.cpp index 396ac70c..52dcadbf 100644 --- a/src/ebusd/network.cpp +++ b/src/ebusd/network.cpp @@ -41,8 +41,8 @@ void Connection::run() int ret; struct timespec tdiff; - // set select timeout 10 secs - tdiff.tv_sec = 10; + // set timeout + tdiff.tv_sec = 2; tdiff.tv_nsec = 0; #ifdef HAVE_PPOLL @@ -70,6 +70,8 @@ void Connection::run() #endif #endif + time_t listenSince = 0; + for (;;) { #ifdef HAVE_PPOLL @@ -84,42 +86,45 @@ void Connection::run() #endif #endif - if (ret == 0) - continue; - + bool newData = false; + if (ret != 0) { #ifdef HAVE_PPOLL - // new data from notify - if (fds[0].revents & POLLIN) - break; + // new data from notify + if (fds[0].revents & POLLIN) + break; - // new data from socket - if (fds[1].revents & POLLIN) { + // new data from socket + newData = fds[1].revents & POLLIN; #else #ifdef HAVE_PSELECT - // new data from notify - if (FD_ISSET(m_notify.notifyFD(), &readfds)) - break; + // new data from notify + if (FD_ISSET(m_notify.notifyFD(), &readfds)) + break; - // new data from socket - if (FD_ISSET(m_socket->getFD(), &readfds)) { + // new data from socket + newData = FD_ISSET(m_socket->getFD(), &readfds); #endif #endif + } + if (newData == true || m_listening == true) { char data[256]; - size_t datalen; + size_t datalen = 0; - if (m_socket->isValid() == true) - datalen = m_socket->recv(data, sizeof(data)-1); - else - break; + if (newData == true) { + if (m_socket->isValid() == true) + datalen = m_socket->recv(data, sizeof(data)-1); + else + break; - // removed closed socket - if (datalen <= 0 || strncasecmp(data, "QUIT", 4) == 0) - break; + // removed closed socket + if (datalen <= 0 || strncasecmp(data, "QUIT", 4) == 0) + break; + } - // send data + // decode client data data[datalen] = '\0'; - NetMessage message(data); + NetMessage message(data, m_listening, listenSince); m_netQueue->add(&message); // wait for result @@ -129,11 +134,14 @@ void Connection::run() L.log(net, debug, "[%05d] result added", getID()); string result = message.getResult(); - if (m_socket->isValid() == true) - m_socket->send(result.c_str(), result.size()); - else + // remove help sign for Connection::waitSignal() + result.erase(remove(result.begin(), result.end(), '\r'), result.end()); + + if (m_socket->isValid() == false) break; + m_socket->send(result.c_str(), result.size()); + m_listening = message.isListening(listenSince); } } @@ -180,7 +188,7 @@ void Network::run() int ret; struct timespec tdiff; - // set select timeout 1 secs + // set timeout tdiff.tv_sec = 1; tdiff.tv_nsec = 0; diff --git a/src/ebusd/network.h b/src/ebusd/network.h index 7f4c4db3..2181e3fe 100644 --- a/src/ebusd/network.h +++ b/src/ebusd/network.h @@ -41,10 +41,13 @@ class NetMessage public: /** - * @brief constructs a new instance with message and source client address. + * @brief constructs a new instance with data received from the client. * @param data from client. + * @param listening whether the client is in listening mode. + * @param listenSince start timestamp of listening update. */ - NetMessage(const string data) : m_data(data) + NetMessage(const string data, const bool listening, const time_t listenSince) + : m_data(data), m_listening(listening), m_listenSince(listenSince) { pthread_mutex_init(&m_mutex, NULL); pthread_cond_init(&m_cond, NULL); @@ -78,10 +81,20 @@ public: string getResult() const { return m_result; } /** - * @brief set the result string. + * @brief Set the result string. * @param result the result string. + * @param listening whether the client is in listening mode. + * @param listenUntil end timestamp of last listening update. */ - void setResult(const string result) { m_result = result; } + void setResult(const string result, const bool listening, const time_t listenUntil) + { m_result = result; m_listening = listening; m_listenSince = listenUntil; } + + /** + * @brief Return whether the client is in listening mode. + * @param listenSince start timestamp of listening update. + * @return whether the client is in listen mode. + */ + bool isListening(time_t& listenSince) { listenSince = m_listenSince; return m_listening; } /** * @brief wait on notification. @@ -119,6 +132,12 @@ private: /** condition variable for exclusive lock */ pthread_cond_t m_cond; + /** whether the client is in listening mode */ + bool m_listening; + + /** start timestamp of listening update */ + time_t m_listenSince; + }; /** @@ -134,7 +153,7 @@ public: * @param netQueue the remote queue for network messages. */ Connection(TCPSocket* socket, WQueue* netQueue) - : m_socket(socket), m_netQueue(netQueue) + : m_socket(socket), m_netQueue(netQueue), m_listening(false) { m_id = ++m_ids; } /** @@ -169,6 +188,9 @@ private: /** sumary for opened connections */ static int m_ids; + /** whether the client is in listening mode */ + bool m_listening; + }; /**