diff --git a/examples/echo/echo-client.cpp b/examples/echo/echo-client.cpp index 8fe7ad7..f6d59a8 100644 --- a/examples/echo/echo-client.cpp +++ b/examples/echo/echo-client.cpp @@ -7,6 +7,7 @@ #include #include #include +#include using namespace std; @@ -31,22 +32,20 @@ static const int THREADS = 3; static bool spin = true; +EchoClient *g_client = NULL; + +DBus::Pipe *thread_pipe_list[THREADS]; + void *greeter_thread(void *arg) { - DBus::Connection *conn = reinterpret_cast(arg); - - EchoClient client(*conn, ECHO_SERVER_PATH, ECHO_SERVER_NAME); - char idstr[16]; + int i = (int) arg; snprintf(idstr, sizeof(idstr), "%lu", pthread_self()); - for (int i = 0; i < 30 && spin; ++i) - { - cout << client.Hello(idstr) << endl; - } + thread_pipe_list[i]->write (idstr, strlen (idstr) + 1); - cout << idstr << " done " << endl; + cout << idstr << " done (" << i << ")" << endl; return NULL; } @@ -60,6 +59,36 @@ void niam(int sig) dispatcher.leave(); } +void handler1 (const void *data, void *buffer, unsigned int nbyte) +{ + char *str = (char*) buffer; + cout << "buffer1: " << str << endl; + for (int i = 0; i < 30 && spin; ++i) + { + cout << g_client->Hello (str) << endl; + } +} + +void handler2 (const void *data, void *buffer, unsigned int nbyte) +{ + char *str = (char*) buffer; + cout << "buffer2: " << str << endl; + for (int i = 0; i < 30 && spin; ++i) + { + cout << g_client->Hello (str) << endl; + } +} + +void handler3 (const void *data, void *buffer, unsigned int nbyte) +{ + char *str = (char*) buffer; + cout << "buffer3: " << str << endl; + for (int i = 0; i < 30 && spin; ++i) + { + cout << g_client->Hello (str) << endl; + } +} + int main() { signal(SIGTERM, niam); @@ -71,13 +100,19 @@ int main() DBus::Connection conn = DBus::Connection::SessionBus(); + EchoClient client (conn, ECHO_SERVER_PATH, ECHO_SERVER_NAME); + g_client = &client; + pthread_t threads[THREADS]; + thread_pipe_list[0] = dispatcher.add_pipe (handler1, NULL); + thread_pipe_list[1] = dispatcher.add_pipe (handler2, NULL); + thread_pipe_list[2] = dispatcher.add_pipe (handler3, NULL); for (int i = 0; i < THREADS; ++i) { - pthread_create(threads+i, NULL, greeter_thread, &conn); + pthread_create(threads+i, NULL, greeter_thread, (void*) i); } - + dispatcher.enter(); cout << "terminating" << endl; @@ -87,5 +122,9 @@ int main() pthread_join(threads[i], NULL); } + dispatcher.del_pipe (thread_pipe_list[0]); + dispatcher.del_pipe (thread_pipe_list[1]); + dispatcher.del_pipe (thread_pipe_list[2]); + return 0; } diff --git a/include/dbus-c++/eventloop-integration.h b/include/dbus-c++/eventloop-integration.h index b8e02c7..a061aa8 100644 --- a/include/dbus-c++/eventloop-integration.h +++ b/include/dbus-c++/eventloop-integration.h @@ -57,30 +57,51 @@ class DXXAPI BusWatch : public Watch, public DefaultWatch friend class BusDispatcher; }; +class DXXAPI Pipe +{ +public: + /*! + * Write some data into the communication pipe. + * + * @param buffer The raw data to write. + * @param nbytes The number of bytes to write from the buffer. + */ + void write(const void *buffer, unsigned int nbytes); + + /*! + * Simply write one single byte into the pipe. This is a shortcut + * if there's really no data to transport, but to actvate the handler. + */ + void signal(); + +private: + void(*_handler)(const void *data, void *buffer, unsigned int nbyte); + int fd_write; + int fd_read; + const void *data; + + // allow construction only in BusDipatcher + Pipe () {}; + ~Pipe () {}; + +friend class BusDispatcher; +}; + class DXXAPI BusDispatcher : public Dispatcher, public DefaultMainLoop { public: - - int _pipe[2]; - - BusDispatcher() : _running(false) - { - //pipe to create a new fd used to unlock a dispatcher at any - // moment (used by leave function) - int ret = pipe(_pipe); - if (ret == -1) throw Error("PipeError:errno", toString(errno).c_str()); - - _fdunlock[0] = _pipe[0]; - _fdunlock[1] = _pipe[1]; - } - - ~BusDispatcher() - {} + BusDispatcher(); + + ~BusDispatcher() {} virtual void enter(); virtual void leave(); + virtual Pipe *add_pipe(void(*handler)(const void *data, void *buffer, unsigned int nbyte), const void *data); + + virtual void del_pipe (Pipe *pipe); + virtual void do_iteration(); virtual Timeout *add_timeout(Timeout::Internal *); @@ -96,8 +117,9 @@ public: void timeout_expired(DefaultTimeout &); private: - bool _running; + int _pipe[2]; + std::list pipe_list; }; } /* namespace DBus */ diff --git a/src/eventloop-integration.cpp b/src/eventloop-integration.cpp index 8b9c49b..d20154a 100644 --- a/src/eventloop-integration.cpp +++ b/src/eventloop-integration.cpp @@ -31,11 +31,14 @@ #include #include +#include #include #include +#include using namespace DBus; +using namespace std; BusTimeout::BusTimeout(Timeout::Internal *ti, BusDispatcher *bd) : Timeout(ti), DefaultTimeout(Timeout::interval(), true, bd) @@ -71,6 +74,28 @@ void BusWatch::toggle() DefaultWatch::enabled(Watch::enabled()); } +void Pipe::write(const void *buffer, unsigned int nbytes) +{ + ::write(fd_write, buffer, nbytes); +} + +void Pipe::signal() +{ + ::write(fd_write, '\0', 1); +} + +BusDispatcher::BusDispatcher() : + _running(false) +{ + // pipe to create a new fd used to unlock a dispatcher at any + // moment (used by leave function) + int ret = pipe(_pipe); + if (ret == -1) throw Error("PipeError:errno", toString(errno).c_str()); + + _fdunlock[0] = _pipe[0]; + _fdunlock[1] = _pipe[1]; +} + void BusDispatcher::enter() { debug_log("entering dispatcher %p", this); @@ -80,6 +105,27 @@ void BusDispatcher::enter() while (_running) { do_iteration(); + + for (std::list ::const_iterator p_it = pipe_list.begin (); + p_it != pipe_list.end (); + ++p_it) + { + const Pipe* read_pipe = *p_it; + char buf; + char buf_str[1024]; + int i = 0; + + while (read(read_pipe->fd_read, &buf, 1) > 0) + { + buf_str[i] = buf; + ++i; + } + + if (i > 0) + { + read_pipe->_handler (read_pipe->data, buf_str, i); + } + } } debug_log("leaving dispatcher %p", this); @@ -96,6 +142,34 @@ void BusDispatcher::leave() close(_fdunlock[0]); } +Pipe *BusDispatcher::add_pipe(void(*handler)(const void *data, void *buffer, unsigned int nbyte), const void *data) +{ + int fd[2]; + Pipe *new_pipe = new Pipe (); + new_pipe->_handler = handler; + new_pipe->data = data; + pipe_list.push_back (new_pipe); + + if (pipe(fd) == 0) + { + new_pipe->fd_read = fd[0]; + new_pipe->fd_write = fd[1]; + fcntl(new_pipe->fd_read, F_SETFL, O_NONBLOCK); + } + else + { + throw Error("PipeError:errno", toString(errno).c_str()); + } + + return new_pipe; +} + +void BusDispatcher::del_pipe (Pipe *pipe) +{ + pipe_list.remove (pipe); + delete pipe; +} + void BusDispatcher::do_iteration() { dispatch_pending();