OutPortPushConnector.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- coding: euc-jp -*-




from omniORB import *
from omniORB import any

import OpenRTM_aist



class OutPortPushConnector(OpenRTM_aist.OutPortConnector):
  """
  """

  
  def __init__(self, info, consumer, listeners, buffer = 0):
    OpenRTM_aist.OutPortConnector.__init__(self, info)

    self._buffer = buffer
    self._consumer = consumer
    self._listeners = listeners

    # publisher/buffer creation. This may throw std::bad_alloc;
    self._publisher = self.createPublisher(info)
    if not self._buffer:
      self._buffer = self.createBuffer(info)


    if not self._publisher or not self._buffer or not self._consumer:
      raise

    if self._publisher.init(info.properties) != self.PORT_OK:
      raise
        
    if self._profile.properties.hasKey("serializer"):
      endian = self._profile.properties.getProperty("serializer.cdr.endian")
      if not endian:
        self._rtcout.RTC_ERROR("write(): endian is not set.")
        raise
        
      endian = OpenRTM_aist.split(endian, ",") # Maybe endian is ["little","big"]
      endian = OpenRTM_aist.normalize(endian) # Maybe self._endian is "little" or "big"
      if endian == "little":
        self._endian = True
      elif endian == "big":
        self._endian = False
      else:
        self._endian = None

    else:
      self._endian = True # little endian

    self._buffer.init(info.properties.getNode("buffer"))
    self._consumer.init(info.properties)
    self._publisher.setConsumer(self._consumer)
    self._publisher.setBuffer(self._buffer)
    self._publisher.setListener(self._profile, self._listeners)

    self.onConnect()
    return


  
  def __del__(self):
    return

  
  def write(self, data):
    self._rtcout.RTC_TRACE("write()")

    # data -> (conversion) -> CDR stream
    cdr_data = None
    if self._endian is not None:
      cdr_data = cdrMarshal(any.to_any(data).typecode(), data, self._endian)
    else:
      self._rtcout.RTC_ERROR("write(): endian %s is not support.",self._endian)
      return self.UNKNOWN_ERROR

    return self._publisher.write(cdr_data, 0, 0)


  
  def disconnect(self):
    self._rtcout.RTC_TRACE("disconnect()")
    self.onDisconnect()
    # delete publisher
    if self._publisher:
      self._rtcout.RTC_DEBUG("delete publisher")
      pfactory = OpenRTM_aist.PublisherFactory.instance()
      pfactory.deleteObject(self._publisher)

    self._publisher = 0
    
    # delete consumer
    if self._consumer:
      self._rtcout.RTC_DEBUG("delete consumer")
      cfactory = OpenRTM_aist.InPortConsumerFactory.instance()
      cfactory.deleteObject(self._consumer)

    self._consumer = 0

    # delete buffer
    if self._buffer:
      self._rtcout.RTC_DEBUG("delete buffer")
      bfactory = OpenRTM_aist.CdrBufferFactory.instance()
      bfactory.deleteObject(self._buffer)

    self._buffer = 0
    self._rtcout.RTC_TRACE("disconnect() done")

    return self.PORT_OK


  
  def activate(self):
    self._publisher.activate()
    return


  
  def deactivate(self):
    self._publisher.deactivate()
    return

    
  
  def getBuffer(self):
    return self._buffer


  
  def createPublisher(self, info):
    pub_type = info.properties.getProperty("subscription_type","flush")
    pub_type = OpenRTM_aist.normalize([pub_type])
    return OpenRTM_aist.PublisherFactory.instance().createObject(pub_type)


  
  def createBuffer(self, info):
    buf_type = info.properties.getProperty("buffer_type",
                                           "ring_buffer")

    return OpenRTM_aist.CdrBufferFactory.instance().createObject(buf_type)

    
  
  def onConnect(self):
    if self._listeners and self._profile:
      self._listeners.connector_[OpenRTM_aist.ConnectorListenerType.ON_CONNECT].notify(self._profile)
    return

  
  def onDisconnect(self):
    if self._listeners and self._profile:
      self._listeners.connector_[OpenRTM_aist.ConnectorListenerType.ON_DISCONNECT].notify(self._profile)
    return