PublisherNew.h

Go to the documentation of this file.

// -*- C++ -*-
 
#ifndef RTC_PUBLISHERNEW_H
#define RTC_PUBLISHERNEW_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/DataPortStatus.h>
#include <rtm/SystemLogger.h>
#include <rtm/ConnectorBase.h>
#include <rtm/ConnectorListener.h>
#include <rtm/ByteData.h>
 
namespace coil
{
  class Properties;
} // namespace coil
 
namespace RTC
{
  class InPortConsumer;

  class PublisherNew
    : public PublisherBase
  {
  public:

    PublisherNew();

    ~PublisherNew() 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 onBufferWriteOverwrite(ByteData& data)
    {
      m_listeners->notifyOut(ConnectorDataListenerType::ON_BUFFER_OVERWRITE, 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 onSenderError()
    {
      m_listeners->notify(ConnectorListenerType::ON_SENDER_ERROR, m_profile);
    }
 
 
  private:
    Logger rtclog{"PublisherNew"};
    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};
    int m_leftskip{0};
    ByteData m_data;
  };
} // namespace RTC
 
extern "C"
{
  void PublisherNewInit();
}
 
#endif  // RTC_PUBLISHERNEW_H