OutPort.h

Go to the documentation of this file.

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

#include <functional>
#include <string>

#include <coil/TimeValue.h>
#include <coil/Time.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>

template <class DataType>
void setTimestamp(DataType& data)
{
      // set timestamp
      coil::TimeValue tm(coil::gettimeofday());
      data.tm.sec  = tm.sec();
      data.tm.nsec = tm.usec() * 1000;
}

namespace RTC
{
  template <class DataType>
  class OutPort
      : public OutPortBase, DirectOutPortBase<DataType>
  {
      typedef coil::Guard<coil::Mutex> Guard;
  public:
    OutPort(const char* name, DataType& value)
#if defined(__GNUC__) && (__GNUC__ <= 3 && __GNUC_MINOR__ <= 3) 
      : OutPortBase(name, ::CORBA_Util::toRepositoryIdOfStruct<DataType>()),
#else
      : OutPortBase(name, ::CORBA_Util::toRepositoryId<DataType>()),
#endif
      DirectOutPortBase<DataType>(value),
      m_value(value), m_onWrite(0), m_onWriteConvert(0),
      m_directNewData(false), m_directValue(value)
    {

      this->addConnectorDataListener(ON_BUFFER_WRITE,
                                     new Timestamp<DataType>("on_write"));
      this->addConnectorDataListener(ON_SEND,
                                     new Timestamp<DataType>("on_send"));
      m_directport = this;

    }
    
    virtual ~OutPort(void)
    {
    }
    
    virtual bool write(DataType& value)
    {
      RTC_TRACE(("DataType write()"));

      if (m_onWrite != NULL)
        {
          (*m_onWrite)(value);
          RTC_TRACE(("OnWrite called"));
        }


      bool result(true);
      std::vector<const char *> disconnect_ids;
      {
        Guard 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)
          {

            ReturnCode ret;
            if (!m_connectors[i]->pullDirectMode())
            {
                if (m_onWriteConvert != NULL)
                {
                    RTC_DEBUG(("m_connectors.OnWriteConvert called"));
                    ret = m_connectors[i]->write(((*m_onWriteConvert)(value)));
                }
                else
                {
                    RTC_DEBUG(("m_connectors.write called"));
                    ret = m_connectors[i]->write(value);
                }
            }
            else
            {
                Guard guard(m_valueMutex);
                if (m_onWriteConvert != NULL)
                {
                    RTC_DEBUG(("m_connectors.OnWriteConvert called"));
                    m_directValue = ((*m_onWriteConvert)(value));
                }
                else
                {
                    m_directValue = value;
                }
                m_directNewData = true;
                ret = PORT_OK;
            }
            m_status[i] = ret;

            if (ret == PORT_OK) { continue; }
      
            result = false;
            const char* id(m_connectors[i]->profile().id.c_str());

            if (ret == CONNECTION_LOST)
              {
                RTC_WARN(("connection_lost id: %s", id));
                if (m_onConnectionLost != 0)
                  {
                    RTC::ConnectorProfile prof(findConnProfile(id));
                    (*m_onConnectionLost)(prof);
                  }
                disconnect_ids.push_back(id);
              }
          }
      }
      std::for_each(disconnect_ids.begin(),disconnect_ids.end(),
                    std::bind1st(std::mem_fun(&PortBase::disconnect),this));
      return result;
    }
    
    bool write()
    {
      return write(m_value);
    }
    
    bool operator<<(DataType& value)
    {
      return write(value);
    }

    DataPortStatus::Enum 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;
    }


    virtual void read(DataType& data)
    {
        Guard guard(m_valueMutex);
        m_directNewData = false;
        data = m_directValue;
    }
    virtual bool isEmpty()
    {
        return !m_directNewData;
    }
    virtual bool isNew()
    {
        return m_directNewData;
    }
    
  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;

    coil::Mutex m_valueMutex;
    bool m_directNewData;
    DataType m_directValue;


  };
}; // namespace RTC

#endif // RTC_OUTPORT_H