Partial support for message/value ports and the message context.

This use case now works:
 - Add an event input and the "print" plugin from imum.lv2 to ingen
 - Connect the event input to the input of "print"
 - Hook Ingen up to JACK and play some MIDI events
   (or get events to the print plugin from anywhere else)
 - The "print" plugin will print the received events to the console
   in the message context (i.e. the audio thread is realtime safe)


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

23 files changed, 271 insertions, 64 deletions

diff --git a/src/engine/Buffer.cpp b/src/engine/Buffer.cpp
index 4b66b63f..15752846 100644
--- a/src/engine/Buffer.cpp
+++ b/src/engine/Buffer.cpp
@@ -35,7 +35,7 @@ Buffer::create(Engine& engine, Shared::PortType type, size_t size)
 		return new AudioBuffer(type, size);
 	else if (type.is_events())
 		return new EventBuffer(size);
-	else if (type.is_value())
+	else if (type.is_value() || type.is_message())
 		return new ObjectBuffer(std::max(size,
 					sizeof(LV2_Object) + sizeof(void*)));
 	else
diff --git a/src/engine/BufferFactory.cpp b/src/engine/BufferFactory.cpp
index cfbe72e6..d84cf658 100644
--- a/src/engine/BufferFactory.cpp
+++ b/src/engine/BufferFactory.cpp
@@ -97,7 +97,7 @@ BufferFactory::create(Shared::PortType type, size_t size)
 		if (size == 0)
 			size = _engine.audio_driver()->buffer_size() * 4; // FIXME
 		buffer = new EventBuffer(size);
-	} else if (type.is_value()) {
+	} else if (type.is_value() || type.is_message()) {
 		if (size == 0)
 			size = 32; // FIXME
 		buffer = new ObjectBuffer(std::max(size, sizeof(LV2_Object) + sizeof(void*)));
diff --git a/src/engine/BufferFactory.hpp b/src/engine/BufferFactory.hpp
index 8ec29697..25164db9 100644
--- a/src/engine/BufferFactory.hpp
+++ b/src/engine/BufferFactory.hpp
@@ -49,7 +49,8 @@ private:
 		case PortType::AUDIO:   return _free_audio;
 		case PortType::CONTROL: return _free_control;
 		case PortType::EVENTS:  return _free_event;
-		case PortType::VALUE:   return _free_object;
+		case PortType::VALUE:
+		case PortType::MESSAGE: return _free_object;
 		default: throw;
 		}
 	}
diff --git a/src/engine/ConnectionImpl.cpp b/src/engine/ConnectionImpl.cpp
index 24f42545..27786e00 100644
--- a/src/engine/ConnectionImpl.cpp
+++ b/src/engine/ConnectionImpl.cpp
@@ -18,12 +18,15 @@
 #include <algorithm>
 #include "raul/Maid.hpp"
 #include "util.hpp"
+#include "AudioBuffer.hpp"
+#include "BufferFactory.hpp"
 #include "ConnectionImpl.hpp"
+#include "Engine.hpp"
+#include "EventBuffer.hpp"
+#include "InputPort.hpp"
+#include "MessageContext.hpp"
 #include "PortImpl.hpp"
-#include "AudioBuffer.hpp"
 #include "ProcessContext.hpp"
-#include "InputPort.hpp"
-#include "BufferFactory.hpp"
 
 namespace Ingen {
 
@@ -35,7 +38,8 @@ using namespace Shared;
  * user (InputPort).
  */
 ConnectionImpl::ConnectionImpl(BufferFactory& bufs, PortImpl* src_port, PortImpl* dst_port)
-	: _bufs(bufs)
+	: _queue(NULL)
+	, _bufs(bufs)
 	, _src_port(src_port)
 	, _dst_port(dst_port)
 	, _pending_disconnection(false)
@@ -44,15 +48,14 @@ ConnectionImpl::ConnectionImpl(BufferFactory& bufs, PortImpl* src_port, PortImpl
 	assert(dst_port);
 	assert(src_port != dst_port);
 	assert(src_port->path() != dst_port->path());
-	assert(src_port->type() == dst_port->type()
-			|| ( (src_port->type() == PortType::CONTROL || src_port->type() == PortType::AUDIO)
-				&& (dst_port->type() == PortType::CONTROL || dst_port->type() == PortType::AUDIO) ));
 
-	/*assert((src_port->parent_node()->poly() == dst_port->parent_node()->poly())
-		|| (src_port->parent_node()->poly() == 1 || dst_port->parent_node()->poly() == 1));*/
-
-	if (must_mix())
+	if (must_mix() || must_queue())
 		_local_buffer = bufs.get(dst_port->type(), dst_port->buffer_size());
+
+	if (must_queue())
+		_queue = new Raul::RingBuffer<LV2_Object>(src_port->buffer_size() * 2);
+
+	dump();
 }
 
 
@@ -60,7 +63,8 @@ void
 ConnectionImpl::dump() const
 {
 	cerr << _src_port->path() << " -> " << _dst_port->path()
-		<< (must_mix() ? " MIX" : " DIRECT") << endl;
+		<< (must_mix()   ? " (MIX) " : " (DIRECT) ")
+		<< (must_queue() ? " (QUEUE)" : " (NOQUEUE)") << endl;
 }
 
 
@@ -96,6 +100,27 @@ ConnectionImpl::apply_poly(Raul::Maid& maid, uint32_t poly)
 void
 ConnectionImpl::process(Context& context)
 {
+	if (must_queue()) {
+		SharedPtr<EventBuffer> src_buf = PtrCast<EventBuffer>(_src_port->buffer(0));
+		if (!src_buf) {
+			cerr << "ERROR: Queued connection but source is not an EventBuffer" << endl;
+			return;
+		}
+
+		SharedPtr<ObjectBuffer> local_buf = PtrCast<ObjectBuffer>(_local_buffer);
+		if (!local_buf) {
+			cerr << "ERROR: Queued connection but source is not an EventBuffer" << endl;
+			return;
+		}
+
+		if (_queue->read_space()) {
+			LV2_Object obj;
+			_queue->full_peek(sizeof(LV2_Object), &obj);
+			_queue->full_read(sizeof(LV2_Object) + obj.size, local_buf->object());
+		}
+		return;
+	}
+
 	if (!must_mix())
 		return;
 
@@ -108,5 +133,33 @@ ConnectionImpl::process(Context& context)
 }
 
 
+void
+ConnectionImpl::queue(Context& context)
+{
+	if (!must_queue())
+		return;
+
+	SharedPtr<EventBuffer> src_buf = PtrCast<EventBuffer>(_src_port->buffer(0));
+	if (!src_buf) {
+		cerr << "ERROR: Queued connection but source is not an EventBuffer" << endl;
+		return;
+	}
+
+	while (src_buf->is_valid()) {
+		LV2_Object* obj = src_buf->get_object();
+		/*cout << _src_port->path() << " -> " << _dst_port->path()
+			<< " QUEUE OBJECT TYPE " << obj->type << ":";
+		for (size_t i = 0; i < obj->size; ++i)
+			cout << " " << std::hex << (int)obj->body[i];
+		cout << endl;*/
+
+		_queue->write(sizeof(LV2_Object) + obj->size, obj);
+		src_buf->increment();
+
+		context.engine().message_context()->run(_dst_port->parent_node());
+	}
+}
+
+
 } // namespace Ingen
 
diff --git a/src/engine/ConnectionImpl.hpp b/src/engine/ConnectionImpl.hpp
index 76224269..9daea1f1 100644
--- a/src/engine/ConnectionImpl.hpp
+++ b/src/engine/ConnectionImpl.hpp
@@ -24,6 +24,7 @@
 #include "raul/Deletable.hpp"
 #include "interface/PortType.hpp"
 #include "interface/Connection.hpp"
+#include "object.lv2/object.h"
 #include "PortImpl.hpp"
 #include "PortImpl.hpp"
 
@@ -62,6 +63,7 @@ public:
 	void pending_disconnection(bool b) { _pending_disconnection = b; }
 
 	void process(Context& context);
+	void queue(Context& context);
 
 	/** Get the buffer for a particular voice.
 	 * A Connection is smart - it knows the destination port requesting the
@@ -69,7 +71,7 @@ public:
 	 * voice in a mono->poly connection).
 	 */
 	inline SharedPtr<Buffer> buffer(uint32_t voice) const {
-		if (must_mix()) {
+		if (must_mix() || must_queue()) {
 			return _local_buffer;
 		} else if ( ! _src_port->polyphonic()) {
 			return _src_port->buffer(0);
@@ -85,9 +87,14 @@ public:
 	/** Returns true if this connection must mix down voices into a local buffer */
 	inline bool must_mix() const { return _src_port->poly() > _dst_port->poly(); }
 
+	/** Returns true if this connection crosses contexts and must buffer */
+	inline bool must_queue() const { return _src_port->context() != _dst_port->context(); }
+
 protected:
 	void dump() const;
 
+	Raul::RingBuffer<LV2_Object>* _queue;
+
 	BufferFactory&    _bufs;
 	PortImpl* const   _src_port;
 	PortImpl* const   _dst_port;
diff --git a/src/engine/Engine.cpp b/src/engine/Engine.cpp
index a12f6e2e..07529caf 100644
--- a/src/engine/Engine.cpp
+++ b/src/engine/Engine.cpp
@@ -185,6 +185,8 @@ Engine::activate(size_t parallelism)
 {
 	assert(_audio_driver);
 
+	_message_context->Thread::start();
+
 	if (!_midi_driver)
 		_midi_driver = new DummyMidiDriver();
 
diff --git a/src/engine/EventBuffer.hpp b/src/engine/EventBuffer.hpp
index b779a837..d75c91ba 100644
--- a/src/engine/EventBuffer.hpp
+++ b/src/engine/EventBuffer.hpp
@@ -60,6 +60,10 @@ public:
 		return _buf->get_event(frames, subframes, type, size, data);
 	}
 
+	LV2_Object* get_object() const {
+		return _buf->get_object();
+	}
+
 	inline bool append(uint32_t       frames,
 	                   uint32_t       subframes,
 	                   uint16_t       type,
diff --git a/src/engine/JackAudioDriver.cpp b/src/engine/JackAudioDriver.cpp
index 86f946da..edb1ba9a 100644
--- a/src/engine/JackAudioDriver.cpp
+++ b/src/engine/JackAudioDriver.cpp
@@ -21,21 +21,22 @@
 #include <iostream>
 #include <cstdlib>
 #include "raul/List.hpp"
+#include "AudioBuffer.hpp"
+#include "DuplexPort.hpp"
 #include "Engine.hpp"
-#include "util.hpp"
 #include "Event.hpp"
-#include "ThreadManager.hpp"
-#include "QueuedEvent.hpp"
 #include "EventSource.hpp"
-#include "PostProcessor.hpp"
+#include "EventSource.hpp"
+#include "JackMidiDriver.hpp"
+#include "MessageContext.hpp"
+#include "MidiDriver.hpp"
 #include "PatchImpl.hpp"
 #include "PortImpl.hpp"
-#include "MidiDriver.hpp"
-#include "DuplexPort.hpp"
-#include "EventSource.hpp"
-#include "AudioBuffer.hpp"
+#include "PostProcessor.hpp"
 #include "ProcessSlave.hpp"
-#include "JackMidiDriver.hpp"
+#include "QueuedEvent.hpp"
+#include "ThreadManager.hpp"
+#include "util.hpp"
 
 using namespace std;
 using namespace Raul;
@@ -343,16 +344,9 @@ JackAudioDriver::_process_cb(jack_nframes_t nframes)
 	// FIXME: support nframes != buffer_size, even though that never damn well happens
 	assert(nframes == _buffer_size / sizeof(Sample));
 
-	// Jack can elect to not call this function for a cycle, if overloaded
-	// FIXME: this doesn't make sense, and the start time isn't used anyway
+	// Note that Jack can not call this function for a cycle, if overloaded
 	const jack_nframes_t start_of_current_cycle = jack_last_frame_time(_client);
-
-	const jack_nframes_t end_of_current_cycle = start_of_current_cycle + nframes;
-#ifndef NDEBUG
-	// FIXME: support changing cycle length
-	const jack_nframes_t start_of_last_cycle = start_of_current_cycle - nframes;
-	assert(start_of_current_cycle - start_of_last_cycle == nframes);
-#endif
+	const jack_nframes_t end_of_current_cycle   = start_of_current_cycle + nframes;
 
 	_transport_state = jack_transport_query(_client, &_position);
 
@@ -378,6 +372,10 @@ JackAudioDriver::_process_cb(jack_nframes_t nframes)
 	if (_root_patch)
 		_root_patch->process(_process_context);
 
+	// Signal message context to run if necessary
+	if (_engine.message_context()->has_requests())
+		_engine.message_context()->signal();
+
 	if (_engine.midi_driver())
 		_engine.midi_driver()->post_process(_process_context);
 
diff --git a/src/engine/LV2BlobFeature.hpp b/src/engine/LV2BlobFeature.hpp
index c672b7ac..13b6753e 100644
--- a/src/engine/LV2BlobFeature.hpp
+++ b/src/engine/LV2BlobFeature.hpp
@@ -39,10 +39,10 @@ struct BlobFeature : public Shared::LV2Features::Feature {
 	                     LV2_Reference*        reference,
 	                     LV2_Blob_Destroy      destroy_func,
 	                     uint32_t              type,
-	                     uint32_t              size) {}
+	                     size_t                size) {}
 
 	static LV2_Blob* reference_get(LV2_Blob_Support_Data data,
-	                               LV2_Reference*    ref) { return 0; }
+	                               LV2_Reference*        ref) { return 0; }
 
 	static void reference_copy(LV2_Blob_Support_Data data,
 	                           LV2_Reference*        dst,
diff --git a/src/engine/LV2EventBuffer.cpp b/src/engine/LV2EventBuffer.cpp
index 3f11790e..20df73c4 100644
--- a/src/engine/LV2EventBuffer.cpp
+++ b/src/engine/LV2EventBuffer.cpp
@@ -120,6 +120,20 @@ LV2EventBuffer::get_event(uint32_t* frames,
 }
 
 
+/** Get the object currently pointed to, or NULL if invalid.
+ */
+LV2_Object*
+LV2EventBuffer::get_object() const
+{
+	if (lv2_event_is_valid(&_iter)) {
+		uint8_t* data;
+		LV2_Event* ev = lv2_event_get(&_iter, &data);
+		return LV2_OBJECT_FROM_EVENT(ev);
+	}
+	return NULL;
+}
+
+
 /** Append an event to the buffer.
  *
  * \a timestamp must be >= the latest event in the buffer.
diff --git a/src/engine/LV2EventBuffer.hpp b/src/engine/LV2EventBuffer.hpp
index ce75d7cb..61cdba4c 100644
--- a/src/engine/LV2EventBuffer.hpp
+++ b/src/engine/LV2EventBuffer.hpp
@@ -20,6 +20,7 @@
 
 #include "event.lv2/event.h"
 #include "event.lv2/event-helpers.h"
+#include "object.lv2/object.h"
 
 namespace Ingen {
 
@@ -58,6 +59,8 @@ public:
 	               uint16_t* size,
 	               uint8_t** data) const;
 
+	LV2_Object* get_object() const;
+
 	bool append(uint32_t       frames,
 	            uint32_t       subframes,
 	            uint16_t       type,
diff --git a/src/engine/LV2Info.cpp b/src/engine/LV2Info.cpp
index 99c2ff85..93f4b386 100644
--- a/src/engine/LV2Info.cpp
+++ b/src/engine/LV2Info.cpp
@@ -38,6 +38,7 @@ LV2Info::LV2Info(Ingen::Shared::World* world)
 	, audio_class(slv2_value_new_uri(world->slv2_world, SLV2_PORT_CLASS_AUDIO))
 	, event_class(slv2_value_new_uri(world->slv2_world, SLV2_PORT_CLASS_EVENT))
 	, value_port_class(slv2_value_new_uri(world->slv2_world, LV2_OBJECT_URI "#ValuePort"))
+	, message_port_class(slv2_value_new_uri(world->slv2_world, LV2_OBJECT_URI "#MessagePort"))
 	, _world(world)
 {
 	assert(world);
@@ -58,6 +59,7 @@ LV2Info::~LV2Info()
 	slv2_value_free(audio_class);
 	slv2_value_free(event_class);
 	slv2_value_free(value_port_class);
+	slv2_value_free(message_port_class);
 }
 
 } // namespace Ingen
diff --git a/src/engine/LV2Info.hpp b/src/engine/LV2Info.hpp
index 76dfefa7..2a63f983 100644
--- a/src/engine/LV2Info.hpp
+++ b/src/engine/LV2Info.hpp
@@ -51,6 +51,7 @@ public:
 	SLV2Value audio_class;
 	SLV2Value event_class;
 	SLV2Value value_port_class;
+	SLV2Value message_port_class;
 
 	Ingen::Shared::World& world()     { return *_world; }
 	SLV2World             lv2_world() { return _world->slv2_world; }
diff --git a/src/engine/LV2Node.cpp b/src/engine/LV2Node.cpp
index dff994b9..e2f91927 100644
--- a/src/engine/LV2Node.cpp
+++ b/src/engine/LV2Node.cpp
@@ -189,13 +189,14 @@ LV2Node::instantiate(BufferFactory& bufs)
 	SLV2Value context_pred = slv2_value_new_uri(info->lv2_world(),
 			"http://lv2plug.in/ns/dev/contexts#context");
 
-	// FIXME: Why doesn't this just use lv2:default?
 	SLV2Value default_pred = slv2_value_new_uri(info->lv2_world(),
-			"http://lv2plug.in/ns/dev/string-port#default");
+			"http://lv2plug.in/ns/lv2core#default");
 
-	// FIXME: Make this a separate extension
-	SLV2Value size_pred = slv2_value_new_uri(info->lv2_world(),
-			"http://lv2plug.in/ns/dev/string-port#requiredSpace");
+	SLV2Value min_size_pred = slv2_value_new_uri(info->lv2_world(),
+			"http://lv2plug.in/ns/dev/resize-port#minimumSize");
+
+	//SLV2Value as_large_as_pred = slv2_value_new_uri(info->lv2_world(),
+	//		"http://lv2plug.in/ns/dev/resize-port#asLargeAs");
 
 	for (uint32_t j=0; j < num_ports; ++j) {
 		SLV2Port id = slv2_plugin_get_port_by_index(plug, j);
@@ -219,6 +220,11 @@ LV2Node::instantiate(BufferFactory& bufs)
 			port_buffer_size = _buffer_size;
 		} else if (slv2_port_is_a(plug, id, info->value_port_class)) {
 			data_type = PortType::VALUE;
+		} else if (slv2_port_is_a(plug, id, info->message_port_class)) {
+			data_type = PortType::MESSAGE;
+		}
+
+		if (data_type == PortType::VALUE || data_type == PortType::MESSAGE) {
 			port_buffer_size = 0;
 
 			// Get default value, and its length
@@ -236,7 +242,7 @@ LV2Node::instantiate(BufferFactory& bufs)
 			}
 
 			// Get minimum size, if set in data
-			SLV2Values sizes = slv2_port_get_value(plug, id, size_pred);
+			SLV2Values sizes = slv2_port_get_value(plug, id, min_size_pred);
 			for (uint32_t i = 0; i < slv2_values_size(sizes); ++i) {
 				SLV2Value d = slv2_values_get_at(sizes, i);
 				if (slv2_value_is_int(d)) {
diff --git a/src/engine/MessageContext.cpp b/src/engine/MessageContext.cpp
index ba03088d..5a180a10 100644
--- a/src/engine/MessageContext.cpp
+++ b/src/engine/MessageContext.cpp
@@ -24,20 +24,68 @@
 #include "PatchImpl.hpp"
 #include "PortImpl.hpp"
 #include "ProcessContext.hpp"
+#include "ThreadManager.hpp"
 
 using namespace std;
 
 namespace Ingen {
 
+
 void
 MessageContext::run(NodeImpl* node)
 {
+	if (ThreadManager::current_thread_id() == THREAD_PRE_PROCESS) {
+		assert(node);
+		Glib::Mutex::Lock lock(_mutex);
+		_request = node;
+		_sem.post();
+		_cond.wait(_mutex);
+	} else if (ThreadManager::current_thread_id() == THREAD_PROCESS) {
+		_requests.write(sizeof(NodeImpl*), &node);
+		// signal() will be called at the end of this process cycle
+	} else if (ThreadManager::current_thread_id() == THREAD_MESSAGE) {
+		cout << "Message context recursion at " << node->path() << endl;
+	} else {
+		cout << "[MessageContext] ERROR: Run requested from unknown thread" << endl;
+	}
+}
+
+
+void
+MessageContext::_run()
+{
+	NodeImpl* node = NULL;
+
+	while (true) {
+		_sem.wait();
+
+		// Run a node requested by the pre-process thread
+		{
+			Glib::Mutex::Lock lock(_mutex);
+			node = _request.get();
+			if (node) {
+				_cond.broadcast(); // Notify caller we got the message
+				run_node(node);
+			}
+		}
+
+		// Run nodes requested by the audio thread
+		while (has_requests()) {
+			_requests.full_read(sizeof(NodeImpl*), &node);
+			run_node(node);
+		}
+	}
+}
+
+
+void
+MessageContext::run_node(NodeImpl* node)
+{
 	node->message_run(*this);
 
 	void*      valid_ports = node->valid_ports();
 	PatchImpl* patch       = node->parent_patch();
 
-	//cout << "MESSAGE RUN " << node->path() << " {" << endl;
 	for (uint32_t i = 0; i < node->num_ports(); ++i) {
 		PortImpl* p = node->port_impl(i);
 		if (p->is_output() && p->context() == Context::MESSAGE &&
@@ -47,12 +95,11 @@ MessageContext::run(NodeImpl* node)
 				ConnectionImpl* ci = dynamic_cast<ConnectionImpl*>(c->get());
 				if (ci->src_port() == p) {
 					ci->dst_port()->pre_process(*_engine.message_context());
-					run(ci->dst_port()->parent_node());
+					run_node(ci->dst_port()->parent_node());
 				}
 			}
 		}
 	}
-	//cout << "}" << endl;
 
 	node->reset_valid_ports();
 }
diff --git a/src/engine/MessageContext.hpp b/src/engine/MessageContext.hpp
index 9d17d920..c871ad1c 100644
--- a/src/engine/MessageContext.hpp
+++ b/src/engine/MessageContext.hpp
@@ -18,7 +18,14 @@
 #ifndef MESSAGECONTEXT_H
 #define MESSAGECONTEXT_H
 
+#include <glibmm/thread.h>
+#include "raul/Thread.hpp"
+#include "raul/Semaphore.hpp"
+#include "raul/AtomicPtr.hpp"
+#include "object.lv2/object.h"
 #include "Context.hpp"
+#include "ThreadManager.hpp"
+#include "tuning.hpp"
 
 namespace Ingen {
 
@@ -33,14 +40,41 @@ class NodeImpl;
  *
  * \ingroup engine
  */
-class MessageContext : public Context
+class MessageContext : public Context, public Raul::Thread
 {
 public:
 	MessageContext(Engine& engine)
 		: Context(engine, MESSAGE)
-	{}
+		, Raul::Thread("message-context")
+		, _sem(0)
+		, _requests(message_context_queue_size)
+	{
+		Thread::set_context(THREAD_MESSAGE);
+	}
 
+	/** Request a run starting at node.
+	 *
+	 * Safe to call from either process thread or pre-process thread.
+	 */
 	void run(NodeImpl* node);
+
+	inline void signal() { _sem.post(); }
+	inline bool has_requests() const {
+		return _requests.read_space() >= sizeof(NodeImpl*) || _request.get();
+	}
+
+protected:
+	/** Thread run method (wait for and execute requests from process thread */
+	void _run();
+
+	/** Actually execute and propagate from node */
+	void run_node(NodeImpl* node);
+
+	Raul::Semaphore             _sem;
+	Raul::RingBuffer<NodeImpl*> _requests;
+	Glib::Mutex                 _mutex;
+	Glib::Cond                  _cond;
+	Raul::AtomicPtr<NodeImpl>   _request;
 };
 
 
diff --git a/src/engine/PatchImpl.cpp b/src/engine/PatchImpl.cpp
index 6388f648..6958c8b5 100644
--- a/src/engine/PatchImpl.cpp
+++ b/src/engine/PatchImpl.cpp
@@ -144,19 +144,22 @@ PatchImpl::process(ProcessContext& context)
 
 	NodeBase::pre_process(context);
 
-	/*if (_ports)
-		for (size_t i=0; i < _ports->size(); ++i)
-			if (_ports->at(i)->is_input() && _ports->at(i)->type() == PortType::MIDI)
-				cerr << _ports->at(i)->path() << " "
-					<< _ports->at(i)->buffer(0) << " # events: "
-					<< ((MidiBuffer*)_ports->at(i)->buffer(0))->event_count() << endl;*/
-
-	/* Run */
+	// Run all nodes
 	if (_compiled_patch && _compiled_patch->size() > 0) {
-		if (_engine.process_slaves().size() > 0)
+		if (_engine.process_slaves().size() > 0) {
 			process_parallel(context);
-		else
+		} else {
 			process_single(context);
+		}
+	}
+
+	// Queue any cross-context connections
+	for (Connections::iterator i = _connections.begin(); i != _connections.end(); ++i) {
+		ConnectionImpl* const c = (ConnectionImpl*)i->get();
+		if (c->src_port()->context() == Context::AUDIO &&
+				c->dst_port()->context() == Context::MESSAGE) {
+			c->queue(context);
+		}
 	}
 
 	NodeBase::post_process(context);
diff --git a/src/engine/PortImpl.cpp b/src/engine/PortImpl.cpp
index 23f7b5f8..45ee76da 100644
--- a/src/engine/PortImpl.cpp
+++ b/src/engine/PortImpl.cpp
@@ -18,6 +18,7 @@
 #include <iostream>
 #include "raul/Array.hpp"
 #include "raul/Maid.hpp"
+#include "contexts.lv2/contexts.h"
 #include "interface/PortType.hpp"
 #include "events/SendPortValue.hpp"
 #include "events/SendPortActivity.hpp"
@@ -75,7 +76,8 @@ PortImpl::PortImpl(BufferFactory&  bufs,
 	else
 		_polyphonic = true;
 
-	add_property("rdf:type", Atom(Atom::URI, type.uri()));
+	add_property("rdf:type",    Atom(Atom::URI, type.uri()));
+	set_context(_context);
 
 	if (type == PortType::EVENTS)
 		_broadcast = true; // send activity blips
@@ -188,6 +190,7 @@ PortImpl::broadcast_value(Context& context, bool force)
 		}
 		break;
 	case PortType::VALUE:
+	case PortType::MESSAGE:
 		LV2Object::to_atom(context.engine().world(), ((ObjectBuffer*)buffer(0).get())->object(), val);
 		break;
 	}
@@ -201,4 +204,19 @@ PortImpl::broadcast_value(Context& context, bool force)
 }
 
 
+void
+PortImpl::set_context(Context::ID c)
+{
+	_context = c;
+	switch (c) {
+	case Context::AUDIO:
+		set_property("ctx:context", Atom(Atom::URI, "ctx:AudioContext"));
+		break;
+	case Context::MESSAGE:
+		set_property("ctx:context", Atom(Atom::URI, "ctx:MessageContext"));
+		break;
+	}
+}
+
+
 } // namespace Ingen
diff --git a/src/engine/PortImpl.hpp b/src/engine/PortImpl.hpp
index e3de9f7f..e004e03f 100644
--- a/src/engine/PortImpl.hpp
+++ b/src/engine/PortImpl.hpp
@@ -108,7 +108,7 @@ public:
 	void raise_set_by_user_flag() { _set_by_user = true; }
 
 	Context::ID context() const            { return _context; }
-	void        set_context(Context::ID c) { _context = c; }
+	void        set_context(Context::ID c);
 
 protected:
 	PortImpl(BufferFactory&     bufs,
diff --git a/src/engine/ThreadManager.hpp b/src/engine/ThreadManager.hpp
index c088816f..6821949c 100644
--- a/src/engine/ThreadManager.hpp
+++ b/src/engine/ThreadManager.hpp
@@ -26,7 +26,8 @@ namespace Ingen {
 enum ThreadID {
 	THREAD_PRE_PROCESS,
 	THREAD_PROCESS,
-	THREAD_POST_PROCESS
+	THREAD_POST_PROCESS,
+	THREAD_MESSAGE,
 };
 
 
diff --git a/src/engine/events/Connect.cpp b/src/engine/events/Connect.cpp
index 18c31e03..a120aee4 100644
--- a/src/engine/events/Connect.cpp
+++ b/src/engine/events/Connect.cpp
@@ -75,9 +75,21 @@ Connect::pre_process()
 		return;
 	}
 
-	if ( ! (_src_port->type() == _dst_port->type()
-			|| ( (_src_port->type() == PortType::CONTROL || _src_port->type() == PortType::AUDIO)
-				&& (_dst_port->type() == PortType::CONTROL || _dst_port->type() == PortType::AUDIO) ))) {
+	const PortType src_type = _src_port->type();
+	const PortType dst_type = _dst_port->type();
+
+	if (	!(
+			// Equal types
+			(src_type == dst_type)
+
+			|| // or Control=>Audio or Audio=>Control
+			((src_type == PortType::CONTROL || src_type == PortType::AUDIO)
+				&& (dst_type == PortType::CONTROL || dst_type == PortType::AUDIO))
+
+			|| // or Events=>Message or Message=>Events
+			((src_type == PortType::EVENTS || src_type == PortType::MESSAGE)
+				&& (dst_type == PortType::EVENTS || dst_type == PortType::MESSAGE))
+			)) {
 		_error = TYPE_MISMATCH;
 		QueuedEvent::pre_process();
 		return;
diff --git a/src/engine/events/RequestMetadata.cpp b/src/engine/events/RequestMetadata.cpp
index 82165460..f51651fc 100644
--- a/src/engine/events/RequestMetadata.cpp
+++ b/src/engine/events/RequestMetadata.cpp
@@ -96,7 +96,7 @@ RequestMetadata::execute(ProcessContext& context)
 		if (port) {
 			if (port->type() == PortType::CONTROL || port->type() == PortType::AUDIO)
 				_value = ((AudioBuffer*)port->buffer(0).get())->value_at(0); // TODO: offset
-			else if (port->type() == PortType::VALUE)
+			else if (port->type() == PortType::VALUE || port->type() == PortType::MESSAGE)
 				LV2Object::to_atom(context.engine().world(),
 						((ObjectBuffer*)port->buffer(0).get())->object(), _value);
 		} else {
diff --git a/src/engine/tuning.hpp b/src/engine/tuning.hpp
index ea7c08c3..1ddaf71a 100644
--- a/src/engine/tuning.hpp
+++ b/src/engine/tuning.hpp
@@ -29,6 +29,7 @@ static const size_t event_queue_size           = 1024;
 static const size_t pre_processor_queue_size   = 1024;
 static const size_t post_processor_queue_size  = 1024;
 static const size_t maid_queue_size            = 1024;
+static const size_t message_context_queue_size = 1024;
 
 //static const timespec main_rate     = { 0, 500000000 }; // 1/2 second
 static const timespec main_rate       = { 0, 125000000 }; // 1/8 second