Move disconnect_all to CommonInterface and implement in clients.

Factor out process thread implementation of Disconnect into Disconnect::Impl.
Implement DisconnectAll in terms of DisconnectImpl rather than abusing Disconnect.
Dramatically reduce notification communication overhead of DisconnectAll.


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

9 files changed, 178 insertions, 126 deletions

diff --git a/src/engine/ClientBroadcaster.hpp b/src/engine/ClientBroadcaster.hpp
index ec0e7492..6c371a50 100644
--- a/src/engine/ClientBroadcaster.hpp
+++ b/src/engine/ClientBroadcaster.hpp
@@ -98,6 +98,11 @@ public:
 		BROADCAST(disconnect, src_port_path, dst_port_path);
 	}
 
+	void disconnect_all(const Raul::Path& parent_patch_path,
+	                    const Raul::Path& path) {
+		BROADCAST(disconnect_all, parent_patch_path, path);
+	}
+
 	void set_property(const Raul::URI&  subject,
 	                  const Raul::URI&  predicate,
 	                  const Raul::Atom& value) {
diff --git a/src/engine/HTTPClientSender.cpp b/src/engine/HTTPClientSender.cpp
index 83d09b97..8a03e2bd 100644
--- a/src/engine/HTTPClientSender.cpp
+++ b/src/engine/HTTPClientSender.cpp
@@ -98,7 +98,12 @@ HTTPClientSender::connect(const Path& src_path, const Path& dst_path)
 void
 HTTPClientSender::disconnect(const Path& src_path, const Path& dst_path)
 {
-	//send("/ingen/disconnection", "ss", src_path.c_str(), dst_path.c_str(), LO_ARGS_END);
+}
+
+void
+HTTPClientSender::disconnect_all(const Raul::Path& parent_patch_path,
+                                 const Raul::Path& path)
+{
 }
 
 void
diff --git a/src/engine/HTTPClientSender.hpp b/src/engine/HTTPClientSender.hpp
index 1b741071..501343a9 100644
--- a/src/engine/HTTPClientSender.hpp
+++ b/src/engine/HTTPClientSender.hpp
@@ -85,6 +85,9 @@ public:
 	virtual void disconnect(const Raul::Path& src_port_path,
 	                        const Raul::Path& dst_port_path);
 
+	virtual void disconnect_all(const Raul::Path& parent_patch_path,
+	                            const Raul::Path& path);
+
 	virtual void set_property(const Raul::URI&  subject_path,
 	                          const Raul::URI&  predicate,
 	                          const Raul::Atom& value);
diff --git a/src/engine/OSCClientSender.cpp b/src/engine/OSCClientSender.cpp
index 0d3c5cf6..88b979bf 100644
--- a/src/engine/OSCClientSender.cpp
+++ b/src/engine/OSCClientSender.cpp
@@ -192,6 +192,21 @@ OSCClientSender::disconnect(const Path& src_port_path,
 }
 
 /** @page client_osc_namespace
+ * <h2>/disconnect_all</h2>
+ * @arg @p parent-patch-path :: String
+ * @arg @p path :: String
+ *
+ * @par
+ * Notification all connections to an object have been disconnected.
+ */
+void
+OSCClientSender::disconnect_all(const Raul::Path& parent_patch_path,
+                                const Raul::Path& path)
+{
+	send("/disconnect_all", "ss", parent_patch_path.c_str(), path.c_str(), LO_ARGS_END);
+}
+
+/** @page client_osc_namespace
  * <h2>/set_property</h2>
  * @arg @p path :: String
  * @arg @p key :: URI String
diff --git a/src/engine/OSCClientSender.hpp b/src/engine/OSCClientSender.hpp
index 7352ebc2..a3031258 100644
--- a/src/engine/OSCClientSender.hpp
+++ b/src/engine/OSCClientSender.hpp
@@ -88,6 +88,9 @@ public:
 	virtual void disconnect(const Raul::Path& src_port_path,
 	                        const Raul::Path& dst_port_path);
 
+	virtual void disconnect_all(const Raul::Path& parent_patch_path,
+	                            const Raul::Path& path);
+
 	virtual void set_property(const Raul::URI&  subject,
 	                          const Raul::URI&  predicate,
 	                          const Raul::Atom& value);
diff --git a/src/engine/events/Disconnect.cpp b/src/engine/events/Disconnect.cpp
index 33a65ed8..aa59b618 100644
--- a/src/engine/events/Disconnect.cpp
+++ b/src/engine/events/Disconnect.cpp
@@ -31,6 +31,7 @@
 #include "PortImpl.hpp"
 #include "ProcessContext.hpp"
 #include "Request.hpp"
+#include "ThreadManager.hpp"
 
 using namespace std;
 using namespace Raul;
@@ -52,59 +53,82 @@ Disconnect::Disconnect(
 	, _src_port(NULL)
 	, _dst_port(NULL)
 	, _compiled_patch(NULL)
-	, _buffers(NULL)
-	, _internal(false)
-	, _reconnect_dst_port(true)
 {
 }
 
-Disconnect::Disconnect(
-		Engine&            engine,
-		SharedPtr<Request> request,
-		SampleCount        timestamp,
-		PortImpl* const    src_port,
-		PortImpl* const    dst_port,
-		bool               reconnect_dst_port)
-	: QueuedEvent(engine, request, timestamp)
-	, _src_port_path(src_port->path())
-	, _dst_port_path(dst_port->path())
-	, _patch(src_port->parent_node()->parent_patch())
-	, _src_port(src_port)
-	, _dst_port(dst_port)
-	, _compiled_patch(NULL)
+Disconnect::Impl::Impl(Engine&     e,
+                       PatchImpl*  patch,
+                       OutputPort* s,
+                       InputPort*  d)
+	: _engine(e)
+	, _src_output_port(s)
+	, _dst_input_port(d)
+	, _patch(patch)
+	, _connection(patch->remove_connection(_src_output_port, _dst_input_port))
 	, _buffers(NULL)
-	, _internal(true)
-	, _reconnect_dst_port(reconnect_dst_port)
 {
+	ThreadManager::assert_thread(THREAD_PRE_PROCESS);
+
+	NodeImpl* const src_node = _src_output_port->parent_node();
+	NodeImpl* const dst_node = _dst_input_port->parent_node();
+
+	for (Raul::List<NodeImpl*>::iterator i = dst_node->providers()->begin();
+	     i != dst_node->providers()->end(); ++i) {
+		if ((*i) == src_node) {
+			delete dst_node->providers()->erase(i);
+			break;
+		}
+	}
+
+	for (Raul::List<NodeImpl*>::iterator i = src_node->dependants()->begin();
+	     i != src_node->dependants()->end(); ++i) {
+		if ((*i) == dst_node) {
+			delete src_node->dependants()->erase(i);
+			break;
+		}
+	}
+
+	_dst_input_port->decrement_num_connections();
+
+	if (_dst_input_port->num_connections() == 0) {
+		_buffers = new Raul::Array<BufferFactory::Ref>(_dst_input_port->poly());
+		_dst_input_port->get_buffers(*_engine.buffer_factory(),
+				_buffers, _dst_input_port->poly());
+
+		const bool  is_control = _dst_input_port->is_a(PortType::CONTROL);
+		const float value      = is_control ? _dst_input_port->value().get_float() : 0;
+		for (uint32_t i = 0; i < _buffers->size(); ++i) {
+			if (is_control) {
+				PtrCast<AudioBuffer>(_buffers->at(i))->set_value(value, 0, 0);
+			} else {
+				_buffers->at(i)->clear();
+			}
+		}
+	}
+
+	_connection->pending_disconnection(true);
 }
 
 void
 Disconnect::pre_process()
 {
-	if (!_internal) {
-		if (_src_port_path.parent().parent() != _dst_port_path.parent().parent()
-				&& _src_port_path.parent() != _dst_port_path.parent().parent()
-				&& _src_port_path.parent().parent() != _dst_port_path.parent()) {
-			_error = PARENT_PATCH_DIFFERENT;
-			QueuedEvent::pre_process();
-			return;
-		}
-
-		_src_port = _engine.engine_store()->find_port(_src_port_path);
-		_dst_port = _engine.engine_store()->find_port(_dst_port_path);
+	if (_src_port_path.parent().parent() != _dst_port_path.parent().parent()
+	    && _src_port_path.parent() != _dst_port_path.parent().parent()
+	    && _src_port_path.parent().parent() != _dst_port_path.parent()) {
+		_error = PARENT_PATCH_DIFFERENT;
+		QueuedEvent::pre_process();
+		return;
 	}
 
+	_src_port = _engine.engine_store()->find_port(_src_port_path);
+	_dst_port = _engine.engine_store()->find_port(_dst_port_path);
+
 	if (_src_port == NULL || _dst_port == NULL) {
 		_error = PORT_NOT_FOUND;
 		QueuedEvent::pre_process();
 		return;
 	}
 
-	_dst_input_port = dynamic_cast<InputPort*>(_dst_port);
-	_src_output_port = dynamic_cast<OutputPort*>(_src_port);
-	assert(_src_output_port);
-	assert(_dst_input_port);
-
 	NodeImpl* const src_node = _src_port->parent_node();
 	NodeImpl* const dst_node = _dst_port->parent_node();
 
@@ -128,7 +152,7 @@ Disconnect::pre_process()
 
 	assert(_patch);
 
-	if (!_patch->has_connection(_src_output_port, _dst_input_port)) {
+	if (!_patch->has_connection(_src_port, _dst_port)) {
 		_error = NOT_CONNECTED;
 		QueuedEvent::pre_process();
 		return;
@@ -140,41 +164,46 @@ Disconnect::pre_process()
 		return;
 	}
 
-	for (Raul::List<NodeImpl*>::iterator i = dst_node->providers()->begin(); i != dst_node->providers()->end(); ++i)
-		if ((*i) == src_node) {
-			delete dst_node->providers()->erase(i);
-			break;
-		}
+	_impl = SharedPtr<Impl>(
+		new Impl(_engine,
+		         _patch,
+		         dynamic_cast<OutputPort*>(_src_port),
+		         dynamic_cast<InputPort*>(_dst_port)));
 
-	for (Raul::List<NodeImpl*>::iterator i = src_node->dependants()->begin(); i != src_node->dependants()->end(); ++i)
-		if ((*i) == dst_node) {
-			delete src_node->dependants()->erase(i);
-			break;
-		}
+	if (_patch->enabled())
+		_compiled_patch = _patch->compile();
 
-	_connection = _patch->remove_connection(_src_port, _dst_port);
-	_dst_input_port->decrement_num_connections();
+	QueuedEvent::pre_process();
+}
 
-	if (_dst_input_port->num_connections() == 0) {
-		_buffers = new Raul::Array<BufferFactory::Ref>(_dst_input_port->poly());
-		_dst_input_port->get_buffers(*_engine.buffer_factory(),
-				_buffers, _dst_input_port->poly());
+bool
+Disconnect::Impl::execute(ProcessContext& context, bool set_dst_buffers)
+{
+	ThreadManager::assert_thread(THREAD_PROCESS);
 
-		const bool  is_control = _dst_input_port->is_a(PortType::CONTROL);
-		const float value      = is_control ? _dst_input_port->value().get_float() : 0;
-		for (uint32_t i = 0; i < _buffers->size(); ++i) {
-			if (is_control) {
-				PtrCast<AudioBuffer>(_buffers->at(i))->set_value(value, 0, 0);
-			} else {
-				_buffers->at(i)->clear();
-			}
+	InputPort::Connections::Node* const port_connections_node
+		= _dst_input_port->remove_connection(context, _src_output_port);
+	if (!port_connections_node) {
+		return false;
+	}
+
+	if (set_dst_buffers) {
+		if (_buffers) {
+			_engine.maid()->push(_dst_input_port->set_buffers(_buffers));
+		} else {
+			_dst_input_port->setup_buffers(*_engine.buffer_factory(),
+			                               _dst_input_port->poly());
 		}
+		_dst_input_port->connect_buffers();
+	} else {
+		_dst_input_port->recycle_buffers();
 	}
 
-	if (!_internal && _patch->enabled())
-		_compiled_patch = _patch->compile();
+	assert(_connection);
+	assert(port_connections_node->elem() == _connection);
 
-	QueuedEvent::pre_process();
+	_engine.maid()->push(port_connections_node);
+	return true;
 }
 
 void
@@ -183,31 +212,13 @@ Disconnect::execute(ProcessContext& context)
 	QueuedEvent::execute(context);
 
 	if (_error == NO_ERROR) {
-		InputPort::Connections::Node* const port_connections_node
-			= _dst_input_port->remove_connection(context, _src_output_port);
-		if (_reconnect_dst_port) {
-			if (_buffers)
-				_engine.maid()->push(_dst_input_port->set_buffers(_buffers));
-			else
-				_dst_input_port->setup_buffers(*_engine.buffer_factory(), _dst_input_port->poly());
-			_dst_input_port->connect_buffers();
-		} else {
-			_dst_input_port->recycle_buffers();
-		}
-
-		if (port_connections_node) {
-			assert(_connection);
-			assert(port_connections_node->elem() == _connection);
-
-			_engine.maid()->push(port_connections_node);
-
-			if (!_internal) {
-				_engine.maid()->push(_patch->compiled_patch());
-				_patch->compiled_patch(_compiled_patch);
-			}
-		} else {
+		if (!_impl->execute(context, true)) {
 			_error = CONNECTION_NOT_FOUND;
+			return;
 		}
+
+		_engine.maid()->push(_patch->compiled_patch());
+		_patch->compiled_patch(_compiled_patch);
 	}
 }
 
diff --git a/src/engine/events/Disconnect.hpp b/src/engine/events/Disconnect.hpp
index 9c741293..0c177d0a 100644
--- a/src/engine/events/Disconnect.hpp
+++ b/src/engine/events/Disconnect.hpp
@@ -53,20 +53,31 @@ public:
 			const Raul::Path&  src_port_path,
 			const Raul::Path&  dst_port_path);
 
-	Disconnect(
-			Engine&            engine,
-			SharedPtr<Request> request,
-			SampleCount        timestamp,
-			PortImpl* const    src_port,
-			PortImpl* const    dst_port,
-			bool               reconnect_dst_port);
-
 	void pre_process();
 	void execute(ProcessContext& context);
 	void post_process();
 
-private:
+	class Impl {
+	public:
+		Impl(Engine&     e,
+		     PatchImpl*  patch,
+		     OutputPort* s,
+		     InputPort*  d);
 
+		bool execute(ProcessContext& context, bool set_dst_buffers);
+
+		InputPort* dst_port() { return _dst_input_port; }
+
+	private:
+		Engine&                          _engine;
+		OutputPort*                      _src_output_port;
+		InputPort*                       _dst_input_port;
+		PatchImpl*                       _patch;
+		SharedPtr<ConnectionImpl>        _connection;
+		Raul::Array<BufferFactory::Ref>* _buffers;
+	};
+
+private:
 	enum ErrorType {
 		NO_ERROR,
 		PARENT_PATCH_DIFFERENT,
@@ -80,19 +91,13 @@ private:
 	Raul::Path _src_port_path;
 	Raul::Path _dst_port_path;
 
-	PatchImpl*  _patch;
-	PortImpl*   _src_port;
-	PortImpl*   _dst_port;
-	OutputPort* _src_output_port;
-	InputPort*  _dst_input_port;
+	PatchImpl* _patch;
+	PortImpl*  _src_port;
+	PortImpl*  _dst_port;
 
-	SharedPtr<ConnectionImpl> _connection;
 	CompiledPatch* _compiled_patch; ///< New process order for Patch
 
-	Raul::Array<BufferFactory::Ref>* _buffers;
-
-	bool _internal;
-	bool _reconnect_dst_port;
+	SharedPtr<Impl> _impl;
 };
 
 } // namespace Events
diff --git a/src/engine/events/DisconnectAll.cpp b/src/engine/events/DisconnectAll.cpp
index c49cd9d7..a79969d3 100644
--- a/src/engine/events/DisconnectAll.cpp
+++ b/src/engine/events/DisconnectAll.cpp
@@ -69,19 +69,21 @@ DisconnectAll::DisconnectAll(Engine& engine, PatchImpl* parent, GraphObjectImpl*
 
 DisconnectAll::~DisconnectAll()
 {
-	for (Raul::List<Disconnect*>::iterator i = _disconnect_events.begin(); i != _disconnect_events.end(); ++i)
+	for (Raul::List<Disconnect::Impl*>::iterator i = _disconnect_events.begin();
+	     i != _disconnect_events.end(); ++i)
 		delete (*i);
 }
 
 void
 DisconnectAll::remove_connection(ConnectionImpl* c)
 {
-	const bool reconnect_input = !_deleting || (c->dst_port()->parent_node() != _node);
-	Disconnect* ev = new Disconnect(_engine, SharedPtr<Request>(), _time,
-	                                c->src_port(), c->dst_port(), reconnect_input);
-	ev->pre_process();
-	_disconnect_events.push_back(new Raul::List<Disconnect*>::Node(ev));
-	c->pending_disconnection(true);
+
+	_disconnect_events.push_back(
+		new Raul::List<Disconnect::Impl*>::Node(
+			new Disconnect::Impl(_engine,
+			                     _parent,
+			                     dynamic_cast<OutputPort*>(c->src_port()),
+			                     dynamic_cast<InputPort*>(c->dst_port()))));
 }
 
 void
@@ -151,9 +153,11 @@ DisconnectAll::execute(ProcessContext& context)
 	QueuedEvent::execute(context);
 
 	if (_error == NO_ERROR) {
-		for (Raul::List<Disconnect*>::iterator i = _disconnect_events.begin();
-				i != _disconnect_events.end(); ++i)
-			(*i)->execute(context);
+		for (Raul::List<Disconnect::Impl*>::iterator i = _disconnect_events.begin();
+		     i != _disconnect_events.end(); ++i) {
+			(*i)->execute(context,
+			              !_deleting || ((*i)->dst_port()->parent_node() != _node));
+		}
 	}
 
 	_engine.maid()->push(_parent->compiled_patch());
@@ -166,9 +170,7 @@ DisconnectAll::post_process()
 	if (_error == NO_ERROR) {
 		if (_request)
 			_request->respond_ok();
-		for (Raul::List<Disconnect*>::iterator i = _disconnect_events.begin();
-				i != _disconnect_events.end(); ++i)
-			(*i)->post_process();
+		_engine.broadcaster()->disconnect_all(_parent_path, _path);
 	} else {
 		if (_request) {
 			boost::format fmt("Unable to disconnect %1% (%2%)");
diff --git a/src/engine/events/DisconnectAll.hpp b/src/engine/events/DisconnectAll.hpp
index ee8fbade..65c9599f 100644
--- a/src/engine/events/DisconnectAll.hpp
+++ b/src/engine/events/DisconnectAll.hpp
@@ -20,6 +20,8 @@
 
 #include "raul/List.hpp"
 #include "raul/Path.hpp"
+
+#include "Disconnect.hpp"
 #include "QueuedEvent.hpp"
 
 namespace Ingen {
@@ -69,16 +71,17 @@ private:
 
 	void remove_connection(ConnectionImpl* c);
 
-	Raul::Path              _parent_path;
-	Raul::Path              _path;
-	PatchImpl*              _parent;
-	NodeImpl*               _node;
-	PortImpl*               _port;
-	Raul::List<Disconnect*> _disconnect_events;
+	Raul::Path _parent_path;
+	Raul::Path _path;
+	PatchImpl* _parent;
+	NodeImpl*  _node;
+	PortImpl*  _port;
+
+	Raul::List<Disconnect::Impl*> _disconnect_events;
 
 	CompiledPatch* _compiled_patch; ///< New process order for Patch
 
-	bool      _deleting;
+	bool _deleting;
 };
 
 } // namespace Engine