X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;ds=sidebyside;f=bindings%2Fpython%2Fbt2%2Fbt2%2Fport.py;h=c91d9889b4ced413856d225b53058f68ca958027;hb=894a8df566ca63e6b53c831d245569d8598c5889;hp=ebd2765045a198423d939c461d5406a3fbe161b7;hpb=31976fe2d70a8b6b7f8b31b9e0b3bc004d415575;p=babeltrace.git diff --git a/bindings/python/bt2/bt2/port.py b/bindings/python/bt2/bt2/port.py index ebd27650..c91d9889 100644 --- a/bindings/python/bt2/bt2/port.py +++ b/bindings/python/bt2/bt2/port.py @@ -20,157 +20,75 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. -from bt2 import native_bt, object, utils -import collections.abc +from bt2 import native_bt, object import bt2.component import bt2.connection import bt2.message_iterator import bt2.message -import copy import bt2 -def _create_from_ptr(ptr): - port_type = native_bt.port_get_type(ptr) - - if port_type == native_bt.PORT_TYPE_INPUT: - cls = _InputPort - elif port_type == native_bt.PORT_TYPE_OUTPUT: - cls = _OutputPort - else: - raise bt2.Error('unknown port type: {}'.format(port_type)) - - return cls._create_from_ptr(ptr) - - -def _create_private_from_ptr(ptr): - pub_ptr = native_bt.port_from_private(ptr) - utils._handle_ptr(pub_ptr, 'cannot get port object from private port object') - port_type = native_bt.port_get_type(pub_ptr) - assert(port_type == native_bt.PORT_TYPE_INPUT or port_type == native_bt.PORT_TYPE_OUTPUT) - - if port_type == native_bt.PORT_TYPE_INPUT: - cls = _PrivateInputPort - elif port_type == native_bt.PORT_TYPE_OUTPUT: - cls = _PrivateOutputPort - - obj = cls._create_from_ptr(ptr) - obj._pub_ptr = pub_ptr - return obj - - class _Port(object._SharedObject): - @staticmethod - def _name(ptr): - name = native_bt.port_get_name(ptr) - assert(name is not None) - return name - - @staticmethod - def _disconnect(ptr): - status = native_bt.port_disconnect(ptr) + @classmethod + def _get_ref(cls, ptr): + ptr = cls._as_port_ptr(ptr) + return native_bt.port_get_ref(ptr) - if status < 0: - raise bt2.Error('cannot disconnect port') + @classmethod + def _put_ref(cls, ptr): + ptr = cls._as_port_ptr(ptr) + return native_bt.port_put_ref(ptr) @property def name(self): - return self._name(self._ptr) - - @property - def component(self): - comp_ptr = native_bt.port_get_component(self._ptr) - - if comp_ptr is None: - return - - return bt2.component._create_generic_component_from_ptr(comp_ptr) + ptr = self._as_port_ptr(self._ptr) + name = native_bt.port_get_name(ptr) + assert name is not None + return name @property def connection(self): - conn_ptr = native_bt.port_get_connection(self._ptr) + ptr = self._as_port_ptr(self._ptr) + conn_ptr = native_bt.port_borrow_connection_const(ptr) if conn_ptr is None: return - return bt2.connection._Connection._create_from_ptr(conn_ptr) + return bt2.connection._Connection._create_from_ptr_and_get_ref(conn_ptr) @property def is_connected(self): return self.connection is not None - def disconnect(self): - self._disconnect(self._ptr) - - def __eq__(self, other): - if type(other) is not type(self): - return False - - return self.addr == other.addr - class _InputPort(_Port): - pass + _as_port_ptr = staticmethod(native_bt.port_input_as_port_const) class _OutputPort(_Port): - def create_message_iterator(self, message_types=None, - colander_component_name=None): - msg_types = bt2.message._msg_types_from_msg_classes(message_types) - - if colander_component_name is not None: - utils._check_str(colander_component_name) - - msg_iter_ptr = native_bt.py3_create_output_port_msg_iter(int(self._ptr), - colander_component_name, - msg_types) + _as_port_ptr = staticmethod(native_bt.port_output_as_port_const) - if msg_iter_ptr is None: - raise bt2.CreationError('cannot create output port message iterator') - return bt2.message_iterator._OutputPortMessageIterator._create_from_ptr(msg_iter_ptr) - - -class _PrivatePort(_Port): - @property - def name(self): - return self._name(self._pub_ptr) - - @property - def component(self): - comp_ptr = native_bt.private_port_get_private_component(self._ptr) - - if comp_ptr is None: - return - - pub_comp_ptr = native_bt.component_from_private(comp_ptr) - assert(pub_comp_ptr) - comp = bt2.component._create_generic_component_from_ptr(pub_comp_ptr) - native_bt.put(comp_ptr) - return comp +class _UserComponentPort(_Port): + @classmethod + def _as_port_ptr(cls, ptr): + ptr = cls._as_self_port_ptr(ptr) + return native_bt.self_component_port_as_port(ptr) @property def connection(self): - conn_ptr = native_bt.private_port_get_private_connection(self._ptr) + ptr = self._as_port_ptr(self._ptr) + conn_ptr = native_bt.port_borrow_connection_const(ptr) if conn_ptr is None: return - return bt2.connection._create_private_from_ptr(conn_ptr) - - def remove_from_component(self): - status = native_bt.private_port_remove_from_component(self._ptr) - - if status < 0: - raise bt2.Error("cannot remove port from component") - - def disconnect(self): - self._disconnect(self._pub_ptr) + return bt2.connection._Connection._create_from_ptr_and_get_ref(conn_ptr) -class _PrivateInputPort(_PrivatePort, _InputPort): - pass +class _UserComponentInputPort(_UserComponentPort, _InputPort): + _as_self_port_ptr = staticmethod(native_bt.self_component_port_input_as_self_component_port) -class _PrivateOutputPort(_PrivatePort, _OutputPort): - pass +class _UserComponentOutputPort(_UserComponentPort, _OutputPort): + _as_self_port_ptr = staticmethod(native_bt.self_component_port_output_as_self_component_port)