diff --git a/src/ebusd/mainloop.cpp b/src/ebusd/mainloop.cpp index b91995bb..d5dd6680 100644 --- a/src/ebusd/mainloop.cpp +++ b/src/ebusd/mainloop.cpp @@ -21,7 +21,6 @@ #include "main.h" #include "log.h" #include "data.h" -#include "config.h" using namespace std; @@ -41,9 +40,23 @@ static const char* columnNames[] = { "fields", "f", }; +/** the known column IDs according to @a columnNames. */ +static const size_t columnIds[] = { + COLUMN_TYPE, COLUMN_TYPE, + COLUMN_CIRCUIT, COLUMN_CIRCUIT, + COLUMN_NAME, COLUMN_NAME, + COLUMN_COMMENT, COLUMN_COMMENT, + COLUMN_QQ, COLUMN_QQ, + COLUMN_ZZ, COLUMN_ZZ, + COLUMN_PBSB, COLUMN_PBSB, + COLUMN_ID, COLUMN_ID, + COLUMN_FIELDS, COLUMN_FIELDS, +}; + /** the number of known column names. */ static const size_t columnCount = sizeof(columnNames) / sizeof(char*); + MainLoop::MainLoop(const struct options opt, Device *device, MessageMap* messages) : Thread(), m_device(device), m_reconnectCount(0), m_messages(messages), m_address(opt.address), m_scanConfig(opt.scanConfig), @@ -87,11 +100,18 @@ MainLoop::MainLoop(const struct options opt, Device *device, MessageMap* message m_htmlPath = opt.htmlPath; m_network = new Network(opt.localOnly, opt.port, opt.httpPort, &m_netQueue); m_network->start("network"); + if (!datahandler_register(m_busHandler, m_dataHandlers)) { + logError(lf_main, "error registering data handlers"); + } } MainLoop::~MainLoop() { join(); + + for (list::iterator it = m_dataHandlers.begin(); it != m_dataHandlers.end(); it++) { + delete *it; + } if (m_dumpFile) { delete m_dumpFile; m_dumpFile = NULL; @@ -121,17 +141,25 @@ MainLoop::~MainLoop() void MainLoop::run() { bool reload = true; - time_t lastTaskRun, now, lastSignal = 0; + time_t lastTaskRun, now, lastSignal = 0, since, sinkSince = 1; int taskDelay = 5; unsigned char lastScanAddress = 0; // 0 is known to be a master time(&now); lastTaskRun = now; + ostringstream updates; + list dataSinks; + deque messages; + for (list::iterator it = m_dataHandlers.begin(); it != m_dataHandlers.end(); it++) { + if ((*it)->isDataSink()) { + dataSinks.push_back(dynamic_cast(*it)); + } + (*it)->start(); + } while (true) { string result; - // pick the next message to handle - NetMessage* message = m_netQueue.pop(taskDelay); + NetMessage* netMessage = m_netQueue.pop(taskDelay); time(&now); if (nowfindAll("", "", false, true, true, true, false, true, sinkSince, now); + for (deque::iterator it = messages.begin(); it != messages.end(); it++) { + Message* message = *it; + for (list::iterator it = dataSinks.begin(); it != dataSinks.end(); it++) { + (*it)->notifyUpdate(message); + } + } + sinkSince = now; + } + if (netMessage==NULL) { continue; } - string request = message->getRequest(); - - time_t since, until; - time(&until); - bool listening = message->isListening(&since); - if (!listening) - since = until; - + string request = netMessage->getRequest(); + bool listening = netMessage->isListening(&since); + if (!listening) { + since = now; + } bool connected = true; if (request.length() > 0) { logDebug(lf_main, ">>> %s", request.c_str()); - result = decodeMessage(request, message->isHttp(), connected, listening, reload); + result = decodeMessage(request, netMessage->isHttp(), connected, listening, reload); - if (result.length() == 0 && !message->isHttp()) + if (result.length() == 0 && !netMessage->isHttp()) { result = getResultCode(RESULT_EMPTY); - - if (result.length() > 100) + } + if (result.length() > 100) { logDebug(lf_main, "<<< %s ...", result.substr(0, 100).c_str()); - else + } else { logDebug(lf_main, "<<< %s", result.c_str()); - - if (result.length() == 0) + } + if (result.length() == 0) { result = "\n"; // only for HTTP - else if (!message->isHttp()) + } else if (!netMessage->isHttp()) { result += "\n\n"; + } } if (listening) { - result += getUpdates(since, until); + messages = m_messages->findAll("", "", false, true, true, true, false, true, since, now); + updates.str(""); + updates.clear(); + for (deque::iterator it = messages.begin(); it != messages.end(); it++) { + Message* message = *it; + updates << message->getCircuit() << " " << message->getName() << " = " << dec; + message->decodeLastData(updates); + updates << endl; + } + result += updates.str(); } - // send result to client - message->setResult(result, listening, until, !connected); + netMessage->setResult(result, listening, now, !connected); } } @@ -265,8 +310,6 @@ void MainLoop::notifyDeviceData(const unsigned char byte, bool received) string MainLoop::decodeMessage(const string& data, const bool isHttp, bool& connected, bool& listening, bool& reload) { - ostringstream result; - // prepare data string token, previous; istringstream stream(data); @@ -840,7 +883,7 @@ string MainLoop::executeFind(vector &args) argPos = 0; // print usage break; } - columns.push_back(idx/2); + columns.push_back(columnIds[idx]); } if (columns.empty()) { argPos = 0; // print usage @@ -926,7 +969,7 @@ string MainLoop::executeFind(vector &args) bool found = false; ostringstream result; char str[32]; - for (deque::iterator it = messages.begin(); it < messages.end();) { + for (deque::iterator it = messages.begin(); it != messages.end();) { Message* message = *it++; if (!id.empty() && !message->checkIdPrefix(id)) { continue; @@ -1183,7 +1226,7 @@ string MainLoop::executeInfo(vector &args) " Report information about the daemon, the configuration, and seen devices."; ostringstream result; - result << "version: " << PACKAGE_STRING << "." REVISION "\n"; + result << "version: " << PACKAGE_STRING "." REVISION "\n"; if (m_busHandler->hasSignal()) { result << "signal: acquired\n"; result << "symbol rate: " << static_cast(m_busHandler->getSymbolRate()) << "\n"; @@ -1301,7 +1344,7 @@ string MainLoop::executeGet(vector &args, bool& connected) result << "{"; string lastCircuit = ""; time_t maxLastUp = 0; - for (deque::iterator it = messages.begin(); ret == RESULT_OK && it < messages.end();) { + for (deque::iterator it = messages.begin(); ret == RESULT_OK && it != messages.end();) { Message* message = *it++; unsigned char dstAddress = message->getDstAddress(); if (dstAddress == SYN) @@ -1450,31 +1493,8 @@ string MainLoop::executeGet(vector &args, bool& connected) result << "500 Internal Server Error"; break; } - result << "\r\nServer: ebusd/" PACKAGE_VERSION "\r\n\r\n"; + result << "\r\nServer: " PACKAGE_NAME "/" PACKAGE_VERSION "\r\n\r\n"; result << data; connected = false; return result.str(); } - -string MainLoop::getUpdates(time_t since, time_t until) -{ - ostringstream result; - - deque messages; - messages = m_messages->findAll("", "", 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->getCircuit() << " " << message->getName() << " = "; - message->decodeLastData(result); - result << endl; - } - - return result.str(); -} diff --git a/src/ebusd/mainloop.h b/src/ebusd/mainloop.h index 0b2cd027..87be896a 100644 --- a/src/ebusd/mainloop.h +++ b/src/ebusd/mainloop.h @@ -22,7 +22,11 @@ #include "message.h" #include "network.h" #include "bushandler.h" +#include "datahandler.h" #include "rotatefile.h" +#ifdef HAVE_CONFIG_H +# include +#endif /** \file mainloop.h */ @@ -33,8 +37,8 @@ using namespace std; */ class MainLoop : public Thread, DeviceListener { - public: + /** * Construct the main loop and create network and bus handling components. * @param opt the program options. @@ -54,11 +58,6 @@ public: */ BusHandler* getBusHandler() { return m_busHandler; } - /** - * Run the main loop. - */ - void run(); - /** * Add a client @a NetMessage to the queue. * @param message the client @a NetMessage to handle. @@ -68,50 +67,13 @@ public: // @copydoc virtual void notifyDeviceData(const unsigned char byte, bool received); +protected: + + // @copydoc + virtual void run(); + private: - /** the @a Device instance. */ - Device* m_device; - - /** the number of reconnects requested from the @a Device. */ - unsigned int m_reconnectCount; - - /** the @a RotateFile for writing sent/received bytes in log format, or NULL. */ - RotateFile* m_logRawFile; - - /** whether raw logging to @p logNotice is enabled (only relevant if m_logRawFile is NULL). */ - bool m_logRawEnabled; - - /** the @a RotateFile for dumping received data, or NULL. */ - RotateFile* m_dumpFile; - - /** the @a MessageMap instance. */ - MessageMap* m_messages; - - /** the own master address for sending on the bus. */ - const unsigned char m_address; - - /** whether to pick configuration files matching initial scan. */ - const bool m_scanConfig; - - /** the initial address to scan for @a m_scanConfig (@a ESC=none, 0xfe=broadcast ident, @a SYN=full scan, else: single slave address). */ - const unsigned char m_initialScan; - - /** whether to enable the hex command. */ - const bool m_enableHex; - - /** the created @a BusHandler instance. */ - BusHandler* m_busHandler; - - /** the created @a Network instance. */ - Network* m_network; - - /** the @a NetMessage @a Queue. */ - Queue m_netQueue; - - /** the path for HTML files served by the HTTP port. */ - string m_htmlPath; - /** * Decode and execute client message. * @param data the data string to decode (may be empty). @@ -246,13 +208,50 @@ private: */ string executeGet(vector &args, bool& connected); - /** - * Get the updates received since the specified time. - * @param since the start time from which to add updates (inclusive). - * @param until the end time to which to add updates (exclusive). - * @return result string to send back to client. - */ - string getUpdates(time_t since, time_t until); + /** the @a Device instance. */ + Device* m_device; + + /** the number of reconnects requested from the @a Device. */ + unsigned int m_reconnectCount; + + /** the @a RotateFile for writing sent/received bytes in log format, or NULL. */ + RotateFile* m_logRawFile; + + /** whether raw logging to @p logNotice is enabled (only relevant if m_logRawFile is NULL). */ + bool m_logRawEnabled; + + /** the @a RotateFile for dumping received data, or NULL. */ + RotateFile* m_dumpFile; + + /** the @a MessageMap instance. */ + MessageMap* m_messages; + + /** the own master address for sending on the bus. */ + const unsigned char m_address; + + /** whether to pick configuration files matching initial scan. */ + const bool m_scanConfig; + + /** the initial address to scan for @a m_scanConfig (@a ESC=none, 0xfe=broadcast ident, @a SYN=full scan, else: single slave address). */ + const unsigned char m_initialScan; + + /** whether to enable the hex command. */ + const bool m_enableHex; + + /** the created @a BusHandler instance. */ + BusHandler* m_busHandler; + + /** the created @a Network instance. */ + Network* m_network; + + /** the @a NetMessage @a Queue. */ + Queue m_netQueue; + + /** the path for HTML files served by the HTTP port. */ + string m_htmlPath; + + /** the registered @a DataHandler instances. */ + list m_dataHandlers; };