PublisherPeriodic.h

Go to the documentation of this file.

// -*- C++ -*-
#ifndef RTC_PUBLISHERPERIODIC_H
#define RTC_PUBLISHERPERIODIC_H
 
#include <coil/Task.h>
#include <condition_variable>
#include <coil/PeriodicTask.h>
 
#include <rtm/RTC.h>
#include <rtm/PublisherBase.h>
#include <rtm/CdrBufferBase.h>
#include <rtm/SystemLogger.h>
#include <rtm/ConnectorBase.h>
#include <rtm/ConnectorListener.h>
 
namespace coil
{
  class Properties;
} // namespace coil
 
namespace RTC
{
  class InPortConsumer;
  class PublisherPeriodic
    : public PublisherBase
  {
  public:
 
    PublisherPeriodic();
 
    ~PublisherPeriodic() override;
 
    DataPortStatus init(coil::Properties& prop) override;
 
    DataPortStatus setConsumer(InPortConsumer* consumer) override;
 
    DataPortStatus setBuffer(CdrBufferBase* buffer) override;
 
    DataPortStatus setListener(ConnectorInfo& info,
                                   ConnectorListenersBase* listeners) override;
    DataPortStatus write(ByteDataStreamBase* data,
                     std::chrono::nanoseconds timeout) override;
    bool isActive() override;
 
    DataPortStatus activate() override;
 
    DataPortStatus deactivate() override;
 
    virtual int svc();
 
  protected:
    enum Policy
      {
        PUBLISHER_POLICY_ALL,
        PUBLISHER_POLICY_FIFO,
        PUBLISHER_POLICY_SKIP,
        PUBLISHER_POLICY_NEW
      };
 
    void setPushPolicy(const coil::Properties& prop);
 
    bool createTask(const coil::Properties& prop);
 
    DataPortStatus pushAll();
 
    DataPortStatus pushFifo();
 
    DataPortStatus pushSkip();
 
    DataPortStatus pushNew();
 
    DataPortStatus convertReturn(BufferStatus status,
                                 ByteData& data);
 
 
    DataPortStatus invokeListener(DataPortStatus status,
                              ByteData& data);
 
    inline void onBufferWrite(ByteData& data)
    {
      m_listeners->notifyOut(ConnectorDataListenerType::ON_BUFFER_WRITE,
                            m_profile, data);
    }
 
    inline void onBufferFull(ByteData& data)
    {
      m_listeners->notifyOut(
        ConnectorDataListenerType::ON_BUFFER_FULL, m_profile, data);
    }
 
    inline void onBufferWriteTimeout(ByteData& data)
    {
      m_listeners->notifyOut(
        ConnectorDataListenerType::ON_BUFFER_WRITE_TIMEOUT, m_profile, data);
    }
 
    inline void onBufferRead(ByteData& data)
    {
      m_listeners->notifyOut(ConnectorDataListenerType::ON_BUFFER_READ, m_profile, data);
    }
 
    inline void onSend(ByteData& data)
    {
      m_listeners->notifyOut(ConnectorDataListenerType::ON_SEND, m_profile, data);
    }
 
    inline void onReceived(ByteData& data)
    {
      m_listeners->notifyOut(ConnectorDataListenerType::ON_RECEIVED, m_profile, data);
    }
 
    inline void onReceiverFull(ByteData& data)
    {
      m_listeners->notifyOut(ConnectorDataListenerType::ON_RECEIVER_FULL, m_profile, data);
    }
 
    inline void onReceiverTimeout(ByteData& data)
    {
      m_listeners->notifyOut(ConnectorDataListenerType::ON_RECEIVER_TIMEOUT, m_profile, data);
    }
 
    inline void onReceiverError(ByteData& data)
    {
      m_listeners->notifyOut(ConnectorDataListenerType::ON_RECEIVER_ERROR, m_profile, data);
    }
 
    inline void onBufferEmpty()
    {
      m_listeners->notify(ConnectorListenerType::ON_BUFFER_EMPTY, m_profile);
    }
 
    inline void onSenderEmpty()
    {
      m_listeners->notify(ConnectorListenerType::ON_SENDER_EMPTY, m_profile);
    }
 
    inline void onSenderError()
    {
      m_listeners->notify(ConnectorListenerType::ON_SENDER_ERROR, m_profile);
    }
 
 
  private:
    bool bufferIsEmpty()
    {
      if (m_buffer->empty() && !m_readback)
        {
          RTC_DEBUG(("buffer empty"));
          onBufferEmpty();
          onSenderEmpty();
          return true;
        }
      return false;
    }
 
    Logger rtclog{"PublisherPeriodic"};
    InPortConsumer* m_consumer{nullptr};
    CdrBufferBase* m_buffer{nullptr};
    ConnectorInfo m_profile;
    coil::PeriodicTaskBase* m_task{nullptr};
    ConnectorListenersBase* m_listeners{nullptr};
    DataPortStatus m_retcode{DataPortStatus::PORT_OK};
    std::mutex m_retmutex;
    Policy m_pushPolicy{PUBLISHER_POLICY_NEW};
    int m_skipn{0};
    bool m_active{false};
    bool m_readback{false};
    int m_leftskip{0};
    ByteData m_data;
  };
} // namespace RTC
 
extern "C"
{
  void PublisherPeriodicInit();
}
 
#endif  // RTC_PUBLISHERPERIODIC_H