nils/ebusd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

new command 'listen' added.

Browse Source
This commit is contained in:
Roland Jax
2015-01-04 17:11:17 +01:00
parent 9fcff1221e
commit a630d132c5
4 changed files with 139 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ebusd/baseloop.cpp
+60 -13
View File
@@ -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<Message*> messages;
messages = m_messages->findAll("", "", -1, false, true, true, true);
for (deque<Message*>::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();
}
src/ebusd/baseloop.h
+15 -4
View File
@@ -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);
};
src/ebusd/network.cpp
+37 -29
View File
@@ -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;
src/ebusd/network.h
+27 -5
View File
@@ -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<NetMessage*>* 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;
};
/**