InPort.h

Go to the documentation of this file.

// -*- C++ -*-
#ifndef RTC_INPORT_H
#define RTC_INPORT_H
 
#include <coil/OS.h>
#include <mutex>
 
#include <rtm/RTC.h>
#include <rtm/Typename.h>
#include <rtm/InPortBase.h>
#include <rtm/CdrBufferBase.h>
#include <rtm/PortCallback.h>
#include <rtm/InPortConnector.h>
#include <rtm/Timestamp.h>
#include <rtm/DirectInPortBase.h>
#include <rtm/CORBA_CdrMemoryStream.h>
#include <rtm/DataTypeUtil.h>
 
 
namespace RTC
{
  template <class DataType>
  class InPort
       : public InPortBase, DirectInPortBase<DataType>
  {
  public:
    InPort(const char* name, DataType& value)
      : InPortBase(name, ::CORBA_Util::toRepositoryId<DataType>()),
        m_name(name), m_value(value),
        m_OnRead(nullptr),  m_OnReadConvert(nullptr),
        m_status(1), m_directNewData(false)
    {
 
      this->initConnectorListeners();
 
      this->addConnectorDataListener(ConnectorDataListenerType::ON_RECEIVED,
                                     new Timestamp<DataType>("on_received"));
      this->addConnectorDataListener(ConnectorDataListenerType::ON_BUFFER_READ,
                                     new Timestamp<DataType>("on_read"));
      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());
    }
 
    ~InPort() override;
 
    virtual const char* name()
    {
      return m_name.c_str();
    }
 
 
    virtual bool isNew(std::string name)
    {
        RTC_TRACE(("isNew()"));
 
        
        {
            std::lock_guard<std::mutex> guard(m_connectorsMutex);
            if (m_connectors.empty())
            {
                RTC_DEBUG(("no connectors"));
                return false;
            }
 
            for(auto & con : m_connectors)
            {
                if (std::string(con->name()) == name)
                {
                    size_t r = con->getBuffer()->readable();
                    if (r > 0)
                    {
                        RTC_DEBUG(("isNew() = true, readable data: %d", r));
                        return true;
                    }
                }
            }
        }
 
        RTC_DEBUG(("isNew() = false, no readable data"));
        return false;
    }
    virtual bool isNew(coil::vstring &names)
    {
        names.clear();
        RTC_TRACE(("isNew()"));
 
 
        
        {
            std::lock_guard<std::mutex> guard(m_connectorsMutex);
            if (m_connectors.empty())
            {
                RTC_DEBUG(("no connectors"));
                return false;
            }
            for(auto & con : m_connectors)
            {
                size_t r = con->getBuffer()->readable();
                if (r > 0)
                {
                    names.emplace_back(con->name());
                }
            }
            
        }
 
        if (!names.empty())
        {
            RTC_DEBUG(("isNew() = true, buffer is not empty"));
            return true;
        }
 
        RTC_DEBUG(("isNew() = false, no readable data"));
        return false;
    }
    bool isNew() override
    {
      RTC_TRACE(("isNew()"));
 
      // In single-buffer mode, all connectors share the same buffer. This
      // means that we only need to read from the first connector to get data
      // received by any connector.
      {
        std::lock_guard<std::mutex> guard(m_valueMutex);
        if (m_directNewData == true)
          {
            RTC_DEBUG(("isNew() returns true because of direct write."));
            return true;
          }
      }
      size_t r(0);
      {
        std::lock_guard<std::mutex> guard(m_connectorsMutex);
        if (m_connectors.empty())
          {
            RTC_DEBUG(("no connectors"));
            return false;
          }
        r = m_connectors[0]->getBuffer()->readable();
      }
 
      if (r > 0)
        {
          RTC_DEBUG(("isNew() = true, readable data: %d", r));
          return true;
        }
 
      RTC_DEBUG(("isNew() = false, no readable data"));
      return false;
    }
 
    virtual bool isEmpty(std::string name)
    {
        RTC_TRACE(("isEmpty()"));
 
 
        {
            std::lock_guard<std::mutex> guard(m_connectorsMutex);
            if (m_connectors.empty())
            {
                RTC_DEBUG(("no connectors"));
                return false;
            }
 
            for(auto & con : m_connectors)
            {
                if (std::string(con->name()) == name)
                {
                    size_t r = con->getBuffer()->readable();
                    if (r == 0)
                    {
                        RTC_DEBUG(("isEmpty() = true, buffer is empty"));
                        return true;
                    }
                }
            }
        }
 
        RTC_DEBUG(("isEmpty() = false, no readable data"));
        return false;
    }
    virtual bool isEmpty(coil::vstring &names)
    {
        names.clear();
        RTC_TRACE(("isEmpty()"));
 
 
 
        {
            std::lock_guard<std::mutex> guard(m_connectorsMutex);
            if (m_connectors.empty())
            {
                RTC_DEBUG(("no connectors"));
                return false;
            }
 
            for(auto & con : m_connectors)
            {
                size_t r = con->getBuffer()->readable();
                if (r == 0)
                {
                    names.emplace_back(con->name());
                }
            }
 
        }
 
        if (!names.empty())
        {
            RTC_DEBUG(("isEmpty() = true, buffer is empty"));
            return true;
        }
 
        RTC_DEBUG(("isEmpty() = false, no readable data"));
        return false;
    }
    bool isEmpty() override
    {
      RTC_TRACE(("isEmpty()"));
      if (m_directNewData == true) { return false; }
      size_t r(0);
 
      {
        std::lock_guard<std::mutex> guard(m_connectorsMutex);
        if (m_connectors.empty())
          {
            RTC_DEBUG(("no connectors"));
            return true;
          }
        // In single-buffer mode, all connectors share the same buffer. This
        // means that we only need to read from the first connector to get data
        // received by any connector.
        r = m_connectors[0]->getBuffer()->readable();
      }
 
      if (r == 0)
        {
          RTC_DEBUG(("isEmpty() = true, buffer is empty"));
          return true;
        }
 
      RTC_DEBUG(("isEmpty() = false, data exists in the buffer"));
      return false;
    }
 
    void write(DataType& data) override
    {
      std::lock_guard<std::mutex> guard(m_valueMutex);
      CORBA_Util::copyData<DataType>(m_value, data);
      m_directNewData = true;
    }
 
    bool read(std::string name="") override
    {
      RTC_TRACE(("DataType read()"));
 
      if (m_OnRead != nullptr)
        {
          (*m_OnRead)();
          RTC_TRACE(("OnRead called"));
        }
      // 1) direct connection
      {
        std::lock_guard<std::mutex> guard(m_valueMutex);
        if (m_directNewData == true)
          {
            RTC_DEBUG(("Direct data transfer"));
            if (m_OnReadConvert != nullptr)
              {
                m_value = (*m_OnReadConvert)(m_value);
                RTC_DEBUG(("OnReadConvert for direct data called"));
                return true;
              }
            m_directNewData = false;
            return true;
          }
      }
      // 2) network connection
      
      DataPortStatus ret;
      {
        std::lock_guard<std::mutex> guard(m_connectorsMutex);
        if (m_connectors.empty())
          {
            RTC_DEBUG(("no connectors"));
            return false;
          }
 
        
      }
 
      InPortConnector* connector = nullptr;
 
      if (name.empty())
      {
          connector = m_connectors[0];
      }
      else
      {
          for(auto & con : m_connectors)
          {
              if (std::string(con->name()) == name)
              {
                  connector = con;
              }
          }
      }
 
      if (connector == nullptr)
      {
          RTC_ERROR(("can not find %s",name.c_str()));
          return false;
      }
 
      if (!connector->getDirectData(m_value))
      {
          {
              std::lock_guard<std::mutex> guard(m_connectorsMutex);
              // In single-buffer mode, all connectors share the same buffer. This
              // means that we only need to read from the first connector to get data
              // received by any connector.
              ret = connector->read(m_value);
          }
          m_status[0] = ret;
          if (ret == DataPortStatus::PORT_OK)
          {
              std::lock_guard<std::mutex> guard(m_valueMutex);
              RTC_DEBUG(("data read succeeded"));
 
              if (m_OnReadConvert != nullptr)
              {
                  m_value = (*m_OnReadConvert)(m_value);
                  RTC_DEBUG(("OnReadConvert called"));
                  return true;
              }
              return true;
          }
          else if (ret == DataPortStatus::BUFFER_EMPTY)
          {
              RTC_WARN(("buffer empty"));
              return false;
          }
          else if (ret == DataPortStatus::BUFFER_TIMEOUT)
          {
              RTC_WARN(("buffer read timeout"));
              return false;
          }
      }
      else
      {
          return true;
      }
      RTC_ERROR(("unknown retern value from buffer.read()"));
      return false;
    }
 
 
    virtual void update()
    {
      this->read();
    }
 
    void operator>>(DataType& rhs)
    {
      this->read();
      CORBA_Util::copyData<DataType>(rhs, m_value);
      return;
    }
 
    DataPortStatus getStatus(int  /*index*/)
    {
      return m_status[0];
    }
    DataPortStatusList getStatusList()
    {
      return m_status;
    }
 
    inline void setOnRead(OnRead<DataType>* on_read)
    {
      m_OnRead = on_read;
    }
 
    inline void setOnReadConvert(OnReadConvert<DataType>* on_rconvert)
    {
      m_OnReadConvert = on_rconvert;
    }
  protected:
    void initConnectorListeners() override
    {
      delete m_listeners;
      m_listeners = new ConnectorListenersT<DataType>();
    }
  private:
    std::string m_typename;
    std::string m_name;
 
    DataType& m_value;
    mutable std::mutex m_valueMutex;
 
    OnRead<DataType>* m_OnRead;
 
    OnReadConvert<DataType>* m_OnReadConvert;
 
    DataPortStatusList m_status;
 
    bool m_directNewData;
  };
 
  template <class T> InPort<T>::~InPort() = default; // No inline for gcc warning, too big
} // namespace RTC
 
#endif  // RTC_INPORT_H