InPort.h

Go to the documentation of this file.

// -*- C++ -*-
#ifndef RTC_INPORT_H
#define RTC_INPORT_H

#include <string>
#include <vector>
#include <iostream>

#include <coil/TimeValue.h>
#include <coil/Time.h>
#include <coil/OS.h>
#include <coil/Mutex.h>
#include <coil/Guard.h>

#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>



namespace RTC
{
  template <class DataType>
  class InPort
      : public InPortBase, DirectInPortBase<DataType>
  {
  public:
    DATAPORTSTATUS_ENUM
    InPort(const char* name, DataType& value,
       int bufsize=64, 
       bool read_block = false, bool write_block = false,
       int read_timeout = 0, int write_timeout = 0)
#if defined(__GNUC__) && (__GNUC__ <= 3 && __GNUC_MINOR__ <= 3) 
      : InPortBase(name, ::CORBA_Util::toRepositoryIdOfStruct<DataType>()),
#else
      : InPortBase(name, ::CORBA_Util::toRepositoryId<DataType>()),
#endif
      DirectInPortBase<DataType>(value),
        m_name(name), m_value(value),
        m_OnRead(NULL),  m_OnReadConvert(NULL),
        m_status(1), m_directNewData(false)
    {
      this->addConnectorDataListener(ON_RECEIVED,
                                     new Timestamp<DataType>("on_received"));
      this->addConnectorDataListener(ON_BUFFER_READ,
                                     new Timestamp<DataType>("on_read"));
      m_directport = this;
    }
    
    virtual ~InPort(void){};

    virtual const char* name()
    {
      return m_name.c_str();
    }

    
    virtual bool isNew()
    {
      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.
      {
        Guard gurad(m_valueMutex);
        if (m_directNewData == true)
          {
            RTC_DEBUG(("isNew() returns true because of direct write."));
            return true;
          }
      }
      size_t r(0);
      {
        Guard guard(m_connectorsMutex);
        if (m_connectors.size() == 0)
          {
            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()
    {
      RTC_TRACE(("isEmpty()"));
      if (m_directNewData == true) { return false; }
      size_t r(0);

      {
        Guard guard(m_connectorsMutex);
        if (m_connectors.size() == 0)
          {
            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;
    }

    virtual void write(const DataType& data)
    {
      Guard guard(m_valueMutex);
      m_value = data;
      m_directNewData = true;
    }

    bool read()
    {
      RTC_TRACE(("DataType read()"));

      if (m_OnRead != NULL) 
        {
          (*m_OnRead)();
          RTC_TRACE(("OnRead called"));
        }
      // 1) direct connection
      {
        Guard guard(m_valueMutex);
        if (m_directNewData == true)
          {
            RTC_DEBUG(("Direct data transfer"));
            if (m_OnReadConvert != 0) 
              {
                m_value = (*m_OnReadConvert)(m_value);
                RTC_DEBUG(("OnReadConvert for direct data called"));
                return true;
              }
            m_directNewData = false;
            return true;
          }
      }
      // 2) network connection
      cdrMemoryStream cdr;
      ReturnCode ret;
      {
        Guard guard(m_connectorsMutex);
        if (m_connectors.size() == 0)
          {
            RTC_DEBUG(("no connectors"));
            return false;
          }

        
      }

      if (!m_connectors[0]->getDirectData(m_value))
      {
          {
              Guard 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 = m_connectors[0]->read(cdr);
          }
          m_status[0] = ret;
          if (ret == PORT_OK)
          {
              Guard guard(m_valueMutex);
              RTC_DEBUG(("data read succeeded"));
              m_value <<= cdr;
              if (m_OnReadConvert != 0)
              {
                  m_value = (*m_OnReadConvert)(m_value);
                  RTC_DEBUG(("OnReadConvert called"));
                  return true;
              }
              return true;
          }
          else if (ret == BUFFER_EMPTY)
          {
              RTC_WARN(("buffer empty"));
              return false;
          }
          else if (ret == 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();
      rhs = m_value;
      return;
    }

    DataPortStatus::Enum 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;
    }
    
  private:
    std::string m_typename;
    std::string m_name;
    
    DataType& m_value;
    mutable coil::Mutex m_valueMutex;
    
    OnRead<DataType>* m_OnRead;
    
    OnReadConvert<DataType>* m_OnReadConvert;

    DataPortStatusList m_status;

    bool m_directNewData;

  };
}; // End of namesepace RTM

#endif // RTC_INPORT_H