InPortPullConnector.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 InPortPullConnector(OpenRTM_aist.InPortConnector):
  """
  """
    
  
  def __init__(self, info, consumer, listeners, buffer = 0):
    OpenRTM_aist.InPortConnector.__init__(self, info, buffer)
    self._consumer = consumer
    self._listeners = listeners
    if buffer == 0:
      self._buffer = self.createBuffer(self._profile)
 
    if self._buffer == 0 or not self._consumer:
      raise
        
    self._buffer.init(info.properties.getNode("buffer"))
    self._consumer.setBuffer(self._buffer)
    self._consumer.setListener(info, self._listeners)
    self.onConnect()
    return
 
 
  
  def __del__(self):
    return
 
 
  
  def read(self, data):
    self._rtcout.RTC_TRACE("InPortPullConnector.read()")
    if not self._consumer:
      return self.PORT_ERROR
 
    cdr_data = [None]
    ret = self._consumer.get(cdr_data)
 
    if ret == self.PORT_OK:
      # CDR -> (conversion) -> data
      if self._endian is not None:
        data[0] = cdrUnmarshal(any.to_any(data[0]).typecode(),cdr_data[0],self._endian)
 
      else:
        self._rtcout.RTC_ERROR("unknown endian from connector")
        return OpenRTM_aist.BufferStatus.PRECONDITION_NOT_MET
    
    return ret
 
 
  
  def disconnect(self):
    self._rtcout.RTC_TRACE("disconnect()")
    self.onDisconnect()
    # delete consumer
    if self._consumer:
      OpenRTM_aist.OutPortConsumerFactory.instance().deleteObject(self._consumer)
    self._consumer = 0
 
    return self.PORT_OK
        
  
  def activate(self): # do nothing
    pass
 
  
  def deactivate(self): # do nothing
    pass
  
  
  def createBuffer(self, profile):
    buf_type = profile.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