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

global Objects (network, commands, ebusloop, cycdata) relocated into new global Object baseloop.

Browse Source
This commit is contained in:
Roland Jax
2014-10-02 18:50:23 +02:00
parent bda1b87952
commit c65e74947f
13 changed files with 332 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChangeLog
+6
View File
@@ -1,6 +1,12 @@
ebusd - ChangeLog
-----------------
2014-10-09 Roland Jax <roland.jax@liwest.at>
* class Network splitted into classes Network and Connection.
* class Message separated from class Baseloop.
* global Objects (network, commands, ebusloop, cycdata) relocated into
new global Object baseloop.
2014-06-26 Roland Jax <roland.jax@liwest.at>
* ebusdump merged into ebusctl.
Makefile.am
+12 -11
View File
@@ -7,16 +7,17 @@ AM_CXXFLAGS = $(LIBEBUS_CFLAGS) -I$(top_srcdir)/lib -fpic -Wall -Wextra
bin_PROGRAMS = ebusd \
ebusctl
ebusd_SOURCES = src/main.cpp \
src/baseloop.cpp \
src/ebusloop.cpp \
src/cycdata.cpp \
src/network.cpp \
ebusd_SOURCES = lib/appl.cpp \
lib/daemon.cpp \
lib/tcpsocket.cpp \
lib/thread.cpp \
lib/logger.cpp \
lib/tcpsocket.cpp \
lib/daemon.cpp \
lib/appl.cpp
src/connection.cpp \
src/network.cpp \
src/ebusloop.cpp \
src/cycdata.cpp \
src/baseloop.cpp \
src/main.cpp
ebusd_LDADD = $(LIBEBUS_LIBS) \
-lpthread
@@ -24,9 +25,9 @@ ebusd_LDADD = $(LIBEBUS_LIBS) \
dist_noinst_SCRIPTS = autogen.sh
ebusctl_SOURCES = tools/ebusctl.cpp \
lib/appl.cpp \
lib/tcpsocket.cpp
ebusctl_SOURCES = lib/appl.cpp \
lib/tcpsocket.cpp \
tools/ebusctl.cpp
ebusctl_LDADD = $(LIBEBUS_LIBS)
src/baseloop.cpp
+85 -9
View File
@@ -20,14 +20,55 @@
#include "baseloop.h"
#include "logger.h"
#include "appl.h"
#include "network.h"
#include <algorithm>
#include <sstream>
#include <unistd.h>
extern LogInstance& L;
extern Appl& A;
BaseLoop::BaseLoop()
{
// create Commands DB
m_commands = ConfigCommands(A.getParam<const char*>("p_ebusconfdir"), CSV).getCommands();
L.log(bas, debug, "ebus configuration dir: %s", A.getParam<const char*>("p_ebusconfdir"));
L.log(bas, event, "commands DB with %d entries created", m_commands->size());
// create EBusLoop
m_ebusloop = new EBusLoop();
m_ebusloop->start("ebusloop");
// create CYCData
m_cycdata = new CYCData(m_ebusloop, m_commands);
m_cycdata->start("cycdata");
// create Network
m_network = new Network(A.getParam<bool>("p_localhost"));
m_network->addQueue(&m_queue);
m_network->start("network");
}
BaseLoop::~BaseLoop()
{
// free Network
if (m_network != NULL)
delete m_network;
// free CYCData
if (m_cycdata != NULL) {
m_cycdata->stop();
delete m_cycdata;
}
// free EBusLoop
if (m_ebusloop != NULL) {
m_ebusloop->stop();
m_ebusloop->join();
delete m_ebusloop;
}
// free Commands DB
if (m_commands != NULL)
delete m_commands;
}
void BaseLoop::start()
{
for (;;) {
@@ -118,7 +159,6 @@ std::string BaseLoop::decodeMessage(const std::string& data)
result << busCommand->getResult();
}
delete busCommand;
} else {
@@ -275,17 +315,53 @@ std::string BaseLoop::decodeMessage(const std::string& data)
result << "done";
break;
//~ case cfgreload:
// free CYCData
//~ if (m_cycdata != NULL) {
//~ m_cycdata->stop();
//~ delete m_cycdata;
//~ }
// free EBusLoop
//~ if (m_ebusloop != NULL) {
//~ m_ebusloop->stop();
//~ m_ebusloop->join();
//~ delete m_ebusloop;
//~ }
// free Commands DB
//~ if (m_commands != NULL)
//~ delete m_commands;
// create Commands DB
//~ m_commands = ConfigCommands(A.getParam<const char*>("p_ebusconfdir"), CSV).getCommands();
//~ L.log(bas, debug, "ebus configuration dir: %s", A.getParam<const char*>("p_ebusconfdir"));
//~ L.log(bas, event, "commands DB with %d entries created", m_commands->size());
// create EBusLoop
//~ m_ebusloop = new EBusLoop();
//~ m_ebusloop->start("ebusloop");
// create CYCData
//~ m_cycdata = new CYCData(m_ebusloop, m_commands);
//~ m_cycdata->start("cycdata");
//~ result << "done";
//~ break;
case help:
result << "commands:" << std::endl
<< " get - fetch ebus data 'get class cmd (sub)'" << std::endl
<< " set - set ebus values 'set class cmd value'" << std::endl
<< " cyc - fetch cycle data 'cyc class cmd (sub)'" << std::endl
<< " hex - send given hex value 'hex type value' (value: ZZPBSBNNDx)" << std::endl
<< " dump - change dump state 'dump state' (state: on|off)" << std::endl
<< " hex - send given hex value 'hex type value' (value: ZZPBSBNNDx)" << std::endl << std::endl
<< " dump - change dump state 'dump state' (state: on|off)" << std::endl << std::endl
<< " logarea - change log area 'logarea area,area,..' (area: bas|net|bus|cyc|all)" << std::endl
<< " loglevel - change log level 'loglevel level' (level: error|event|trace|debug)" << std::endl
<< " loglevel - change log level 'loglevel level' (level: error|event|trace|debug)" << std::endl << std::endl
//~ << " cfgreload - reload ebus configuration" << std::endl << std::endl
<< " stop - stop daemon" << std::endl
<< " quit - close connection" << std::endl
<< " quit - close connection" << std::endl << std::endl
<< " help - print this page";
break;
src/baseloop.h
+9 -25
View File
@@ -21,47 +21,30 @@
#define BASELOOP_H_
#include "libebus.h"
#include "network.h"
#include "ebusloop.h"
#include "cycdata.h"
#include "wqueue.h"
#include <string>
using namespace libebus;
class Connection;
class Message
{
public:
Message(const std::string data, void* source = NULL) : m_data(data), m_source(source) {}
Message(const Message& src) : m_data(src.m_data), m_source(src.m_source) {}
std::string getData() const { return m_data; }
void* getSource() const { return m_source; }
private:
std::string m_data;
void* m_source;
};
class BaseLoop
{
public:
BaseLoop(EBusLoop* ebusloop, CYCData* cycdata, Commands* commands)
: m_ebusloop(ebusloop), m_cycdata(cycdata), m_commands(commands) {}
BaseLoop();
~BaseLoop();
void start();
WQueue<Message*>* getQueue() { return &m_queue; }
void addMessage(Message* message) { m_queue.add(message); }
private:
EBusLoop* m_ebusloop;
CYCData* m_cycdata;
Commands* m_commands;
CYCData* m_cycdata;
EBusLoop* m_ebusloop;
Network* m_network;
WQueue<Message*> m_queue;
enum ClientCommand {
@@ -72,8 +55,8 @@ private:
dump, // change dump state
logarea, // change log area
loglevel, // change log level
//~ cfgreload, // reload ebus configuration
help, // print commands
notfound
};
@@ -86,6 +69,7 @@ private:
if (strcasecmp(item.c_str(), "dump") == 0) return dump;
if (strcasecmp(item.c_str(), "logarea") == 0) return logarea;
if (strcasecmp(item.c_str(), "loglevel") == 0) return loglevel;
//~ if (strcasecmp(item.c_str(), "cfgreload") == 0) return cfgreload;
if (strcasecmp(item.c_str(), "help") == 0) return help;
return notfound;
src/connection.cpp
+111
View File
@@ -0,0 +1,111 @@
/*
* Copyright (C) Roland Jax 2012-2014 <roland.jax@liwest.at>
*
* 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 <cstring>
extern LogInstance& L;
int Connection::m_count = -1;
void Connection::addResult(Message message)
{
Message* tmp = new Message(Message(message));
m_result.add(tmp);
}
void* Connection::run()
{
m_running = true;
int maxfd;
fd_set checkfds;
struct timeval timeout;
FD_ZERO(&checkfds);
FD_SET(m_notify.notifyFD(), &checkfds);
FD_SET(m_socket->getFD(), &checkfds);
(m_notify.notifyFD() > m_socket->getFD()) ?
(maxfd = m_notify.notifyFD()) : (maxfd = m_socket->getFD());
for (;;) {
fd_set readfds;
int ret;
// set select timeout 10 secs
timeout.tv_sec = 10;
timeout.tv_usec = 0;
// set readfds to inital checkfds
readfds = checkfds;
ret = select(maxfd + 1, &readfds, NULL, NULL, &timeout);
if (ret == 0) {
continue;
}
// new data from notify
if (FD_ISSET(m_notify.notifyFD(), &readfds))
break;
// new data from socket
if (FD_ISSET(m_socket->getFD(), &readfds)) {
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 || strncasecmp(data, "quit", 4) == 0)
break;
// send data
data[datalen] = '\0';
m_data->add(new Message(data, this));
// wait for result
L.log(net, debug, "[%08x] wait for result", getID());
Message* message = m_result.remove();
L.log(net, debug, "[%08x] result added", getID());
std::string result(message->getData());
if (m_socket->isValid() == true)
m_socket->send(result.c_str(), result.size());
else
break;
delete message;
}
}
delete m_socket;
m_running = false;
L.log(net, trace, "[%08x] connection closed", getID());
return NULL;
}
src/connection.h
+58
View File
@@ -0,0 +1,58 @@
/*
* Copyright (C) Roland Jax 2012-2014 <roland.jax@liwest.at>
*
* 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 "message.h"
#include "notify.h"
#include "thread.h"
class Connection : public Thread
{
public:
Connection(TCPSocket* socket, WQueue<Message*>* data)
: m_socket(socket), m_data(data), m_running(false) { m_count++; }
~Connection() { m_count--; }
void addResult(Message message);
void* run();
void stop() const { m_notify.notify(); }
bool isRunning() const { return m_running; }
pthread_t getID() { return this->self(); }
int numConnections() const { return m_count; }
private:
TCPSocket* m_socket;
WQueue<Message*>* m_data;
WQueue<Message*> m_result;
Notify m_notify;
bool m_running;
static int m_count;
};
#endif // CONNECTION_H_
src/cycdata.cpp
-1
View File
@@ -19,7 +19,6 @@
#include "cycdata.h"
#include "logger.h"
#include <sstream>
#include <iomanip>
extern LogInstance& L;
src/cycdata.h
+1 -2
View File
@@ -22,11 +22,10 @@
#include "libebus.h"
#include "ebusloop.h"
#include "thread.h"
#include <string>
using namespace libebus;
typedef std::map<int, Command*> map_t;
typedef map_t::const_iterator mapCI_t;
typedef std::pair<int, Command*> pair_t;
src/ebusloop.cpp
-1
View File
@@ -20,7 +20,6 @@
#include "ebusloop.h"
#include "logger.h"
#include "appl.h"
#include <iostream>
extern LogInstance& L;
extern Appl& A;
src/main.cpp
+6 -59
View File
@@ -21,19 +21,8 @@
#include "logger.h"
#include "daemon.h"
#include "appl.h"
#include "network.h"
#include "ebusloop.h"
#include "cycdata.h"
#include "baseloop.h"
#include <iostream>
#include <memory>
#include <csignal>
#include <cstring>
#include <cstdio>
#include <sstream>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
using namespace libebus;
@@ -41,10 +30,7 @@ Appl& A = Appl::Instance();
Daemon& D = Daemon::Instance();
LogInstance& L = LogInstance::Instance();
Network* network;
Commands* commands;
EBusLoop* ebusloop;
CYCData* cycdata;
BaseLoop* baseloop;
void define_args()
{
@@ -121,26 +107,8 @@ void define_args()
void shutdown()
{
// free Network
if (network != NULL)
delete network;
// free CYCData
if (cycdata != NULL) {
cycdata->stop();
delete cycdata;
}
// free EBusLoop
if (ebusloop != NULL) {
ebusloop->stop();
ebusloop->join();
delete ebusloop;
}
// free Commands DB
if (commands != NULL)
delete commands;
// stop threads
delete baseloop;
// reset all signal handlers to default
signal(SIGHUP, SIG_DFL);
@@ -223,32 +191,11 @@ int main(int argc, char* argv[])
usleep(100000);
L.log(bas, event, "ebusd started");
// create Commands DB
commands = ConfigCommands(A.getParam<const char*>("p_ebusconfdir"), CSV).getCommands();
L.log(bas, debug, "ebus configuration dir: %s", A.getParam<const char*>("p_ebusconfdir"));
L.log(bas, event, "commands DB with %d entries created", commands->size());
// create EBusLoop
ebusloop = new EBusLoop();
ebusloop->start("ebusloop");
// create CYCData
cycdata = new CYCData(ebusloop, commands);
cycdata->start("cycdata");
// create Network
network = new Network(A.getParam<bool>("p_localhost"));
// create BaseLoop
BaseLoop baseloop(ebusloop, cycdata, commands);
// start Network
network->addQueue(baseloop.getQueue());
network->start("network");
// start Baseloop
baseloop.start();
baseloop = new BaseLoop();
baseloop->start();
// shutdown
shutdown();
}
src/message.h
+43
View File
@@ -0,0 +1,43 @@
/*
* Copyright (C) Roland Jax 2012-2014 <roland.jax@liwest.at>
*
* 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 MESSAGE_H_
#define MESSAGE_H_
#include <string>
class Message
{
public:
Message(const std::string data, void* source = NULL) : m_data(data), m_source(source) {}
Message(const Message& src) : m_data(src.m_data), m_source(src.m_source) {}
std::string getData() const { return m_data; }
void* getSource() const { return m_source; }
private:
std::string m_data;
void* m_source;
};
#endif // MESSAGE_H_
src/network.cpp
-90
View File
@@ -20,100 +20,10 @@
#include "network.h"
#include "logger.h"
#include "appl.h"
#include <sstream>
#include <cstring>
#include <sys/select.h>
extern LogInstance& L;
extern Appl& A;
int Connection::m_count = -1;
void Connection::addResult(Message message)
{
Message* tmp = new Message(Message(message));
m_result.add(tmp);
}
void* Connection::run()
{
m_running = true;
int maxfd;
fd_set checkfds;
struct timeval timeout;
FD_ZERO(&checkfds);
FD_SET(m_notify.notifyFD(), &checkfds);
FD_SET(m_socket->getFD(), &checkfds);
(m_notify.notifyFD() > m_socket->getFD()) ?
(maxfd = m_notify.notifyFD()) : (maxfd = m_socket->getFD());
for (;;) {
fd_set readfds;
int ret;
// set select timeout 10 secs
timeout.tv_sec = 10;
timeout.tv_usec = 0;
// set readfds to inital checkfds
readfds = checkfds;
ret = select(maxfd + 1, &readfds, NULL, NULL, &timeout);
if (ret == 0) {
continue;
}
// new data from notify
if (FD_ISSET(m_notify.notifyFD(), &readfds))
break;
// new data from socket
if (FD_ISSET(m_socket->getFD(), &readfds)) {
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 || strncasecmp(data, "quit", 4) == 0)
break;
// send data
data[datalen] = '\0';
m_data->add(new Message(data, this));
// wait for result
L.log(net, debug, "[%08x] wait for result", getID());
Message* message = m_result.remove();
L.log(net, debug, "[%08x] result added", getID());
std::string result(message->getData());
if (m_socket->isValid() == true)
m_socket->send(result.c_str(), result.size());
else
break;
delete message;
}
}
delete m_socket;
m_running = false;
L.log(net, trace, "[%08x] connection closed", getID());
return NULL;
}
Network::Network(const bool localhost) : m_listening(false), m_running(false)
{
src/network.h
+1 -36
View File
@@ -20,42 +20,7 @@
#ifndef NETWORK_H_
#define NETWORK_H_
#include "tcpsocket.h"
#include "wqueue.h"
#include "thread.h"
#include "notify.h"
#include "baseloop.h"
#include <list>
#include <unistd.h>
class Connection : public Thread
{
public:
Connection(TCPSocket* socket, WQueue<Message*>* data)
: m_socket(socket), m_data(data), m_running(false) { m_count++; }
~Connection() { m_count--; }
void addResult(Message message);
void* run();
void stop() const { m_notify.notify(); }
bool isRunning() const { return m_running; }
pthread_t getID() { return this->self(); }
int numConnections() const { return m_count; }
private:
TCPSocket* m_socket;
WQueue<Message*>* m_data;
WQueue<Message*> m_result;
Notify m_notify;
bool m_running;
static int m_count;
};
#include "connection.h"
class Network : public Thread
{