OutPortPullConnector.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 OutPortPullConnector(OpenRTM_aist.OutPortConnector):
  """
  """
 
  
  def __init__(self, info, provider, listeners, buffer = 0):
    OpenRTM_aist.OutPortConnector.__init__(self, info)
    self._provider = provider
    self._listeners = listeners
    self._buffer = buffer
 
    if not self._buffer:
      self._buffer = self.createBuffer(info)
 
    if not self._provider or not self._buffer:
      self._rtcout.RTC_ERROR("Exeption: in OutPortPullConnector.__init__().")
      raise
 
    self._buffer.init(info.properties.getNode("buffer"))
    self._provider.setBuffer(self._buffer)
    self._provider.setConnector(self)
    self._provider.setListener(info, self._listeners)
    self.onConnect()
    return
 
 
  
  def __del__(self):
    return
 
 
  
  def write(self, data):
    # 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
    
    self._buffer.write(cdr_data)
    return self.PORT_OK
 
 
  
  def disconnect(self):
    self._rtcout.RTC_TRACE("disconnect()")
    self.onDisconnect()
    # delete provider
    if self._provider:
      OpenRTM_aist.OutPortProviderFactory.instance().deleteObject(self._provider)
    self._provider = 0
 
    # delete buffer
    if self._buffer:
      OpenRTM_aist.CdrBufferFactory.instance().deleteObject(self._buffer)
    self._buffer = 0
 
    return self.PORT_OK
 
 
  
  def getBuffer(self):
    return self._buffer
 
 
  
  def activate(self):  # do nothing
    pass
 
 
  
  def deactivate(self): # do nothing
    pass
 
    
  
  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