ConnectorListener.h

Go to the documentation of this file.

// -*- C++ -*-
#ifndef RTC_CONNECTORLISTENER_H
#define RTC_CONNECTORLISTENER_H
 
#include <mutex>
#include <rtm/RTC.h>
#include <rtm/ConnectorBase.h>
#include <rtm/ByteData.h>
#include <rtm/CORBA_CdrMemoryStream.h>
 
#include <string>
#include <vector>
#include <utility>
#include <array>
 
 
namespace RTC
{
  class ConnectorInfo;
 
 
  class ConnectorListenerStatus
  {
  public:
    enum Enum
      {
        NO_CHANGE    = 0,
        INFO_CHANGED = 1 << 0,
        DATA_CHANGED = 1 << 1,
        BOTH_CHANGED = INFO_CHANGED | DATA_CHANGED
      };
  };
  ConnectorListenerStatus::Enum operator|(ConnectorListenerStatus::Enum L,
                                          ConnectorListenerStatus::Enum R);
  ConnectorListenerStatus::Enum operator&(ConnectorListenerStatus::Enum L,
                                          ConnectorListenerStatus::Enum R);
 
#define USE_CONNLISTENER_STATUS                             \
  using ReturnCode = ::RTC::ConnectorListenerStatus::Enum;  \
  using ::RTC::ConnectorListenerStatus::NO_CHANGE;          \
  using ::RTC::ConnectorListenerStatus::INFO_CHANGED;       \
  using ::RTC::ConnectorListenerStatus::DATA_CHANGED;       \
  using ::RTC::ConnectorListenerStatus::BOTH_CHANGED
 
  enum class ConnectorDataListenerType : uint8_t
    {
      ON_BUFFER_WRITE = 0,
      ON_BUFFER_FULL,
      ON_BUFFER_WRITE_TIMEOUT,
      ON_BUFFER_OVERWRITE,
      ON_BUFFER_READ,
      ON_SEND,
      ON_RECEIVED,
      ON_RECEIVER_FULL,
      ON_RECEIVER_TIMEOUT,
      ON_RECEIVER_ERROR,
      CONNECTOR_DATA_LISTENER_NUM
    };
 
  class ConnectorDataListener
    : public ConnectorListenerStatus
  {
  public:
    USE_CONNLISTENER_STATUS;
    static const char* toString(ConnectorDataListenerType type)
    {
      if (type < ConnectorDataListenerType::CONNECTOR_DATA_LISTENER_NUM)
        {
          static const char* const typeString[] =
          {
            "ON_BUFFER_WRITE",
            "ON_BUFFER_FULL",
            "ON_BUFFER_WRITE_TIMEOUT",
            "ON_BUFFER_OVERWRITE",
            "ON_BUFFER_READ",
            "ON_SEND",
            "ON_RECEIVED",
            "ON_RECEIVER_FULL",
            "ON_RECEIVER_TIMEOUT",
            "ON_RECEIVER_ERROR",
            "CONNECTOR_DATA_LISTENER_NUM"
          };
          return typeString[static_cast<uint8_t>(type)];
        }
      return "";
    }
 
    virtual ~ConnectorDataListener();
 
    virtual ReturnCode operator()(ConnectorInfo& info,
                            ByteData& data, const std::string& marshalingtype) = 0;
  };
 
  template <class DataType>
  class ConnectorDataListenerT
    : public ConnectorDataListener
  {
  public:
    ConnectorDataListenerT() = default;
    ~ConnectorDataListenerT() override
    {
        SerializerFactory::instance().deleteObject(m_cdr);
    }
 
    ReturnCode operator()(ConnectorInfo& info,
                                  ByteData& cdrdata, const std::string& marshalingtype) override
    {
      DataType data;
 
      if(m_cdr == nullptr || m_marshalingtype != marshalingtype)
      {
        m_cdr = createSerializer<DataType>(marshalingtype);
        m_marshalingtype = marshalingtype;
      }
      ::RTC::ByteDataStream<DataType> *cdr = dynamic_cast<::RTC::ByteDataStream<DataType>*>(m_cdr);
      
 
      if (!cdr)
      {
          return NO_CHANGE;
      }
 
      
      // endian type check
      std::string endian_type{coil::normalize(
        info.properties.getProperty("serializer.cdr.endian", "little"))};
      std::vector<std::string> endian(coil::split(endian_type, ","));
 
      if (endian[0] == "little")
      {
          cdr->isLittleEndian(true);
      }
      else if (endian[0] == "big")
      {
          cdr->isLittleEndian(false);
      }
 
 
      cdr->writeData(cdrdata.getBuffer(), cdrdata.getDataLength());
 
      cdr->deserialize(data);
 
      ReturnCode ret = this->operator()(info, data);
      if (ret == DATA_CHANGED || ret == BOTH_CHANGED)
      {
          if (endian[0] == "little")
          {
              cdr->isLittleEndian(true);
          }
          else if (endian[0] == "big")
          {
              cdr->isLittleEndian(false);
          }
 
          cdr->serialize(data);
          cdrdata.setDataLength(cdr->getDataLength());
          cdr->readData(cdrdata.getBuffer(), cdrdata.getDataLength());
      }
 
      return ret;
    }
 
    virtual ReturnCode operator()(ConnectorInfo& info,
                                 DataType& data) = 0;
  private:
      ByteDataStreamBase* m_cdr{nullptr};
      std::string m_marshalingtype;
  };
 
  enum class ConnectorListenerType : uint8_t
    {
      ON_BUFFER_EMPTY = 0,
      ON_BUFFER_READ_TIMEOUT,
      ON_SENDER_EMPTY,
      ON_SENDER_TIMEOUT,
      ON_SENDER_ERROR,
      ON_CONNECT,
      ON_DISCONNECT,
      CONNECTOR_LISTENER_NUM
    };
 
  class ConnectorListener
    : public ConnectorListenerStatus
  {
  public:
    USE_CONNLISTENER_STATUS;
    static const char* toString(ConnectorListenerType type)
    {
      if (type < ConnectorListenerType::CONNECTOR_LISTENER_NUM)
        {
          static const char* const typeStr[] =
          {
            "ON_BUFFER_EMPTY",
            "ON_BUFFER_READ_TIMEOUT",
            "ON_SENDER_EMPTY",
            "ON_SENDER_TIMEOUT",
            "ON_SENDER_ERROR",
            "ON_CONNECT",
            "ON_DISCONNECT",
            "CONNECTOR_LISTENER_NUM"
          };
          return typeStr[static_cast<uint8_t>(type)];
        }
      return "";
    }
 
    virtual ~ConnectorListener();
 
    virtual ReturnCode operator()(ConnectorInfo& info) = 0;
  };
 
 
  class ConnectorDataListenerHolder
    : public ConnectorListenerStatus
  {
    using Entry = std::pair<ConnectorDataListener*, bool>;
  public:
    USE_CONNLISTENER_STATUS;
    ConnectorDataListenerHolder();
    virtual ~ConnectorDataListenerHolder();
 
    void addListener(ConnectorDataListener* listener, bool autoclean);
 
    void removeListener(ConnectorDataListener* listener);
 
    size_t size();
 
    virtual ReturnCode notify(ConnectorInfo& info,
                ByteData& cdrdata, const std::string& marshalingtype);
 
 
    virtual ReturnCode notifyIn(ConnectorInfo& info, ByteData& data);
 
    virtual ReturnCode notifyOut(ConnectorInfo& info, ByteData& data);
 
 
    template <class DataType>
    ReturnCode notifyIn(ConnectorInfo& info, DataType& typeddata)
    {
        std::string type = info.properties.getProperty("marshaling_type", "cdr");
        std::string marshaling_type{coil::eraseBothEndsBlank(
          info.properties.getProperty("inport.marshaling_type", type))};
        return notify(info, typeddata, marshaling_type);
    }
 
    template <class DataType>
    ReturnCode notifyOut(ConnectorInfo& info, DataType& typeddata)
    {
        std::string type = info.properties.getProperty("marshaling_type", "cdr");
        std::string marshaling_type{coil::eraseBothEndsBlank(
          info.properties.getProperty("outport.marshaling_type", type))};
        return notify(info, typeddata, marshaling_type);
    }
    template <class DataType>
    ReturnCode notify(ConnectorInfo& info, DataType& typeddata, const std::string& marshalingtype)
    {
      std::lock_guard<std::mutex> guard(m_mutex);
      ReturnCode ret(NO_CHANGE);
 
      if(m_listeners.empty())
      {
        return ret;
      }
 
      std::string endian_type{coil::normalize(
        info.properties.getProperty("serializer.cdr.endian", "little"))};
      std::vector<std::string> endian(coil::split(endian_type, ","));
 
      for (auto & listener : m_listeners)
        {
          ConnectorDataListenerT<DataType>* datalistener(nullptr);
          datalistener =
          dynamic_cast<ConnectorDataListenerT<DataType>*>(listener.first);
          if (datalistener != nullptr)
            {
              ret = ret | datalistener->operator()(info, typeddata);
            }
          else
            {
              if (m_cdr == nullptr || m_marshalingtype != marshalingtype)
              {
                  m_cdr = createSerializer<DataType>(marshalingtype);
                  m_marshalingtype = marshalingtype;
              }
              ::RTC::ByteDataStream<DataType> *cdr = dynamic_cast<::RTC::ByteDataStream<DataType>*>(m_cdr);
 
              if (!cdr)
              {
                  return NO_CHANGE;
              }
 
              if (endian[0] == "little")
              {
                  cdr->isLittleEndian(true);
              }
              else if (endian[0] == "big")
              {
                  cdr->isLittleEndian(false);
              }
              cdr->serialize(typeddata);
              ByteData tmp = *cdr;
              ret = ret | listener.first->operator()(info, tmp, marshalingtype);
 
            }
        }
      return ret;
    }
 
  protected:
    std::vector<Entry> m_listeners;
    std::mutex m_mutex;
    ByteDataStreamBase* m_cdr{ nullptr };
    std::string m_marshalingtype;
  };
 
  class ConnectorListenerHolder
      : public ConnectorListenerStatus
  {
    using Entry = std::pair<ConnectorListener*, bool>;
  public:
    USE_CONNLISTENER_STATUS;
    ConnectorListenerHolder();
 
    virtual ~ConnectorListenerHolder();
 
    void addListener(ConnectorListener* listener, bool autoclean);
 
    void removeListener(ConnectorListener* listener);
 
    size_t size();
 
    ReturnCode notify(ConnectorInfo& info);
 
  private:
    std::vector<Entry> m_listeners;
    std::mutex m_mutex;
  };
 
  class ConnectorListenersBase
  {
  public:
    virtual ~ConnectorListenersBase();
    virtual ::RTC::ConnectorListenerStatus::Enum notifyIn(ConnectorDataListenerType type, ConnectorInfo& info, ByteData& data) = 0;
    virtual ::RTC::ConnectorListenerStatus::Enum notifyOut(ConnectorDataListenerType type, ConnectorInfo& info, ByteData& data) = 0;
    virtual ::RTC::ConnectorListenerStatus::Enum notify(ConnectorListenerType type, ConnectorInfo& info) = 0;
    virtual bool addListener(ConnectorDataListenerType type, ConnectorDataListener* listener, bool autoclean=true) = 0;
    virtual bool addListener(ConnectorListenerType type, ConnectorListener* listener, bool autoclean=true) = 0;
    virtual bool removeListener(ConnectorDataListenerType type, ConnectorDataListener* listener) = 0;
    virtual bool removeListener(ConnectorListenerType type, ConnectorListener* listener) = 0;
    virtual ConnectorDataListenerHolder* getDataListenerHolder(ConnectorDataListenerType type) = 0;
    template<class DataType> ::RTC::ConnectorListenerStatus::Enum notifyIn(ConnectorDataListenerType type, ConnectorInfo& info, DataType& data)
    {
        ConnectorDataListenerHolder* holder = getDataListenerHolder(type);
        if (holder != nullptr)
        {
            return holder->notifyIn(info, data);
        }
        return ConnectorListenerStatus::NO_CHANGE;
    }
    template<class DataType> ::RTC::ConnectorListenerStatus::Enum notifyOut(ConnectorDataListenerType type, ConnectorInfo& info, DataType& data)
    {
        ConnectorDataListenerHolder* holder = getDataListenerHolder(type);
        if (holder != nullptr)
        {
            return holder->notifyOut(info, data);
        }
        return ConnectorListenerStatus::NO_CHANGE;
    }
  };
 
  class ConnectorListeners
       : public ConnectorListenersBase
  {
  public:
    ConnectorListeners();
    ~ConnectorListeners() override;
    ::RTC::ConnectorListenerStatus::Enum notifyIn(ConnectorDataListenerType type, ConnectorInfo& info, ByteData& data) override;
    ::RTC::ConnectorListenerStatus::Enum notifyOut(ConnectorDataListenerType type, ConnectorInfo& info, ByteData& data) override;
    ::RTC::ConnectorListenerStatus::Enum notify(ConnectorListenerType type, ConnectorInfo& info) override;
    bool addListener(ConnectorDataListenerType type, ConnectorDataListener* listener, bool autoclean=true) override;
    bool addListener(ConnectorListenerType type, ConnectorListener* listener, bool autoclean=true) override;
    bool removeListener(ConnectorDataListenerType type, ConnectorDataListener* listener) override;
    bool removeListener(ConnectorListenerType type, ConnectorListener* listener) override;
    ConnectorDataListenerHolder* getDataListenerHolder(ConnectorDataListenerType type) override;
  
  private:
    std::array<ConnectorDataListenerHolder,
               static_cast<uint8_t>
               (ConnectorDataListenerType::CONNECTOR_DATA_LISTENER_NUM)>
               connectorData_;
    std::array<ConnectorListenerHolder,
               static_cast<uint8_t>
               (ConnectorListenerType::CONNECTOR_LISTENER_NUM)>
               connector_;
  };
 
 
  template <class DataType>
  class ConnectorDataListenerHolderT
      : public ConnectorDataListenerHolder
  {
  public:
      ConnectorDataListenerHolderT() = default;
      ~ConnectorDataListenerHolderT() override
      {
 
      }
 
 
 
      ReturnCode notify(ConnectorInfo& info,
          ByteData& cdrdata, const std::string& marshalingtype) override
      {
          std::lock_guard<std::mutex> guard(m_mutex);
          ConnectorListenerHolder::ReturnCode ret(NO_CHANGE);
          bool endian = true;
 
          if(m_listeners.empty())
          {
            return ret;
          }
 
          DataType data;
 
          if (m_cdr == nullptr || m_marshalingtype != marshalingtype)
          {
              m_cdr = createSerializer<DataType>(marshalingtype);
              m_marshalingtype = marshalingtype;
          }
          ::RTC::ByteDataStream<DataType> *cdr = dynamic_cast<::RTC::ByteDataStream<DataType>*>(m_cdr);
 
 
          if (!cdr)
          {
              return NO_CHANGE;
          }
 
 
          // endian type check
          std::string endian_type{ coil::normalize(
            info.properties.getProperty("serializer.cdr.endian", "little")) };
          std::vector<std::string> endian_str(coil::split(endian_type, ","));
 
          if (endian_str[0] == "little")
          {
              endian = true;
          }
          else if (endian_str[0] == "big")
          {
              endian = false;
          }
 
          cdr->isLittleEndian(endian);
          cdr->writeData(cdrdata.getBuffer(), cdrdata.getDataLength());
          cdr->deserialize(data);
 
 
          for (auto & listener : m_listeners)
          {
              ConnectorDataListenerT<DataType>* datalistener(nullptr);
              datalistener =
                  dynamic_cast<ConnectorDataListenerT<DataType>*>(listener.first);
              if (datalistener != nullptr)
              {
                  ConnectorListenerHolder::ReturnCode linstener_ret(datalistener->operator()(info, data));
                  if (linstener_ret == DATA_CHANGED || linstener_ret == BOTH_CHANGED)
                  {
                      cdr->isLittleEndian(endian);
                      cdr->serialize(data);
                      cdrdata.setDataLength(cdr->getDataLength());
                      cdr->readData(cdrdata.getBuffer(), cdrdata.getDataLength());
                  }
                  ret = ret | linstener_ret;
              }
              else
              {
                  ConnectorListenerHolder::ReturnCode linstener_ret(listener.first->operator()(info, cdrdata, marshalingtype));
                  if (linstener_ret == DATA_CHANGED || linstener_ret == BOTH_CHANGED)
                  {
                      cdr->isLittleEndian(endian);
                      cdr->writeData(cdrdata.getBuffer(), cdrdata.getDataLength());
                      cdr->deserialize(data);
                  }
                  ret = ret | linstener_ret;
              }
          }
 
          return ret;
      }
 
 
      ReturnCode notifyIn(ConnectorInfo& info, ByteData& data) override
      {
          std::string type = info.properties.getProperty("marshaling_type", "cdr");
          std::string marshaling_type{ coil::eraseBothEndsBlank(
            info.properties.getProperty("inport.marshaling_type", type)) };
          return notify(info, data, marshaling_type);
      }
 
      ReturnCode notifyOut(ConnectorInfo& info, ByteData& data) override
      {
          std::string type = info.properties.getProperty("marshaling_type", "cdr");
          std::string marshaling_type{ coil::eraseBothEndsBlank(
            info.properties.getProperty("outport.marshaling_type", type)) };
          return notify(info, data, marshaling_type);
      }
 
  };
 
  template <class DataType>
  class ConnectorListenersT
       : public ConnectorListenersBase
  {
  public:
    ConnectorListenersT()
    {
    }
    ~ConnectorListenersT() override {}
 
    ::RTC::ConnectorListenerStatus::Enum notifyIn(ConnectorDataListenerType type, ConnectorInfo& info, ByteData& data) override
    {
      if (static_cast<uint8_t>(type) < connectorData_.size())
      {
          return connectorData_[static_cast<uint8_t>(type)].notifyIn(info, data);
      }
      return ConnectorListenerStatus::NO_CHANGE;
    }
    ::RTC::ConnectorListenerStatus::Enum notifyOut(ConnectorDataListenerType type, ConnectorInfo& info, ByteData& data) override
    {
        if (static_cast<uint8_t>(type) < connectorData_.size())
        {
            return connectorData_[static_cast<uint8_t>(type)].notifyOut(info, data);
        }
        return ConnectorListenerStatus::NO_CHANGE;
    }
    ::RTC::ConnectorListenerStatus::Enum notify(ConnectorListenerType type, ConnectorInfo& info) override
    {
        if (static_cast<uint8_t>(type) < connector_.size())
        {
            return connector_[static_cast<uint8_t>(type)].notify(info);
        }
        return ConnectorListenerStatus::NO_CHANGE;
    }
    bool addListener(ConnectorDataListenerType type, ConnectorDataListener* listener, bool autoclean=true) override
    {
        if (static_cast<uint8_t>(type) < connectorData_.size())
        {
            connectorData_[static_cast<uint8_t>(type)].addListener(listener, autoclean);
            return true;
        }
        return false;
    }
    bool addListener(ConnectorListenerType type, ConnectorListener* listener, bool autoclean=true) override
    {
        if (static_cast<uint8_t>(type) < connector_.size())
        {
            connector_[static_cast<uint8_t>(type)].addListener(listener, autoclean);
            return true;
        }
        return false;
    }
    bool removeListener(ConnectorDataListenerType type, ConnectorDataListener* listener) override
    {
        if (static_cast<uint8_t>(type) < connectorData_.size())
        {
            connectorData_[static_cast<uint8_t>(type)].removeListener(listener);
            return true;
        }
        return false;
    }
    bool removeListener(ConnectorListenerType type, ConnectorListener* listener) override
    {
        if (static_cast<uint8_t>(type) < connector_.size())
        {
            connector_[static_cast<uint8_t>(type)].removeListener(listener);
            return true;
        }
        return false;
    }
    ConnectorDataListenerHolder* getDataListenerHolder(ConnectorDataListenerType type) override
    {
        if (static_cast<uint8_t>(type) < connectorData_.size())
        {
            return &connectorData_[static_cast<uint8_t>(type)];
        }
        return nullptr;
    }
 
  private:
    std::array<ConnectorDataListenerHolder,
               static_cast<uint8_t>
               (ConnectorDataListenerType::CONNECTOR_DATA_LISTENER_NUM)>
               connectorData_;
    std::array<ConnectorListenerHolder,
               static_cast<uint8_t>
               (ConnectorListenerType::CONNECTOR_LISTENER_NUM)>
               connector_;
 
  };
} // namespace RTC
 
#endif // RTC_CONNECTORLISTENER_H