OutPort.h

Go to the documentation of this file.

// -*- C++ -*-
#ifndef RTC_OUTPORT_H
#define RTC_OUTPORT_H
 
#include <coil/TimeMeasure.h>
#include <coil/OS.h>
 
#include <rtm/RTC.h>
#include <rtm/Typename.h>
#include <rtm/OutPortBase.h>
#include <rtm/CdrBufferBase.h>
#include <rtm/PortCallback.h>
#include <rtm/OutPortConnector.h>
#include <rtm/Timestamp.h>
#include <rtm/DirectOutPortBase.h>
#include <rtm/DataTypeUtil.h>
 
#include <functional>
#include <string>
#include <vector>
 
namespace RTC
{
  template <class DataType>
  class OutPort
      : public OutPortBase, DirectOutPortBase<DataType>
  {
  public:
    OutPort(const char* name, DataType& value)
      : OutPortBase(name, ::CORBA_Util::toRepositoryId<DataType>()),
        m_value(value), m_onWrite(nullptr), m_onWriteConvert(nullptr),
        m_directNewData(false), m_directValue(value)
    {
      this->initConnectorListeners();
      this->addConnectorDataListener(ConnectorDataListenerType::ON_BUFFER_WRITE,
                                     new Timestamp<DataType>("on_write"));
      this->addConnectorDataListener(ConnectorDataListenerType::ON_SEND,
                                     new Timestamp<DataType>("on_send"));
 
      m_directport = this;
 
      CdrMemoryStreamInit<DataType>();
 
      std::string marshaling_types{coil::eraseBlank(coil::flatten(
        getSerializerList<DataType>()))};
 
      RTC_DEBUG(("available marshaling_types: %s", marshaling_types.c_str()));
 
      addProperty("dataport.marshaling_types", marshaling_types.c_str());
    }
 
    ~OutPort() override;
 
    virtual bool write(DataType& value)
    {
      RTC_TRACE(("DataType write()"));
 
      if (m_onWrite != nullptr)
        {
          (*m_onWrite)(value);
          RTC_TRACE(("OnWrite called"));
        }
 
 
      bool result(true);
      std::vector<const char *> disconnect_ids;
      {
        std::lock_guard<std::mutex> con_guard(m_connectorsMutex);
        // check number of connectors
        size_t conn_size(m_connectors.size());
        if (!(conn_size > 0)) { return false; }
 
        m_status.resize(conn_size);
 
        for (size_t i(0), len(conn_size); i < len; ++i)
          {
 
            DataPortStatus ret;
            if (!m_connectors[i]->pullDirectMode())
              {
                if (m_onWriteConvert != nullptr)
                  {
                    RTC_DEBUG(("m_connectors.OnWriteConvert called"));
                    DataType tmp = (*m_onWriteConvert)(value);
                    ret = m_connectors[i]->write(tmp);
                  }
                else
                  {
                    RTC_DEBUG(("m_connectors.write called"));
                    ret = m_connectors[i]->write(value);
                  }
              }
            else
              {
                std::lock_guard<std::mutex> value_guard(m_valueMutex);
                if (m_onWriteConvert != nullptr)
                  {
                    RTC_DEBUG(("m_connectors.OnWriteConvert called"));
                    m_directValue = ((*m_onWriteConvert)(value));
                  }
                else
                  {
                    CORBA_Util::copyData<DataType>(m_directValue, value);
                  }
                m_directNewData = true;
                ret = DataPortStatus::PORT_OK;
              }
            m_status[i] = ret;
 
            if (ret == DataPortStatus::PORT_OK) { continue; }
 
            result = false;
 
            if (ret == DataPortStatus::CONNECTION_LOST)
              {
                const char* id(m_connectors[i]->profile().id.c_str());
                RTC_WARN(("connection_lost id: %s", id));
                if (m_onConnectionLost != nullptr)
                  {
                    RTC::ConnectorProfile prof(findConnProfile(id));
                    (*m_onConnectionLost)(prof);
                  }
                disconnect_ids.emplace_back(id);
              }
          }
      }
      std::for_each(disconnect_ids.begin(), disconnect_ids.end(),
                    [this](const char * id){this->disconnect(id);});
      return result;
    }
 
    bool write() override
    {
      return write(m_value);
    }
 
    bool operator<<(DataType& value)
    {
      return write(value);
    }
 
    DataPortStatus getStatus(int index)
    {
      return m_status[index];
    }
    DataPortStatusList getStatusList()
    {
      return m_status;
    }
 
    inline void setOnWrite(OnWrite<DataType>* on_write)
    {
      m_onWrite = on_write;
    }
 
    inline void setOnWriteConvert(OnWriteConvert<DataType>* on_wconvert)
    {
      m_onWriteConvert = on_wconvert;
    }
 
    void read(DataType& data) override
    {
        std::lock_guard<std::mutex> guard(m_valueMutex);
        m_directNewData = false;
        CORBA_Util::copyData<DataType>(data, m_directValue);
    }
    bool isEmpty() override
    {
        return !m_directNewData;
    }
    bool isNew() override
    {
        return m_directNewData;
    }
 
  protected:
    void initConnectorListeners() override
    {
        delete m_listeners;
        m_listeners = new ConnectorListenersT<DataType>();
    }
  private:
    std::string m_typename;
    DataType& m_value;
 
    OnWrite<DataType>* m_onWrite;
 
    OnWriteConvert<DataType>* m_onWriteConvert;
 
    coil::TimeMeasure m_cdrtime;
 
    DataPortStatusList m_status;
 
    CORBA::Long m_propValueIndex;
 
    std::mutex m_valueMutex;
    bool m_directNewData;
    DataType m_directValue;
  };
 
  template <class T> OutPort<T>::~OutPort() = default; // No inline for gcc warning, too big
} // namespace RTC
 
#endif  // RTC_OUTPORT_H