added pipe mechanism for default main loop to synchronize with other threads

examples/echo/echo-client.cpp

@@ -7,6 +7,7 @@
 #include <pthread.h>
 #include <signal.h>
 #include <stdio.h>
+#include <cstring>
 
 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<DBus::Connection *>(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;
 }

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*> pipe_list;
 };
 
 } /* namespace DBus */

src/eventloop-integration.cpp

@@ -31,11 +31,14 @@
 #include <dbus-c++/debug.h>
 
 #include <sys/poll.h>
+#include <fcntl.h>
 
 #include <dbus/dbus.h>
 #include <errno.h>
+#include <cassert>
 
 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 <Pipe*>::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();