TCP notification stream support (not fully implemented yet, but transport stuff is working).

Support multiple event sources in the engine.
Clean up HTTP/TCP stuff.


git-svn-id: http://svn.drobilla.net/lad/trunk/ingen@1721 a436a847-0d15-0410-975c-d299462d15a1

8 files changed, 97 insertions, 42 deletions

diff --git a/src/client/ClientStore.cpp b/src/client/ClientStore.cpp
index fbcf5929..cf9e06a0 100644
--- a/src/client/ClientStore.cpp
+++ b/src/client/ClientStore.cpp
@@ -55,7 +55,7 @@ ClientStore::ClientStore(SharedPtr<EngineInterface> engine, SharedPtr<SigClientI
 	emitter->signal_property_change.connect(sigc::mem_fun(this, &ClientStore::set_property));
 	emitter->signal_port_value.connect(sigc::mem_fun(this, &ClientStore::set_port_value));
 	emitter->signal_voice_value.connect(sigc::mem_fun(this, &ClientStore::set_voice_value));
-	emitter->signal_port_activity.connect(sigc::mem_fun(this, &ClientStore::port_activity));
+	emitter->signal_activity.connect(sigc::mem_fun(this, &ClientStore::activity));
 }
 
 
@@ -557,13 +557,13 @@ ClientStore::set_voice_value(const string& port_path, uint32_t voice, const Raul
 
 	
 void
-ClientStore::port_activity(const Path& port_path)
+ClientStore::activity(const Path& path)
 {
-	SharedPtr<PortModel> port = PtrCast<PortModel>(object(port_path));
+	SharedPtr<PortModel> port = PtrCast<PortModel>(object(path));
 	if (port)
 		port->signal_activity.emit();
 	else
-		cerr << "ERROR: activity for nonexistant port " << port_path << endl;
+		cerr << "ERROR: activity for nonexistant port " << path << endl;
 }
 
 
diff --git a/src/client/ClientStore.hpp b/src/client/ClientStore.hpp
index 291f9af9..944ab752 100644
--- a/src/client/ClientStore.hpp
+++ b/src/client/ClientStore.hpp
@@ -118,7 +118,7 @@ private:
 	// Slots for SigClientInterface signals
 	void rename(const Path& old_path, const Path& new_path);
 	void patch_cleared(const Path& path);
-	void port_activity(const Path& port_path);
+	void activity(const Path& path);
 	
 	bool attempt_connection(const Path& src_port_path, const Path& dst_port_path, bool add_orphan=false);
 	
diff --git a/src/client/HTTPClientReceiver.cpp b/src/client/HTTPClientReceiver.cpp
index 9116f853..572ff548 100644
--- a/src/client/HTTPClientReceiver.cpp
+++ b/src/client/HTTPClientReceiver.cpp
@@ -20,6 +20,8 @@
 #include <cstring>
 #include <iostream>
 #include <sstream>
+#include <sys/socket.h>
+#include <errno.h>
 #include "module/Module.hpp"
 #include "HTTPClientReceiver.hpp"
 
@@ -49,33 +51,73 @@ HTTPClientReceiver::~HTTPClientReceiver()
 }
 
 
+HTTPClientReceiver::Listener::~Listener()
+{
+	close(_sock);
+}
+
+HTTPClientReceiver::Listener::Listener(SoupSession* session, const std::string uri)
+	: _uri(uri)
+	, _session(session)
+{
+	string port_str = uri.substr(uri.find_last_of(":")+1);
+	int port = atoi(port_str.c_str());
+
+	cout << "HTTP listen URI: " << uri << " port: " << port << endl;
+
+	struct sockaddr_in servaddr;
+	
+	// Create listen address
+	memset(&servaddr, 0, sizeof(servaddr));
+	servaddr.sin_family      = AF_INET;
+	servaddr.sin_port        = htons(port);
+	
+	// Create listen socket
+	if ((_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
+		cerr << "Error creating listening socket: %s" << strerror(errno) << endl;
+		_sock = -1;
+		return;
+	}
+
+	// Set remote address (FIXME: always localhost)
+	if (inet_aton("127.0.0.1", &servaddr.sin_addr) <= 0) {
+		cerr << "Invalid remote IP address" << endl;
+		_sock = -1;
+		return;
+	}
+
+	// Connect to server
+	if (connect(_sock, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0) {
+		cerr << "Error calling connect: " << strerror(errno) << endl;
+		_sock = -1;
+		return;
+	}
+}
+
+
 void
 HTTPClientReceiver::Listener::_run()
 {
-#if 0
-	cout << "LISTENER RUN" << endl;
-	/*const string uri = "http://localhost:16180";
-	SoupMessage* msg = soup_message_new("GET", (uri + "/stream").c_str());
-	soup_message_headers_set_encoding(msg->response_headers, SOUP_ENCODING_CHUNKED);
-	soup_session_send_message(_session, msg);*/
-	
-	size_t offset = 0;
-	soup_message_body_set_accumulate(_msg->response_body, false);
+	char   in   = '\0';
+	char   last = '\0';
+	string recv = "";
+
 	while (true) {
-		SoupBuffer* chunk = soup_message_body_get_chunk(_msg->response_body, offset);
-		if (chunk == NULL) {
-			//cout << "WAITING FOR DATA" << endl;
-		} else if (chunk->length == 0) {
-			cout << "CHUNKED TRANSFER COMPLETED" << endl;
-			break;
-		} else {
-			cout << "RECEIVED CHUNK: " << (char*)chunk->data << endl;
-			offset += chunk->length;
+		while (read(_sock, &in, 1) > 0 ) {
+			recv += in;
+			if (last == '\n' && in == '\n') {
+				if (recv != "") {
+					cout << "RECEIVED UPDATE:\n" << recv << endl;
+					recv = "";
+					last = '\0';
+				}
+				break;
+			}
+			last = in;
 		}
 	}
 
-	cout << "LISTENER FINISHED" << endl;
-#endif
+	cout << "HTTP listener finished" << endl;
 }
 
 
@@ -84,10 +126,10 @@ HTTPClientReceiver::message_callback(SoupSession* session, SoupMessage* msg, voi
 {
 	HTTPClientReceiver* me = (HTTPClientReceiver*)ptr;
 	const string path = soup_message_get_uri(msg)->path;
-	cout << "MESSAGE: " << path << endl;
 	if (path == "/") {
 		me->_target->response_ok(0);
 		me->_target->enable();
+	
 	} else if (path == "/plugins") {
 		if (msg->response_body->data == NULL) {
 			cout << "ERROR: Empty response" << endl;
@@ -98,6 +140,7 @@ HTTPClientReceiver::message_callback(SoupSession* session, SoupMessage* msg, voi
 					Glib::ustring(msg->response_body->data),
 					Glib::ustring("."), Glib::ustring(""));
 		}
+
 	} else if (path == "/patch") {
 		if (msg->response_body->data == NULL) {
 			cout << "ERROR: Empty response" << endl;
@@ -108,10 +151,19 @@ HTTPClientReceiver::message_callback(SoupSession* session, SoupMessage* msg, voi
 					Glib::ustring(msg->response_body->data),
 					Glib::ustring("/patch/"), Glib::ustring(""));
 		}
+
 	} else if (path == "/stream") {
-		cout << "STREAM" << endl;
-		//me->_listener = boost::shared_ptr<Listener>(new Listener(me->_session, msg));
-		//me->_listener->start();
+		if (msg->response_body->data == NULL) {
+			cout << "ERROR: Empty response" << endl;
+		} else {
+			string uri = string(soup_uri_to_string(soup_message_get_uri(msg), false));
+			uri = uri.substr(0, uri.find_last_of(":"));
+			uri += string(":") + msg->response_body->data;
+			cout << "Stream URI: " << uri << endl;
+			me->_listener = boost::shared_ptr<Listener>(new Listener(me->_session, uri));
+			me->_listener->start();
+		}
+
 	} else {
 		cerr << "UNKNOWN MESSAGE: " << path << endl;
 	}
@@ -163,3 +215,4 @@ HTTPClientReceiver::stop()
 
 } // namespace Client
 } // namespace Ingen
+
diff --git a/src/client/HTTPClientReceiver.hpp b/src/client/HTTPClientReceiver.hpp
index 379ffe2d..015a551f 100644
--- a/src/client/HTTPClientReceiver.hpp
+++ b/src/client/HTTPClientReceiver.hpp
@@ -50,11 +50,13 @@ private:
 
 	class Listener : public Raul::Thread {
 	public:
-		Listener(SoupSession* session, SoupMessage* msg) : _session(session), _msg(msg) {}
+		Listener(SoupSession* session, const std::string uri);
+		~Listener();
 		void _run();
 	private:
+		std::string  _uri;
+		int          _sock;
 		SoupSession* _session;
-		SoupMessage* _msg;
 	};
 	
 	friend class Listener;
diff --git a/src/client/OSCClientReceiver.cpp b/src/client/OSCClientReceiver.cpp
index fa191206..bc8659df 100644
--- a/src/client/OSCClientReceiver.cpp
+++ b/src/client/OSCClientReceiver.cpp
@@ -155,7 +155,7 @@ OSCClientReceiver::setup_callbacks()
 	lo_server_thread_add_method(_st, "/ingen/set_property", NULL, set_property_cb, this);
 	lo_server_thread_add_method(_st, "/ingen/set_port_value", "sf", set_port_value_cb, this);
 	lo_server_thread_add_method(_st, "/ingen/set_voice_value", "sif", set_voice_value_cb, this);
-	lo_server_thread_add_method(_st, "/ingen/port_activity", "s", port_activity_cb, this);
+	lo_server_thread_add_method(_st, "/ingen/activity", "s", activity_cb, this);
 	lo_server_thread_add_method(_st, "/ingen/program_add", "siis", program_add_cb, this);
 	lo_server_thread_add_method(_st, "/ingen/program_remove", "sii", program_remove_cb, this);
 }
@@ -321,11 +321,11 @@ OSCClientReceiver::_set_voice_value_cb(const char* path, const char* types, lo_a
 
 
 int
-OSCClientReceiver::_port_activity_cb(const char* path, const char* types, lo_arg** argv, int argc, lo_message msg)
+OSCClientReceiver::_activity_cb(const char* path, const char* types, lo_arg** argv, int argc, lo_message msg)
 {
 	const char* const port_path = &argv[0]->s;
 
-	_target->port_activity(port_path);
+	_target->activity(port_path);
 
 	return 0;	
 }
diff --git a/src/client/OSCClientReceiver.hpp b/src/client/OSCClientReceiver.hpp
index ea5871b3..9203f096 100644
--- a/src/client/OSCClientReceiver.hpp
+++ b/src/client/OSCClientReceiver.hpp
@@ -95,7 +95,7 @@ private:
 	LO_HANDLER(set_property);
 	LO_HANDLER(set_port_value);
 	LO_HANDLER(set_voice_value);
-	LO_HANDLER(port_activity);
+	LO_HANDLER(activity);
 	LO_HANDLER(program_add);
 	LO_HANDLER(program_remove);
 };
diff --git a/src/client/SigClientInterface.hpp b/src/client/SigClientInterface.hpp
index 63586832..36ca44b9 100644
--- a/src/client/SigClientInterface.hpp
+++ b/src/client/SigClientInterface.hpp
@@ -66,7 +66,7 @@ public:
 	sigc::signal<void, string, string, Raul::Atom>         signal_property_change; 
 	sigc::signal<void, string, Raul::Atom>                 signal_port_value; 
 	sigc::signal<void, string, uint32_t, Raul::Atom>       signal_voice_value; 
-	sigc::signal<void, string>                             signal_port_activity; 
+	sigc::signal<void, string>                             signal_activity; 
 	sigc::signal<void, string, uint32_t, uint32_t, string> signal_program_add; 
 	sigc::signal<void, string, uint32_t, uint32_t>         signal_program_remove; 
 	
@@ -139,8 +139,8 @@ protected:
 	void set_voice_value(const string& port_path, uint32_t voice, const Raul::Atom& value)
 		{ if (_enabled) signal_voice_value.emit(port_path, voice, value); }
 	
-	void port_activity(const string& port_path)
-		{ if (_enabled) signal_port_activity.emit(port_path); }
+	void activity(const string& port_path)
+		{ if (_enabled) signal_activity.emit(port_path); }
 
 	void program_add(const string& path, uint32_t bank, uint32_t program, const string& name)
 		{ if (_enabled) signal_program_add.emit(path, bank, program, name); }
diff --git a/src/client/ThreadedSigClientInterface.hpp b/src/client/ThreadedSigClientInterface.hpp
index f31ba37b..968954bc 100644
--- a/src/client/ThreadedSigClientInterface.hpp
+++ b/src/client/ThreadedSigClientInterface.hpp
@@ -64,7 +64,7 @@ public:
 	, variable_change_slot(signal_variable_change.make_slot())
 	, property_change_slot(signal_property_change.make_slot())
 	, port_value_slot(signal_port_value.make_slot())
-	, port_activity_slot(signal_port_activity.make_slot())
+	, activity_slot(signal_activity.make_slot())
 	, program_add_slot(signal_program_add.make_slot())
 	, program_remove_slot(signal_program_remove.make_slot())
 	{
@@ -133,8 +133,8 @@ public:
 	void set_voice_value(const string& port_path, uint32_t voice, const Raul::Atom& value)
 		{ push_sig(sigc::bind(voice_value_slot, port_path, voice, value)); }
 	
-	void port_activity(const string& port_path)
-		{ push_sig(sigc::bind(port_activity_slot, port_path)); }
+	void activity(const string& port_path)
+		{ push_sig(sigc::bind(activity_slot, port_path)); }
 
 	void program_add(const string& path, uint32_t bank, uint32_t program, const string& name)
 		{ push_sig(sigc::bind(program_add_slot, path, bank, program, name)); }
@@ -172,7 +172,7 @@ private:
 	sigc::slot<void, string, string, Raul::Atom>         property_change_slot; 
 	sigc::slot<void, string, Raul::Atom>                 port_value_slot; 
 	sigc::slot<void, string, uint32_t, Raul::Atom>       voice_value_slot; 
-	sigc::slot<void, string>                             port_activity_slot; 
+	sigc::slot<void, string>                             activity_slot; 
 	sigc::slot<void, string, uint32_t, uint32_t, string> program_add_slot; 
 	sigc::slot<void, string, uint32_t, uint32_t>         program_remove_slot; 
 };