This patch removes the dependency on EventDrivenActivity for DataDrivenTask
components. The event-triggering ports management is now done in DataFlowInterface
(where port management belongs) and the DataDrivenTask only relies on information
in that class. The new DataFlowInterface::addEventPort method sets which ports
should fire events for this task. the DataDrivenTask::start method picks this up
registers a callback and this one collects the data for updateHook(updated_ports).

---
src/DataDrivenTask.cpp | 56 +++++++++++++++-----------------------------
src/DataDrivenTask.hpp | 11 ++++++--
src/DataFlowInterface.cpp | 28 ++++++++++++++++++++++
src/DataFlowInterface.hpp | 37 +++++++++++++++++++++++++++--
src/EventService.hpp | 4 +++
src/PortInterface.cpp | 16 ++++++------
src/PortInterface.hpp | 17 ++-----------
7 files changed, 104 insertions(+), 65 deletions(-)

diff --git a/src/DataDrivenTask.cpp b/src/DataDrivenTask.cpp
index 177be44..8351433 100644
--- a/src/DataDrivenTask.cpp
+++ b/src/DataDrivenTask.cpp
@@ -11,53 +11,35 @@ DataDrivenTask::DataDrivenTask(const std::string& name, ExecutionEngine* parent,

bool DataDrivenTask::start()
{
- EventDrivenActivity* activity = dynamic_cast(engine()->getActivity());
- if (activity)
+ size_t port_count = 0;
+ const DataFlowInterface::Ports& ports = this->ports()->getEventPorts();
+ for (DataFlowInterface::Ports::const_iterator it = ports.begin(); it != ports.end(); ++it)
{
- size_t port_count = 0;
- log(Info) << getName() << " is attached to an EventDrivenActivity. Registering ports." << endlog();
- DataFlowInterface::Ports ports = this->ports()->getPorts();
- for (DataFlowInterface::Ports::iterator it = ports.begin(); it != ports.end(); ++it)
+ int porttype = (*it)->getPortType();
+ if (porttype == PortInterface::ReadPort || porttype == PortInterface::ReadWritePort)
{
- int porttype = (*it)->getPortType();
- if (porttype == PortInterface::ReadPort || porttype == PortInterface::ReadWritePort)
- {
- PortInterface::NewDataEvent* ev = (*it)->getNewDataEvent();
- if (!activity->addEvent(ev))
- return false;
- port_count++;
-
- log(Info) << getName() << " will be triggered when new data is available on " << (*it)->getName() << endlog();
- }
+ (*it)->getNewDataOnPortEvent()->connect(boost::bind(&DataDrivenTask::dataOnPort, this, _1), this->events()->getEventProcessor());
+ port_count++;
+ log(Info) << getName() << " will be triggered when new data is available on " << (*it)->getName() << endlog();
}
- updated_ports.resize(port_count);
- }
- else
- {
- log(Warning) << "activity is not an EventDrivenActivity" << endlog();
}
-
+ updated_ports.reserve(port_count);
return TaskContext::start();
}

-void DataDrivenTask::updateHook()
+void DataDrivenTask::dataOnPort(PortInterface* port)
{
- EventDrivenActivity* activity = dynamic_cast(engine()->getActivity());
- if (activity)
- {
- typedef std::vector< Event* > TriggerSet;
- TriggerSet const& updates = activity->getWakeupEvents();
- updated_ports.clear();
-
- for (TriggerSet::const_iterator it = updates.begin(); it != updates.end(); ++it)
- {
- PortInterface::NewDataEvent* ev = static_cast(*it);
- if (ev)
- updated_ports.push_back(ev->getPort());
- }
- }
+ // Since this handler is executed in our thread, we are always running.
+ if (find(updated_ports.begin(), updated_ports.end(), port) == updated_ports.end() )
+ updated_ports.push_back(port);
+ // this is in essence superfluous. We are already triggered.
+ //this->getActivity()->trigger();
+}

+void DataDrivenTask::updateHook()
+{
updateHook(updated_ports);
+ updated_ports.clear();
}

void DataDrivenTask::updateHook(std::vector const& updated_ports)
diff --git a/src/DataDrivenTask.hpp b/src/DataDrivenTask.hpp
index 9ae33de..e9b2d3b 100644
--- a/src/DataDrivenTask.hpp
+++ b/src/DataDrivenTask.hpp
@@ -7,13 +7,18 @@
namespace RTT {
class DataDrivenTask : public TaskContext
{
+ protected:
std::vector< PortInterface* > updated_ports;
-
+ /**
+ * This callback is called each time data arrived on an
+ * event port.
+ */
+ void dataOnPort(PortInterface*);
public:
/**
* Create a DataDrivenTask.
*
- * It's ExecutionEngine will be newly constructed with private
+ * It's ExecutionEngine will be newly constructed with private
* processing of commands, events, programs and state machines.
*
* @param name The name of this component.
@@ -23,7 +28,7 @@ namespace RTT {
DataDrivenTask( const std::string& name, TaskState initial_state = Stopped );

/**
- * Create a DataDrivenTask.
+ * Create a DataDrivenTask.
*
* Its commands programs and state machines are processed by \a parent.
* Use this constructor to share execution engines among task contexts, such that
diff --git a/src/DataFlowInterface.cpp b/src/DataFlowInterface.cpp
index 759d83e..5823b24 100644
--- a/src/DataFlowInterface.cpp
+++ b/src/DataFlowInterface.cpp
@@ -74,6 +74,16 @@ namespace RTT
return true;
}

+ bool DataFlowInterface::addEventPort(PortInterface* port, PortInterface::NewDataOnPortEvent::SlotFunction callback) {
+ if (this->addPort(port)) {
+ if (callback)
+ port->getNewDataOnPortEvent()->connect(callback, mparent->events()->getEventProcessor() );
+ eports.push_back(port);
+ return true;
+ }
+ return false;
+ }
+
bool DataFlowInterface::addPort(PortInterface* port, std::string description) {
if (this->addPort(port) == false)
return false;
@@ -86,6 +96,21 @@ namespace RTT
return true;
}

+ bool DataFlowInterface::addEventPort(PortInterface* port, std::string description, PortInterface::NewDataOnPortEvent::SlotFunction callback) {
+ if (this->addPort(port)) {
+ if (callback)
+ port->getNewDataOnPortEvent()->connect(callback, mparent->events()->getEventProcessor() );
+ eports.push_back(port);
+ return true;
+ }
+ return false;
+ }
+
+ const DataFlowInterface::Ports& DataFlowInterface::getEventPorts() const
+ {
+ return eports;
+ }
+
void DataFlowInterface::removePort(const std::string& name) {
for ( PortStore::iterator it(mports.begin());
it != mports.end();
@@ -93,6 +118,9 @@ namespace RTT
if ( it->first->getName() == name ) {
if (mparent)
mparent->removeObject( name );
+ Ports::iterator ep = find(eports.begin(), eports.end(),it->first);
+ if ( ep!= eports.end() )
+ eports.erase( ep );
mports.erase(it);
return;
}
diff --git a/src/DataFlowInterface.hpp b/src/DataFlowInterface.hpp
index a0ea188..e61b1af 100644
--- a/src/DataFlowInterface.hpp
+++ b/src/DataFlowInterface.hpp
@@ -53,9 +53,6 @@ namespace RTT
*/
class DataFlowInterface
{
- typedef std::vector > PortStore;
- PortStore mports;
- OperationInterface* mparent;
public:
/**
* A sequence of pointers to ports.
@@ -86,6 +83,23 @@ namespace RTT
bool addPort(PortInterface* port);

/**
+ * Add an Event triggering Port to this task. It is only added to the C++
+ * interface and can not be used in scripting.
+ * @param port The port to add.
+ * @return true if the port could be added, false if already added.
+ * @param callback (Optional) provide a function which will be called asynchronously
+ * when new data arrives on this port. You can add more functions by using the port
+ * directly using PortInterface::getNewDataOnPort().
+ */
+ bool addEventPort(PortInterface* port, PortInterface::NewDataOnPortEvent::SlotFunction callback = PortInterface::NewDataOnPortEvent::SlotFunction() );
+
+ /**
+ * Returns the list of all ports emitting events when new
+ * data arrives.
+ */
+ const Ports& getEventPorts() const;
+
+ /**
* Add a Port to the interface of this task. It is added to
* both the C++ interface and the scripting interface.
* @param port The port to add.
@@ -94,6 +108,17 @@ namespace RTT
bool addPort(PortInterface* port, std::string description);

/**
+ * Add an Event triggering Port to the interface of this task. It is added to
+ * both the C++ interface and the scripting interface.
+ * @param port The port to add.
+ * @param description A user readable description of this port.
+ * @param callback (Optional) provide a function which will be called asynchronously
+ * when new data arrives on this port. You can add more functions by using the port
+ * directly using PortInterface::getNewDataOnPort().
+ */
+ bool addEventPort(PortInterface* port, std::string description, PortInterface::NewDataOnPortEvent::SlotFunction callback = PortInterface::NewDataOnPortEvent::SlotFunction() );
+
+ /**
* Remove a Port from this interface.
* @param port The port to remove.
*/
@@ -156,6 +181,12 @@ namespace RTT
* all associated TaskObjects.
*/
void clear();
+ protected:
+ typedef std::vector > PortStore;
+ Ports eports;
+ PortStore mports;
+ OperationInterface* mparent;
+
};

}
diff --git a/src/EventService.hpp b/src/EventService.hpp
index 768d792..821692e 100644
--- a/src/EventService.hpp
+++ b/src/EventService.hpp
@@ -145,6 +145,10 @@ namespace RTT
* @param Signature Signature of the Event, for example void(int, int)
* @param ename The name of the event to lookup.
* @return A shared pointer which is to be assigned to an Event object.
+ * @internal We need Signature for two reasons: 1. for consistency with the general
+ * getPrimitive(name) API in RTT and 2. because we may need the type to reconstruct the
+ * object dynamically, for example, when the Event is remote. See the CommandRepository
+ * for and example where this is necessary (see also RTT::RemoteCommand )
*/
template
boost::shared_ptr getEvent(const std::string& ename)
diff --git a/src/PortInterface.cpp b/src/PortInterface.cpp
index 787684f..7ceac96 100644
--- a/src/PortInterface.cpp
+++ b/src/PortInterface.cpp
@@ -43,11 +43,11 @@
namespace RTT
{

- PortInterface::PortInterface(const std::string& name) : portname(name), new_data_event(0) {}
+ PortInterface::PortInterface(const std::string& name) : portname(name), new_data_on_port_event(0) {}

PortInterface::~PortInterface()
{
- delete new_data_event;
+ delete new_data_on_port_event;

}

@@ -208,14 +208,14 @@ namespace RTT

void PortInterface::signal()
{
- if (new_data_event)
- (*new_data_event)();
+ if (new_data_on_port_event)
+ (*new_data_on_port_event)(this);
}

- PortInterface::NewDataEvent* PortInterface::getNewDataEvent()
+ PortInterface::NewDataOnPortEvent* PortInterface::getNewDataOnPortEvent()
{
- if (!new_data_event)
- new_data_event = new NewDataEvent(this);
- return new_data_event;
+ if (!new_data_on_port_event)
+ new_data_on_port_event = new NewDataOnPortEvent( this->getName() );
+ return new_data_on_port_event;
}
}
diff --git a/src/PortInterface.hpp b/src/PortInterface.hpp
index 9c972a3..8e13836 100644
--- a/src/PortInterface.hpp
+++ b/src/PortInterface.hpp
@@ -55,22 +55,11 @@ namespace RTT
class PortInterface
{
public:
- class NewDataEvent : public Event< void() >
- {
- PortInterface* port;
-
- public:
- NewDataEvent(PortInterface* port)
- : Event< void() >(port->getName() + "Trigger")
- , port(port) {}
-
- PortInterface* getPort() const { return port; }
- };
-
+ typedef Event NewDataOnPortEvent;
protected:
friend class ConnectionInterface;
std::string portname;
- NewDataEvent* new_data_event;
+ NewDataOnPortEvent* new_data_on_port_event;
PortInterface(const std::string& name);

/**
@@ -121,7 +110,7 @@ namespace RTT
* Returns an Event object which is triggered every time new data is
* made available on this port
*/
- NewDataEvent* getNewDataEvent();
+ NewDataOnPortEvent* getNewDataOnPortEvent();

/**
* Get the ConnectionModel of this port.