OutPortPushConnector.cpp

Go to the documentation of this file.

// -*- C++ -*-
#include <coil/stringutil.h>

#include <rtm/OutPortPushConnector.h>
#include <rtm/ConnectorListener.h>

namespace RTC
{
  OutPortPushConnector::OutPortPushConnector(ConnectorInfo info, 
                                             InPortConsumer* consumer,
                                             ConnectorListeners& listeners,
                                             CdrBufferBase* buffer)
    : OutPortConnector(info),
      m_consumer(consumer), m_publisher(0),
      m_listeners(listeners), m_buffer(buffer)
  {
    // publisher/buffer creation. This may throw std::bad_alloc;
    m_publisher = createPublisher(info);
    if (m_buffer == 0)
      {
        m_buffer = createBuffer(info);
      }
    if (m_publisher == 0 || m_buffer == 0 || m_consumer == 0) 
      { throw std::bad_alloc(); }
    
    if (m_publisher->init(info.properties) != PORT_OK)
      {
        throw std::bad_alloc();
      }
    m_buffer->init(info.properties.getNode("buffer"));
    m_consumer->init(info.properties);
    
    m_publisher->setConsumer(m_consumer);
    m_publisher->setBuffer(m_buffer);
    m_publisher->setListener(m_profile, &m_listeners);

    onConnect();
  }
  
  OutPortPushConnector::~OutPortPushConnector()
  {
    onDisconnect();
    disconnect();
  }
  
  ConnectorBase::ReturnCode
  OutPortPushConnector::write(const cdrMemoryStream& data)
  {
    RTC_TRACE(("write()"));
    RTC_PARANOID(("data size = %d bytes", data.bufSize()));
    
    return m_publisher->write(data, 0, 0);
  }
  
  ConnectorBase::ReturnCode OutPortPushConnector::disconnect()
  {
    RTC_TRACE(("disconnect()"));
    // delete publisher
    if (m_publisher != 0)
      {
        RTC_DEBUG(("delete publisher"));
        PublisherFactory& pfactory(PublisherFactory::instance());
        pfactory.deleteObject(m_publisher);
      }
    m_publisher = 0;
    
    // delete consumer
    if (m_consumer != 0)
      {
        RTC_DEBUG(("delete consumer"));
        InPortConsumerFactory& cfactory(InPortConsumerFactory::instance());
        cfactory.deleteObject(m_consumer);
      }
    m_consumer = 0;
    
    // delete buffer
    if (m_buffer != 0)
      {
        RTC_DEBUG(("delete buffer"));
        CdrBufferFactory& bfactory(CdrBufferFactory::instance());
        bfactory.deleteObject(m_buffer);
      }
    m_buffer = 0;
    RTC_TRACE(("disconnect() done"));
    return PORT_OK;
  }

  void OutPortPushConnector::activate()
  {
    m_publisher->activate();
  }

  void OutPortPushConnector::deactivate()
  {
    m_publisher->deactivate();
  }
  
  CdrBufferBase* OutPortPushConnector::getBuffer()
  {
    return m_buffer;
  }
  
  PublisherBase* OutPortPushConnector::createPublisher(ConnectorInfo& info)
  {
    std::string pub_type;
    pub_type = info.properties.getProperty("subscription_type",
                                              "flush");
    coil::normalize(pub_type);
    return PublisherFactory::instance().createObject(pub_type);
  }
  
  CdrBufferBase* OutPortPushConnector::createBuffer(ConnectorInfo& info)
  {
    std::string buf_type;
    buf_type = info.properties.getProperty("buffer_type",
                                              "ring_buffer");
    return CdrBufferFactory::instance().createObject(buf_type);
  }

  void OutPortPushConnector::onConnect()
  {
    m_listeners.connector_[ON_CONNECT].notify(m_profile);
  }

  void OutPortPushConnector::onDisconnect()
  {
    m_listeners.connector_[ON_DISCONNECT].notify(m_profile);
  }
};