5 files changed, 26 insertions, 16 deletions

diff --git a/src/engine/ConnectionImpl.cpp b/src/engine/ConnectionImpl.cpp
index c6ccf3ba..8fa770f7 100644
--- a/src/engine/ConnectionImpl.cpp
+++ b/src/engine/ConnectionImpl.cpp
@@ -72,12 +72,17 @@ ConnectionImpl::dump() const
 
 void
 ConnectionImpl::get_sources(Context& context, uint32_t voice,
-		Buffer** srcs, uint32_t max_num_srcs, uint32_t& num_srcs)
+		IntrusivePtr<Buffer>* srcs, uint32_t max_num_srcs, uint32_t& num_srcs)
 {
-	if (must_queue())
-		return;
-
-	if (must_mix()) {
+	if (must_queue() && _queue->read_space() > 0) {
+		LV2_Object obj;
+		_queue->peek(sizeof(LV2_Object), &obj);
+		IntrusivePtr<Buffer> buf = context.engine().buffer_factory()->get(
+				dst_port()->buffer_type(), sizeof(LV2_Object) + obj.size);
+		void* data = buf->port_data(PortType::MESSAGE, context.offset());
+		_queue->full_read(sizeof(LV2_Object) + obj.size, (LV2_Object*)data);
+		srcs[num_srcs++] = buf;
+	} else if (must_mix()) {
 		// Mixing down voices: every src voice mixed into every dst voice
 		for (uint32_t v = 0; v < _src_port->poly(); ++v) {
 			assert(num_srcs < max_num_srcs);
diff --git a/src/engine/ConnectionImpl.hpp b/src/engine/ConnectionImpl.hpp
index 2725dd83..000477e2 100644
--- a/src/engine/ConnectionImpl.hpp
+++ b/src/engine/ConnectionImpl.hpp
@@ -22,6 +22,7 @@
 #include <boost/utility.hpp>
 #include "raul/log.hpp"
 #include "raul/Deletable.hpp"
+#include "raul/IntrusivePtr.hpp"
 #include "interface/PortType.hpp"
 #include "interface/Connection.hpp"
 #include "object.lv2/object.h"
@@ -67,7 +68,7 @@ public:
 	void queue(Context& context);
 
 	void get_sources(Context& context, uint32_t voice,
-			Buffer** srcs, uint32_t max_num_srcs, uint32_t& num_srcs);
+			IntrusivePtr<Buffer>* srcs, uint32_t max_num_srcs, uint32_t& num_srcs);
 
 	/** Get the buffer for a particular voice.
 	 * A Connection is smart - it knows the destination port requesting the
diff --git a/src/engine/InputPort.cpp b/src/engine/InputPort.cpp
index 4c5b56e2..95af5725 100644
--- a/src/engine/InputPort.cpp
+++ b/src/engine/InputPort.cpp
@@ -92,7 +92,7 @@ InputPort::get_buffers(BufferFactory& bufs, Raul::Array<BufferFactory::Ref>* buf
 
 	} else if (num_connections == 1) {
 		if (ThreadManager::current_thread_id() == THREAD_PROCESS) {
-			if (!_connections.front()->must_mix()) {
+			if (!_connections.front()->must_mix() && !_connections.front()->must_queue()) {
 				// Single non-mixing conneciton, use buffers directly
 				for (uint32_t v = 0; v < poly; ++v)
 					buffers->at(v) = _connections.front()->buffer(v);
@@ -180,7 +180,7 @@ InputPort::pre_process(Context& context)
 	for (Connections::iterator c = _connections.begin(); c != _connections.end(); ++c)
 		max_num_srcs += (*c)->src_port()->poly();
 
-	Buffer* srcs[max_num_srcs];
+	IntrusivePtr<Buffer> srcs[max_num_srcs];
 
 	if (_connections.empty()) {
 		for (uint32_t v = 0; v < _poly; ++v) {
@@ -217,8 +217,8 @@ InputPort::post_process(Context& context)
 bool
 InputPort::direct_connect() const
 {
-	ThreadManager::assert_thread(THREAD_PROCESS);
-	return _connections.size() == 1 && !_connections.front()->must_mix();
+	return _connections.size() == 1 && !_connections.front()->must_mix()
+			&& !_connections.front()->must_queue();
 }
 
 
diff --git a/src/engine/events/Connect.cpp b/src/engine/events/Connect.cpp
index 5eefb181..51378ca1 100644
--- a/src/engine/events/Connect.cpp
+++ b/src/engine/events/Connect.cpp
@@ -135,7 +135,8 @@ Connect::pre_process()
 	_patch->add_connection(_connection);
 	_dst_input_port->increment_num_connections();
 
-	if ((_dst_input_port->num_connections() == 1 && _connection->must_mix())
+	if ((_dst_input_port->num_connections() == 1
+				&& (_connection->must_mix() || _connection->must_queue()))
 			|| _dst_input_port->num_connections() == 2) {
 		_buffers = new Raul::Array<BufferFactory::Ref>(_dst_input_port->poly());
 		_dst_input_port->get_buffers(*_engine.buffer_factory(),
diff --git a/src/engine/mix.hpp b/src/engine/mix.hpp
index ad12bad0..e84a63e6 100644
--- a/src/engine/mix.hpp
+++ b/src/engine/mix.hpp
@@ -28,19 +28,19 @@ using namespace Raul;
 namespace Ingen {
 
 inline void
-mix(Context& context, Buffer* dst, const Buffer*const* srcs, uint32_t num_srcs)
+mix(Context& context, Buffer* dst, const IntrusivePtr<Buffer>* srcs, uint32_t num_srcs)
 {
 	using Shared::PortType;
 	switch (dst->type().symbol()) {
 	case PortType::AUDIO:
 	case PortType::CONTROL:
 		// Copy the first source
-		dst->copy(context, srcs[0]);
+		dst->copy(context, srcs[0].get());
 
 		// Mix in the rest
 		for (uint32_t i = 1; i < num_srcs; ++i) {
 			assert(srcs[i]->type() == PortType::AUDIO || srcs[i]->type() == PortType::CONTROL);
-			((AudioBuffer*)dst)->accumulate(context, (AudioBuffer*)srcs[i]);
+			((AudioBuffer*)dst)->accumulate(context, (AudioBuffer*)srcs[i].get());
 		}
 
 		break;
@@ -55,7 +55,7 @@ mix(Context& context, Buffer* dst, const Buffer*const* srcs, uint32_t num_srcs)
 		while (true) {
 			const EventBuffer* first = NULL;
 			for (uint32_t i = 0; i < num_srcs; ++i) {
-				const EventBuffer* const src = (const EventBuffer*)srcs[i];
+				const EventBuffer* const src = (const EventBuffer*)srcs[i].get();
 				if (src->is_valid()) {
 					if (!first || src->get_event()->frames < first->get_event()->frames)
 						first = src;
@@ -76,7 +76,10 @@ mix(Context& context, Buffer* dst, const Buffer*const* srcs, uint32_t num_srcs)
 		break;
 
 	default:
-		error << "Mix of unsupported buffer types" << std::endl;
+		if (num_srcs == 1)
+			dst->copy(context, srcs[0].get());
+		else
+			error << "Mix of unsupported buffer types" << std::endl;
 		return;
 	}
 }