Cloooser...

Bundling of OSC communication both ways (previous was just engine->client).
Factor out common OSC*Sender functionality (bundling stuff).
Fully type-safe and polyphony-aware port value setting/getting, from RDF through OSC through engine and back again.


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

13 files changed, 125 insertions, 215 deletions

diff --git a/src/libs/engine/ClientBroadcaster.cpp b/src/libs/engine/ClientBroadcaster.cpp
index 7e54f994..949784c3 100644
--- a/src/libs/engine/ClientBroadcaster.cpp
+++ b/src/libs/engine/ClientBroadcaster.cpp
@@ -235,10 +235,10 @@ ClientBroadcaster::send_variable_change(const string& node_path, const string& k
  * forcing clients to ignore things to avoid feedback loops etc).
  */
 void
-ClientBroadcaster::send_control_change(const string& port_path, float value)
+ClientBroadcaster::send_port_value(const string& port_path, const Raul::Atom& value)
 {
 	for (Clients::const_iterator i = _clients.begin(); i != _clients.end(); ++i)
-		(*i).second->control_change(port_path, value);
+		(*i).second->set_port_value(port_path, value);
 }
 
 
diff --git a/src/libs/engine/ClientBroadcaster.hpp b/src/libs/engine/ClientBroadcaster.hpp
index 800f6950..323a1912 100644
--- a/src/libs/engine/ClientBroadcaster.hpp
+++ b/src/libs/engine/ClientBroadcaster.hpp
@@ -80,7 +80,7 @@ public:
 	void send_patch_disable(const string& patch_path);
 	void send_patch_polyphony(const string& patch_path, uint32_t poly);
 	void send_variable_change(const string& node_path, const string& key, const Raul::Atom& value);
-	void send_control_change(const string& port_path, float value);
+	void send_port_value(const string& port_path, const Raul::Atom& value);
 	void send_port_activity(const string& port_path);
 	void send_program_add(const string& node_path, int bank, int program, const string& name);
 	void send_program_remove(const string& node_path, int bank, int program);
diff --git a/src/libs/engine/OSCClientSender.cpp b/src/libs/engine/OSCClientSender.cpp
index c9b156db..60edbe07 100644
--- a/src/libs/engine/OSCClientSender.cpp
+++ b/src/libs/engine/OSCClientSender.cpp
@@ -35,90 +35,6 @@ using namespace std;
 
 namespace Ingen {
 
-void
-OSCClientSender::bundle_begin()
-{
-	assert(!_transfer);
-	_transfer = lo_bundle_new(LO_TT_IMMEDIATE);
-	_send_state = SendingBundle;
-
-}
-
-void
-OSCClientSender::bundle_end()
-{
-	transfer_end();
-}
-
-
-void
-OSCClientSender::transfer_begin()
-{
-	//cerr << "TRANSFER {" << endl;
-	assert(!_transfer);
-	_transfer = lo_bundle_new(LO_TT_IMMEDIATE);
-	_send_state = SendingTransfer;
-}
-
-
-void
-OSCClientSender::transfer_end()
-{
-	//cerr << "} TRANSFER" << endl;
-	assert(_transfer);
-	lo_send_bundle(_address, _transfer);
-	lo_bundle_free(_transfer);
-	_transfer = NULL;
-	_send_state = Immediate;
-}
-
-
-int
-OSCClientSender::send(const char *path, const char *types, ...)
-{
-	if (!_enabled)
-		return 0;
-
-	va_list args;
-	va_start(args, types);
-	
-	lo_message msg = lo_message_new();
-	int ret = lo_message_add_varargs(msg, types, args);
-    
-	if (!ret)
-		send_message(path, msg);
-    
-	va_end(args);
-
-	return ret;
-}
-
-
-void
-OSCClientSender::send_message(const char* path, lo_message msg)
-{
-	// FIXME: size?  liblo doesn't export this.
-	// Don't want to exceed max UDP packet size (1500 bytes?)
-	static const size_t MAX_BUNDLE_SIZE = 1500 - 32*5;
-
-	if (!_enabled)
-		return;
-		
-	if (_transfer) {
-		if (lo_bundle_length(_transfer) + lo_message_length(msg, path) > MAX_BUNDLE_SIZE) {
-			if (_send_state == SendingBundle)
-				cerr << "WARNING: Maximum bundle size reached, bundle split" << endl;
-			lo_send_bundle(_address, _transfer);
-			_transfer = lo_bundle_new(LO_TT_IMMEDIATE);
-		}
-		lo_bundle_add_message(_transfer, path, msg);
-
-	} else {
-		lo_send_message(_address, path, msg);
-	}
-}
-
-
 
 /*! \page client_osc_namespace Client OSC Namespace Documentation
  *
@@ -460,14 +376,33 @@ OSCClientSender::set_variable(const std::string& path, const std::string& key, c
 
 
 /** \page client_osc_namespace
- * <p> \b /ingen/control_change - Notification the value of a port has changed
+ * <p> \b /ingen/set_port_value - Notification the value of a port has changed
+ * \arg \b path (string) - Path of port
+ * \arg \b value (any) - New value of port </p> \n \n
+ */
+void
+OSCClientSender::set_port_value(const std::string& port_path, const Raul::Atom& value)
+{
+	lo_message m = lo_message_new();
+	lo_message_add_string(m, port_path.c_str());
+	Raul::AtomLiblo::lo_message_add_atom(m, value);
+	send_message("/ingen/set_port_value", m);
+}
+
+
+/** \page client_osc_namespace
+ * <p> \b /ingen/set_port_value - Notification the value of a port has changed
  * \arg \b path (string) - Path of port
- * \arg \b value (float) - New value of port </p> \n \n
+ * \arg \b voice (int) - Voice which is set to this value
+ * \arg \b value (any) - New value of port </p> \n \n
  */
 void
-OSCClientSender::control_change(const std::string& port_path, float value)
+OSCClientSender::set_voice_value(const std::string& port_path, uint32_t voice, const Raul::Atom& value)
 {
-	send("/ingen/control_change", "sf", port_path.c_str(), value, LO_ARGS_END);
+	lo_message m = lo_message_new();
+	lo_message_add_string(m, port_path.c_str());
+	Raul::AtomLiblo::lo_message_add_atom(m, value);
+	send_message("/ingen/set_port_value", m);
 }
 
 
diff --git a/src/libs/engine/OSCClientSender.hpp b/src/libs/engine/OSCClientSender.hpp
index cc700e38..662ba75b 100644
--- a/src/libs/engine/OSCClientSender.hpp
+++ b/src/libs/engine/OSCClientSender.hpp
@@ -21,15 +21,11 @@
 #include <cassert>
 #include <string>
 #include <iostream>
-#include <list>
 #include <lo/lo.h>
 #include <pthread.h>
 #include "types.hpp"
 #include "interface/ClientInterface.hpp"
-
-using std::list;
-using std::string;
-using std::cerr;
+#include "shared/OSCSender.hpp"
 
 namespace Ingen {
 
@@ -40,35 +36,33 @@ namespace Shared { class EngineInterface; }
  *
  * \ingroup engine
  */
-class OSCClientSender : public Shared::ClientInterface
+class OSCClientSender : public Shared::ClientInterface, public Shared::OSCSender
 {
 public:
 	OSCClientSender(const std::string& url)
-		: Shared::ClientInterface(url)
-		, _address(lo_address_new_from_url(url.c_str()))
-		, _transfer(NULL)
-		, _enabled(true)
-	{}
+		: _url(url)
+	{
+		_address = lo_address_new_from_url(url.c_str());
+	}
 
 	virtual ~OSCClientSender()
 	{ lo_address_free(_address); }
-
-	lo_address address() const { return _address; }
+	
+	void enable()  { _enabled = true; }
+	void disable() { _enabled = false; }
+	
+	void bundle_begin()   { OSCSender::bundle_begin(); }
+	void bundle_end()     { OSCSender::bundle_end(); }
+	void transfer_begin() { OSCSender::transfer_begin(); }
+	void transfer_end()   { OSCSender::transfer_end(); }
+	
+	std::string uri() const { return lo_address_get_url(_address); }
 	
     void subscribe(Shared::EngineInterface* engine) { }
 
 	/* *** ClientInterface Implementation Below *** */
 
 	//void client_registration(const std::string& url, int client_id);
-	
-	void enable()  { _enabled = true; }
-	void disable() { _enabled = false; }
-
-	void bundle_begin();
-	void bundle_end();
-	
-	void transfer_begin();
-	void transfer_end();
 
 	void response_ok(int32_t id);
 	void response_error(int32_t id, const std::string& msg);
@@ -120,8 +114,12 @@ public:
 	                          const std::string& predicate,
 	                          const Raul::Atom&  value);
 	
-	virtual void control_change(const std::string& port_path,
-	                            float              value);
+	virtual void set_port_value(const std::string& port_path,
+	                            const Raul::Atom&  value);
+	
+	virtual void set_voice_value(const std::string& port_path,
+	                             uint32_t           voice,
+	                             const Raul::Atom&  value);
 	
 	virtual void port_activity(const std::string& port_path);
 	
@@ -135,16 +133,7 @@ public:
 	                            uint32_t           program);
 
 private:
-	int  send(const char *path, const char *types, ...);
-	void send_message(const char* path, lo_message m);
-
-	enum SendState { Immediate, SendingBundle, SendingTransfer };
-
-	string     _url;
-	lo_address _address;
-	SendState  _send_state;
-	lo_bundle  _transfer;
-	bool       _enabled;
+	std::string _url;
 };
 
 
diff --git a/src/libs/engine/OSCEngineReceiver.cpp b/src/libs/engine/OSCEngineReceiver.cpp
index 4cc27367..74947917 100644
--- a/src/libs/engine/OSCEngineReceiver.cpp
+++ b/src/libs/engine/OSCEngineReceiver.cpp
@@ -639,23 +639,26 @@ OSCEngineReceiver::_set_port_value_immediate_cb(const char* path, const char* ty
 		return 1;
 	
 	const char* port_path = &argv[1]->s;
+	using Raul::Atom;
 
 	if (!strcmp(types, "isf")) { // float, all voices
 		const float value = argv[2]->f;
-		set_port_value_immediate(port_path, "ingen:Float", sizeof(float), &value);
+		set_port_value_immediate(port_path, Atom(value));
 	} else if (!strcmp(types, "isif")) { // float, specific voice
 		const float value = argv[3]->f;
-		set_port_value_immediate(port_path, "ingen:Float", argv[2]->i, sizeof(float), &value);
+		set_voice_value_immediate(port_path, argv[2]->i, Atom(value));
 	} else if (!strcmp(types, "issb")) { // blob (event), all voices
+		const char* type = &argv[2]->s;
 		lo_blob b = argv[3];
 		size_t data_size = lo_blob_datasize(b);
 		void* data = lo_blob_dataptr(b);
-		set_port_value_immediate(port_path, &argv[2]->s, data_size, data);
+		set_port_value_immediate(port_path, Atom(type, data_size, data));
 	} else if (!strcmp(types, "isisb")) { // blob (event), specific voice
+		const char* type = &argv[3]->s;
 		lo_blob b = argv[4];
 		size_t data_size = lo_blob_datasize(b);
 		void* data = lo_blob_dataptr(b);
-		set_port_value_immediate(port_path, &argv[3]->s, argv[2]->i, data_size, data);
+		set_voice_value_immediate(port_path, argv[2]->i, Atom(type, data_size, data));
 	} else {
 		return 1;
 	}
@@ -700,22 +703,26 @@ OSCEngineReceiver::_set_port_value_cb(const char* path, const char* types, lo_ar
 	
 	const char* port_path = &argv[1]->s;
 
+	using Raul::Atom;
+
 	if (!strcmp(types, "isf")) { // float, all voices
 		const float value = argv[2]->f;
-		set_port_value_immediate(port_path, "ingen:Float", sizeof(float), &value);
+		set_port_value_immediate(port_path, Atom(value));
 	} else if (!strcmp(types, "isif")) { // float, specific voice
 		const float value = argv[3]->f;
-		set_port_value_immediate(port_path, "ingen:Float", argv[2]->i, sizeof(float), &value);
+		set_voice_value_immediate(port_path, argv[2]->i, Atom(value));
 	} else if (!strcmp(types, "issb")) { // blob (event), all voices
+		const char* type = &argv[2]->s;
 		lo_blob b = argv[3];
 		size_t data_size = lo_blob_datasize(b);
 		void* data = lo_blob_dataptr(b);
-		set_port_value_immediate(port_path, &argv[2]->s, data_size, data);
+		set_port_value_immediate(port_path, Atom(type, data_size, data));
 	} else if (!strcmp(types, "isisb")) { // blob (event), specific voice
+		const char* type = &argv[3]->s;
 		lo_blob b = argv[4];
 		size_t data_size = lo_blob_datasize(b);
 		void* data = lo_blob_dataptr(b);
-		set_port_value_immediate(port_path, &argv[3]->s, argv[2]->i, data_size, data);
+		set_voice_value_immediate(port_path, argv[2]->i, Atom(type, data_size, data));
 	} else {
 		return 1;
 	}
diff --git a/src/libs/engine/ObjectSender.cpp b/src/libs/engine/ObjectSender.cpp
index 5e456dbc..0bbe6533 100644
--- a/src/libs/engine/ObjectSender.cpp
+++ b/src/libs/engine/ObjectSender.cpp
@@ -132,7 +132,7 @@ ObjectSender::send_port(ClientInterface* client, const PortImpl* port)
 	if (port->type() == DataType::CONTROL) {
 		const Sample value = dynamic_cast<const AudioBuffer*>(port->buffer(0))->value_at(0);
 		//cerr << port->path() << " sending default value " << default_value << endl;
-		client->control_change(port->path(), value);
+		client->set_port_value(port->path(), value);
 	}
 	
 	client->bundle_end();
diff --git a/src/libs/engine/QueuedEngineInterface.cpp b/src/libs/engine/QueuedEngineInterface.cpp
index 65f2c14b..6593ec81 100644
--- a/src/libs/engine/QueuedEngineInterface.cpp
+++ b/src/libs/engine/QueuedEngineInterface.cpp
@@ -250,44 +250,36 @@ QueuedEngineInterface::disconnect_all(const string& patch_path,
 
 
 void
-QueuedEngineInterface::set_port_value(const string& port_path,
-	                                  const string& type_uri,
-                                      uint32_t      data_size,
-                                      const void*   data)
+QueuedEngineInterface::set_port_value(const string&     port_path,
+                                      const Raul::Atom& value)
 {
-	push_queued(new SetPortValueEvent(_engine, _responder, true, now(), port_path, type_uri, data_size, data));
+	push_queued(new SetPortValueEvent(_engine, _responder, true, now(), port_path, value));
 }
 
 
 void
-QueuedEngineInterface::set_port_value(const string& port_path,
-	                                  const string& type_uri,
-                                      uint32_t      voice,
-                                      uint32_t      data_size,
-                                      const void*   data)
+QueuedEngineInterface::set_voice_value(const string&     port_path,
+                                       uint32_t          voice,
+                                       const Raul::Atom& value)
 {
-	push_queued(new SetPortValueEvent(_engine, _responder, true, now(), voice, port_path, type_uri, data_size, data));
+	push_queued(new SetPortValueEvent(_engine, _responder, true, now(), voice, port_path, value));
 }
 
 
 void
-QueuedEngineInterface::set_port_value_immediate(const string& port_path,
-	                                            const string& type_uri,
-                                                uint32_t      data_size,
-                                                const void*   data)
+QueuedEngineInterface::set_port_value_immediate(const string&     port_path,
+                                                const Raul::Atom& value)
 {
-	push_stamped(new SetPortValueEvent(_engine, _responder, false, now(), port_path, type_uri, data_size, data));
+	push_stamped(new SetPortValueEvent(_engine, _responder, false, now(), port_path, value));
 }
 
 
 void
-QueuedEngineInterface::set_port_value_immediate(const string& port_path,
-	                                            const string& type_uri,
-                                                uint32_t      voice,
-                                                uint32_t      data_size,
-                                                const void*   data)
+QueuedEngineInterface::set_voice_value_immediate(const string&     port_path,
+                                                 uint32_t          voice,
+                                                 const Raul::Atom& value)
 {
-	push_stamped(new SetPortValueEvent(_engine, _responder, false, now(), voice, port_path, type_uri, data_size, data));
+	push_stamped(new SetPortValueEvent(_engine, _responder, false, now(), voice, port_path, value));
 }
 
 
diff --git a/src/libs/engine/QueuedEngineInterface.hpp b/src/libs/engine/QueuedEngineInterface.hpp
index 280e9b4e..3d8b8d12 100644
--- a/src/libs/engine/QueuedEngineInterface.hpp
+++ b/src/libs/engine/QueuedEngineInterface.hpp
@@ -123,27 +123,19 @@ public:
 	virtual void disconnect_all(const string& patch_path,
 	                            const string& node_path);
 
-	virtual void set_port_value(const string& port_path,
-	                            const string& type_uri,
-	                            uint32_t      data_size,
-	                            const void*   data);
+	virtual void set_port_value(const string&     port_path,
+	                            const Raul::Atom& value);
 	
-	virtual void set_port_value(const string& port_path,
-	                            const string& type_uri,
-	                            uint32_t      voice,
-	                            uint32_t      data_size,
-	                            const void*   data);
+	virtual void set_voice_value(const string&     port_path,
+	                             uint32_t          voice,
+	                             const Raul::Atom& value);
 	
 	virtual void set_port_value_immediate(const string& port_path,
-	                                      const string& type_uri,
-	                                      uint32_t      data_size,
-	                                      const void*   data);
+	                                      const Raul::Atom& value);
 	
-	virtual void set_port_value_immediate(const string& port_path,
-	                                      const string& type_uri,
-	                                      uint32_t      voice,
-	                                      uint32_t      data_size,
-	                                      const void*   data);
+	virtual void set_voice_value_immediate(const string&     port_path,
+	                                       uint32_t          voice,
+	                                       const Raul::Atom& value);
 	
 	virtual void enable_port_broadcasting(const string& port_path);
 	
diff --git a/src/libs/engine/events/MidiLearnEvent.cpp b/src/libs/engine/events/MidiLearnEvent.cpp
index 45216e70..2f37f30d 100644
--- a/src/libs/engine/events/MidiLearnEvent.cpp
+++ b/src/libs/engine/events/MidiLearnEvent.cpp
@@ -32,7 +32,7 @@ namespace Ingen {
 void
 MidiLearnResponseEvent::post_process()
 {
-	_engine.broadcaster()->send_control_change(_port_path, _value);
+	_engine.broadcaster()->send_port_value(_port_path, _value);
 }
 
 
diff --git a/src/libs/engine/events/RequestPortValueEvent.cpp b/src/libs/engine/events/RequestPortValueEvent.cpp
index 20203f88..025d3700 100644
--- a/src/libs/engine/events/RequestPortValueEvent.cpp
+++ b/src/libs/engine/events/RequestPortValueEvent.cpp
@@ -70,7 +70,7 @@ RequestPortValueEvent::post_process()
 		_responder->respond_error("Unable to find port for get_value responder.");
 	} else if (_responder->client()) {
 		_responder->respond_ok();
-		_responder->client()->control_change(_port_path, _value);
+		_responder->client()->set_port_value(_port_path, _value);
 	} else {
 		_responder->respond_error("Unable to find client to send port value");
 	}
diff --git a/src/libs/engine/events/SendPortValueEvent.cpp b/src/libs/engine/events/SendPortValueEvent.cpp
index 89e8c9e0..d3fb0d36 100644
--- a/src/libs/engine/events/SendPortValueEvent.cpp
+++ b/src/libs/engine/events/SendPortValueEvent.cpp
@@ -32,9 +32,9 @@ SendPortValueEvent::post_process()
 	// FIXME...
 	
 	if (_omni) {
-		_engine.broadcaster()->send_control_change(_port->path(), _value);
+		_engine.broadcaster()->send_port_value(_port->path(), _value);
 	} else {
-		_engine.broadcaster()->send_control_change(_port->path(), _value);
+		_engine.broadcaster()->send_port_value(_port->path(), _value);
 	}
 }
 
diff --git a/src/libs/engine/events/SetPortValueEvent.cpp b/src/libs/engine/events/SetPortValueEvent.cpp
index 01263d2c..f69ba9a3 100644
--- a/src/libs/engine/events/SetPortValueEvent.cpp
+++ b/src/libs/engine/events/SetPortValueEvent.cpp
@@ -39,21 +39,16 @@ SetPortValueEvent::SetPortValueEvent(Engine&              engine,
                                      bool                 queued,
                                      SampleCount          timestamp,
                                      const string&        port_path,
-                                     const string&        data_type,
-                                     uint32_t             data_size,
-                                     const void*          data)
+                                     const Raul::Atom&    value)
 	: QueuedEvent(engine, responder, timestamp)
 	, _queued(queued)
 	, _omni(true)
 	, _voice_num(0)
 	, _port_path(port_path)
-	, _data_type(data_type)
-	, _data_size(data_size)
-	, _data(malloc(data_size))
+    , _value(value) 
 	, _port(NULL)
 	, _error(NO_ERROR)
 {
-	memcpy(_data, data, data_size);
 }
 
 
@@ -64,27 +59,21 @@ SetPortValueEvent::SetPortValueEvent(Engine&              engine,
                                      SampleCount          timestamp,
                                      uint32_t             voice_num,
                                      const string&        port_path,
-                                     const string&        data_type,
-                                     uint32_t             data_size,
-                                     const void*          data)
+                                     const Raul::Atom&    value)
 	: QueuedEvent(engine, responder, timestamp)
 	, _queued(queued)
 	, _omni(false)
 	, _voice_num(voice_num)
 	, _port_path(port_path)
-	, _data_type(data_type)
-	, _data_size(data_size)
-	, _data(malloc(data_size))
+    , _value(value) 
 	, _port(NULL)
 	, _error(NO_ERROR)
 {
-	memcpy(_data, data, data_size);
 }
 
 
 SetPortValueEvent::~SetPortValueEvent()
 {
-	free(_data);
 }
 
 	
@@ -129,13 +118,19 @@ SetPortValueEvent::execute(ProcessContext& context)
 		Buffer* const buf = _port->buffer(0);
 		AudioBuffer* const abuf = dynamic_cast<AudioBuffer*>(buf);
 		if (abuf) {
+			if (_value.type() != Atom::FLOAT) {
+				_error = TYPE_MISMATCH;
+				return;
+			}
+
 			if (_omni) {
 				for (uint32_t i=0; i < _port->poly(); ++i)
-					((AudioBuffer*)_port->buffer(i))->set_value(*(float*)_data, context.start(), _time);
+					((AudioBuffer*)_port->buffer(i))->set_value(
+							_value.get_float(), context.start(), _time);
 			} else {
 				if (_voice_num < _port->poly())
 					((AudioBuffer*)_port->buffer(_voice_num))->set_value(
-							*(float*)_data, context.start(), _time);
+							_value.get_float(), context.start(), _time);
 				else
 					_error = ILLEGAL_VOICE;
 			}
@@ -144,7 +139,8 @@ SetPortValueEvent::execute(ProcessContext& context)
 		
 		EventBuffer* const ebuf = dynamic_cast<EventBuffer*>(buf);
 		// FIXME: eliminate string comparisons
-		if (ebuf && _data_type == "lv2_midi:MidiEvent") {
+		if (ebuf && _value.type() == Atom::BLOB
+				&& !strcmp(_value.get_blob_type(), "lv2_midi:MidiEvent")) {
 			const LV2Features::Feature* f = _engine.world()->lv2_features->feature(LV2_URI_MAP_URI);
 			LV2URIMap* map = (LV2URIMap*)f->controller;
 			const uint32_t type_id = map->uri_to_id(NULL, "http://lv2plug.in/ns/ext/midi#MidiEvent");
@@ -152,14 +148,18 @@ SetPortValueEvent::execute(ProcessContext& context)
 			ebuf->prepare_write(context.start(), context.nframes());
 			// FIXME: how should this work? binary over OSC, ick
 			// Message is an event:
-			ebuf->append(frames, 0, type_id, _data_size, (const unsigned char*)_data);
+			ebuf->append(frames, 0, type_id, _value.data_size(), (const uint8_t*)_value.get_blob());
 			// Message is an event buffer:
 			//ebuf->append((LV2_Event_Buffer*)_data);
 			_port->raise_set_by_user_flag();
 			return;
 		}
 
-		cerr << "WARNING: Unknown value type " << _data_type << ", ignoring" << endl;
+
+		if (_value.type() == Atom::BLOB)
+			cerr << "WARNING: Unknown value blob type " << _value.get_blob_type() << endl;
+		else
+			cerr << "WARNING: Unknown value type " << (int)_value.type() << endl;
 	}
 }
 
@@ -170,7 +170,7 @@ SetPortValueEvent::post_process()
 	if (_error == NO_ERROR) {
 		assert(_port != NULL);
 		_responder->respond_ok();
-		_engine.broadcaster()->send_control_change(_port_path, *(float*)_data);
+		_engine.broadcaster()->send_port_value(_port_path, _value);
 	
 	} else if (_error == ILLEGAL_PATH) {
 		string msg = "Illegal port path \"";
@@ -189,7 +189,7 @@ SetPortValueEvent::post_process()
 	
 	} else if (_error == NO_SPACE) {
 		std::ostringstream msg("Attempt to write ");
-		msg << _data_size << " bytes to " << _port_path << ", with capacity "
+		msg << _value.data_size() << " bytes to " << _port_path << ", with capacity "
 			<< _port->buffer_size() << endl;
 		_responder->respond_error(msg.str());
 	}
diff --git a/src/libs/engine/events/SetPortValueEvent.hpp b/src/libs/engine/events/SetPortValueEvent.hpp
index a509af33..2fc68d9b 100644
--- a/src/libs/engine/events/SetPortValueEvent.hpp
+++ b/src/libs/engine/events/SetPortValueEvent.hpp
@@ -45,9 +45,7 @@ public:
 	                  bool                 queued,
 	                  SampleCount          timestamp,
 	                  const string&        port_path,
-	                  const string&        data_type,
-	                  uint32_t             data_size,
-	                  const void*          data);
+	                  const Raul::Atom&    value);
 	
 	SetPortValueEvent(Engine&              engine,
 	                  SharedPtr<Responder> responder,
@@ -55,9 +53,7 @@ public:
 	                  SampleCount          timestamp,
 	                  uint32_t             voice_num,
 	                  const string&        port_path,
-	                  const string&        data_type,
-	                  uint32_t             data_size,
-	                  const void*          data);
+	                  const Raul::Atom&    value);
 
 	~SetPortValueEvent();
 
@@ -66,17 +62,16 @@ public:
 	void post_process();
 
 private:
-	enum ErrorType { NO_ERROR, PORT_NOT_FOUND, NO_SPACE, ILLEGAL_PATH, ILLEGAL_VOICE };
+	enum ErrorType { NO_ERROR, PORT_NOT_FOUND, NO_SPACE,
+		ILLEGAL_PATH, ILLEGAL_VOICE, TYPE_MISMATCH };
 	
-	bool         _queued;
-	bool         _omni;
-	uint32_t     _voice_num;
-	const string _port_path;
-	const string _data_type;
-	uint32_t     _data_size;
-	void*        _data;
-	PortImpl*    _port;
-	ErrorType    _error;
+	bool             _queued;
+	bool             _omni;
+	uint32_t         _voice_num;
+	const string     _port_path;
+	const Raul::Atom _value;
+	PortImpl*        _port;
+	ErrorType        _error;
 };