# THE SOFTWARE.
from bt2 import native_bt, object, utils
-import bt2.notification_iterator
+import bt2.message_iterator
import collections.abc
-import bt2.values
+import bt2.value
import traceback
import bt2.port
import sys
# have been created by Python code, but since we only have the pointer,
# we can only wrap it in a generic way and lose the original Python
# class.
-class _GenericComponentClass(object._Object):
+#
+# Subclasses must implement some methods that this base class uses:
+#
+# - _as_component_class_ptr: static method, convert the passed component class
+# pointer to a 'bt_component_class *'.
+
+class _GenericComponentClass(object._SharedObject):
@property
def name(self):
- name = native_bt.component_class_get_name(self._ptr)
- assert(name is not None)
+ ptr = self._as_component_class_ptr(self._ptr)
+ name = native_bt.component_class_get_name(ptr)
+ assert name is not None
return name
@property
def description(self):
- return native_bt.component_class_get_description(self._ptr)
+ ptr = self._as_component_class_ptr(self._ptr)
+ return native_bt.component_class_get_description(ptr)
@property
def help(self):
- return native_bt.component_class_get_help(self._ptr)
+ ptr = self._as_component_class_ptr(self._ptr)
+ return native_bt.component_class_get_help(ptr)
+
+ def _component_class_ptr(self):
+ return self._as_component_class_ptr(self._ptr)
def __eq__(self, other):
if not isinstance(other, _GenericComponentClass):
class _GenericSourceComponentClass(_GenericComponentClass):
- pass
+ _get_ref = staticmethod(native_bt.component_class_source_get_ref)
+ _put_ref = staticmethod(native_bt.component_class_source_put_ref)
+ _as_component_class_ptr = staticmethod(native_bt.component_class_source_as_component_class)
class _GenericFilterComponentClass(_GenericComponentClass):
- pass
+ _get_ref = staticmethod(native_bt.component_class_filter_get_ref)
+ _put_ref = staticmethod(native_bt.component_class_filter_put_ref)
+ _as_component_class_ptr = staticmethod(native_bt.component_class_filter_as_component_class)
class _GenericSinkComponentClass(_GenericComponentClass):
- pass
+ _get_ref = staticmethod(native_bt.component_class_sink_get_ref)
+ _put_ref = staticmethod(native_bt.component_class_sink_put_ref)
+ _as_component_class_ptr = staticmethod(native_bt.component_class_sink_as_component_class)
def _handle_component_status(status, gen_error_msg):
- if status == native_bt.COMPONENT_STATUS_END:
+ if status == native_bt.SELF_COMPONENT_STATUS_END:
raise bt2.Stop
- elif status == native_bt.COMPONENT_STATUS_AGAIN:
+ elif status == native_bt.SELF_COMPONENT_STATUS_AGAIN:
raise bt2.TryAgain
- elif status == native_bt.COMPONENT_STATUS_UNSUPPORTED:
- raise bt2.UnsupportedFeature
- elif status == native_bt.COMPONENT_STATUS_REFUSE_PORT_CONNECTION:
+ elif status == native_bt.SELF_COMPONENT_STATUS_REFUSE_PORT_CONNECTION:
raise bt2.PortConnectionRefused
- elif status == native_bt.COMPONENT_STATUS_GRAPH_IS_CANCELED:
- raise bt2.GraphCanceled
elif status < 0:
raise bt2.Error(gen_error_msg)
raise StopIteration
comp_ports = self._comp_ports
- comp_ptr = comp_ports._component._ptr
- port_ptr = comp_ports._get_port_at_index_fn(comp_ptr, self._at)
- assert(port_ptr)
-
- if comp_ports._is_private:
- port_pub_ptr = native_bt.port_from_private_port(port_ptr)
- name = native_bt.port_get_name(port_pub_ptr)
- native_bt.put(port_pub_ptr)
- else:
- name = native_bt.port_get_name(port_ptr)
+ comp_ptr = comp_ports._component_ptr
+
+ port_ptr = comp_ports._borrow_port_ptr_at_index(comp_ptr, self._at)
+ assert port_ptr is not None
+
+ name = native_bt.port_get_name(comp_ports._port_pycls._as_port_ptr(port_ptr))
+ assert name is not None
- assert(name is not None)
- native_bt.put(port_ptr)
self._at += 1
return name
class _ComponentPorts(collections.abc.Mapping):
- def __init__(self, is_private, component,
- get_port_by_name_fn, get_port_at_index_fn,
- get_port_count_fn):
- self._is_private = is_private
- self._component = component
- self._get_port_by_name_fn = get_port_by_name_fn
- self._get_port_at_index_fn = get_port_at_index_fn
- self._get_port_count_fn = get_port_count_fn
+
+ # component_ptr is a bt_component_source *, bt_component_filter * or
+ # bt_component_sink *. Its type must match the type expected by the
+ # functions passed as arguments.
+
+ def __init__(self, component_ptr,
+ borrow_port_ptr_by_name,
+ borrow_port_ptr_at_index,
+ get_port_count,
+ port_pycls):
+ self._component_ptr = component_ptr
+ self._borrow_port_ptr_by_name = borrow_port_ptr_by_name
+ self._borrow_port_ptr_at_index = borrow_port_ptr_at_index
+ self._get_port_count = get_port_count
+ self._port_pycls = port_pycls
def __getitem__(self, key):
utils._check_str(key)
- port_ptr = self._get_port_by_name_fn(self._component._ptr, key)
+ port_ptr = self._borrow_port_ptr_by_name(self._component_ptr, key)
if port_ptr is None:
raise KeyError(key)
- if self._is_private:
- return bt2.port._create_private_from_ptr(port_ptr)
- else:
- return bt2.port._create_from_ptr(port_ptr)
+ return self._port_pycls._create_from_ptr_and_get_ref(port_ptr)
def __len__(self):
- if self._is_private:
- pub_ptr = native_bt.component_from_private_component(self._component._ptr)
- count = self._get_port_count_fn(pub_ptr)
- native_bt.put(pub_ptr)
- else:
- count = self._get_port_count_fn(self._component._ptr)
-
- assert(count >= 0)
+ count = self._get_port_count(self._component_ptr)
+ assert count >= 0
return count
def __iter__(self):
# This class holds the methods which are common to both generic
-# component objects and Python user component objects. They use the
-# internal native _ptr, however it was set, to call native API
-# functions.
+# component objects and Python user component objects.
+#
+# Subclasses must provide these methods or property:
+#
+# - _borrow_component_class_ptr: static method, must return a pointer to the
+# specialized component class (e.g. 'bt_component_class_sink *') of the
+# passed specialized component pointer (e.g. 'bt_component_sink *').
+# - _comp_cls_type: property, one of the native_bt.COMPONENT_CLASS_TYPE_*
+# constants.
+# - _as_component_ptr: static method, must return the passed specialized
+# component pointer (e.g. 'bt_component_sink *') as a 'bt_component *'.
+
class _Component:
@property
def name(self):
- name = native_bt.component_get_name(self._ptr)
- assert(name is not None)
+ ptr = self._as_component_ptr(self._ptr)
+ name = native_bt.component_get_name(ptr)
+ assert name is not None
return name
@property
- def graph(self):
- ptr = native_bt.component_get_graph(self._ptr)
- assert(ptr)
- return bt2.Graph._create_from_ptr(ptr)
-
- @property
- def component_class(self):
- cc_ptr = native_bt.component_get_class(self._ptr)
- assert(cc_ptr)
- return _create_generic_component_class_from_ptr(cc_ptr)
+ def cls(self):
+ cc_ptr = self._borrow_component_class_ptr(self._ptr)
+ assert cc_ptr is not None
+ return _create_component_class_from_ptr_and_get_ref(cc_ptr, self._comp_cls_type)
def __eq__(self, other):
if not hasattr(other, 'addr'):
class _SourceComponent(_Component):
- pass
+ _borrow_component_class_ptr = staticmethod(native_bt.component_source_borrow_class_const)
+ _comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
+ _as_component_class_ptr = staticmethod(native_bt.component_class_source_as_component_class)
+ _as_component_ptr = staticmethod(native_bt.component_source_as_component_const)
class _FilterComponent(_Component):
- pass
+ _borrow_component_class_ptr = staticmethod(native_bt.component_filter_borrow_class_const)
+ _comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
+ _as_component_class_ptr = staticmethod(native_bt.component_class_filter_as_component_class)
+ _as_component_ptr = staticmethod(native_bt.component_filter_as_component_const)
class _SinkComponent(_Component):
- pass
+ _borrow_component_class_ptr = staticmethod(native_bt.component_sink_borrow_class_const)
+ _comp_cls_type = native_bt.COMPONENT_CLASS_TYPE_SINK
+ _as_component_class_ptr = staticmethod(native_bt.component_class_sink_as_component_class)
+ _as_component_ptr = staticmethod(native_bt.component_sink_as_component_const)
# This is analogous to _GenericSourceComponentClass, but for source
# component objects.
-class _GenericSourceComponent(object._Object, _SourceComponent):
+class _GenericSourceComponent(object._SharedObject, _SourceComponent):
+ _get_ref = staticmethod(native_bt.component_source_get_ref)
+ _put_ref = staticmethod(native_bt.component_source_put_ref)
+
@property
def output_ports(self):
- return _ComponentPorts(False, self,
- native_bt.component_source_get_output_port_by_name,
- native_bt.component_source_get_output_port_by_index,
- native_bt.component_source_get_output_port_count)
+ return _ComponentPorts(self._ptr,
+ native_bt.component_source_borrow_output_port_by_name_const,
+ native_bt.component_source_borrow_output_port_by_index_const,
+ native_bt.component_source_get_output_port_count,
+ bt2.port._OutputPort)
# This is analogous to _GenericFilterComponentClass, but for filter
# component objects.
-class _GenericFilterComponent(object._Object, _FilterComponent):
+class _GenericFilterComponent(object._SharedObject, _FilterComponent):
+ _get_ref = staticmethod(native_bt.component_filter_get_ref)
+ _put_ref = staticmethod(native_bt.component_filter_put_ref)
+
@property
def output_ports(self):
- return _ComponentPorts(False, self,
- native_bt.component_filter_get_output_port_by_name,
- native_bt.component_filter_get_output_port_by_index,
- native_bt.component_filter_get_output_port_count)
+ return _ComponentPorts(self._ptr,
+ native_bt.component_filter_borrow_output_port_by_name_const,
+ native_bt.component_filter_borrow_output_port_by_index_const,
+ native_bt.component_filter_get_output_port_count,
+ bt2.port._OutputPort)
@property
def input_ports(self):
- return _ComponentPorts(False, self,
- native_bt.component_filter_get_input_port_by_name,
- native_bt.component_filter_get_input_port_by_index,
- native_bt.component_filter_get_input_port_count)
+ return _ComponentPorts(self._ptr,
+ native_bt.component_filter_borrow_input_port_by_name_const,
+ native_bt.component_filter_borrow_input_port_by_index_const,
+ native_bt.component_filter_get_input_port_count,
+ bt2.port._InputPort)
# This is analogous to _GenericSinkComponentClass, but for sink
# component objects.
-class _GenericSinkComponent(object._Object, _SinkComponent):
+class _GenericSinkComponent(object._SharedObject, _SinkComponent):
+ _get_ref = staticmethod(native_bt.component_sink_get_ref)
+ _put_ref = staticmethod(native_bt.component_sink_put_ref)
+
@property
def input_ports(self):
- return _ComponentPorts(False, self,
- native_bt.component_sink_get_input_port_by_name,
- native_bt.component_sink_get_input_port_by_index,
- native_bt.component_sink_get_input_port_count)
+ return _ComponentPorts(self._ptr,
+ native_bt.component_sink_borrow_input_port_by_name_const,
+ native_bt.component_sink_borrow_input_port_by_index_const,
+ native_bt.component_sink_get_input_port_count,
+ bt2.port._InputPort)
_COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS = {
}
-def _create_generic_component_from_ptr(ptr):
- comp_cls_type = native_bt.component_get_class_type(ptr)
+# Create a component Python object of type _GenericSourceComponent,
+# _GenericFilterComponent or _GenericSinkComponent, depending on
+# comp_cls_type.
+#
+# Steals the reference to ptr from the caller.
+
+def _create_component_from_ptr(ptr, comp_cls_type):
return _COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS[comp_cls_type]._create_from_ptr(ptr)
-def _create_generic_component_class_from_ptr(ptr):
- comp_cls_type = native_bt.component_class_get_type(ptr)
- return _COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[comp_cls_type]._create_from_ptr(ptr)
+# Same as the above, but acquire a new reference instead of stealing the
+# reference from the caller.
+
+def _create_component_from_ptr_and_get_ref(ptr, comp_cls_type):
+ return _COMP_CLS_TYPE_TO_GENERIC_COMP_PYCLS[comp_cls_type]._create_from_ptr_and_get_ref(ptr)
+
+
+# Create a component class Python object of type
+# _GenericSourceComponentClass, _GenericFilterComponentClass or
+# _GenericSinkComponentClass, depending on comp_cls_type.
+#
+# Acquires a new reference to ptr.
+
+def _create_component_class_from_ptr_and_get_ref(ptr, comp_cls_type):
+ return _COMP_CLS_TYPE_TO_GENERIC_COMP_CLS_PYCLS[comp_cls_type]._create_from_ptr_and_get_ref(ptr)
def _trim_docstring(docstring):
# finalized.
#
# User-defined source and filter component classes must use the
-# `notification_iterator_class` class parameter to specify the
-# notification iterator class to use for this component class:
+# `message_iterator_class` class parameter to specify the
+# message iterator class to use for this component class:
#
-# class MyNotificationIterator(bt2._UserNotificationIterator):
+# class MyMessageIterator(bt2._UserMessageIterator):
# ...
#
# class MySource(bt2._UserSourceComponent,
-# notification_iterator_class=MyNotificationIterator):
+# message_iterator_class=MyMessageIterator):
# ...
#
-# This notification iterator class must inherit
-# bt2._UserNotificationIterator, and it must define the _get() and
-# _next() methods. The notification iterator class can also define an
+# This message iterator class must inherit
+# bt2._UserMessageIterator, and it must define the _get() and
+# _next() methods. The message iterator class can also define an
# __init__() method: this method has access to the original Python
# component object which was used to create it as the `component`
-# property. The notification iterator class can also define a
+# property. The message iterator class can also define a
# _finalize() method (again, do NOT use __del__()): this is called when
-# the notification iterator is (really) destroyed.
+# the message iterator is (really) destroyed.
#
# When the user-defined class is destroyed, this metaclass's __del__()
# method is called: the native BT component class pointer is put (not
if len(lines) >= 3:
comp_cls_help = '\n'.join(lines[2:])
- iter_cls = kwargs.get('notification_iterator_class')
+ iter_cls = kwargs.get('message_iterator_class')
if _UserSourceComponent in bases:
_UserComponentType._set_iterator_class(cls, iter_cls)
# create instance, not user-initialized yet
self = cls.__new__(cls)
- # pointer to native private component object (weak/borrowed)
+ # pointer to native self component object (weak/borrowed)
self._ptr = comp_ptr
# call user's __init__() method
if params_ptr is not None:
- native_bt.get(params_ptr)
- params = bt2.values._create_from_ptr(params_ptr)
+ params = bt2.value._create_from_ptr_and_get_ref(params_ptr)
else:
params = None
@staticmethod
def _set_iterator_class(cls, iter_cls):
if iter_cls is None:
- raise bt2.IncompleteUserClass("cannot create component class '{}': missing notification iterator class".format(cls.__name__))
+ raise bt2.IncompleteUserClass("cannot create component class '{}': missing message iterator class".format(cls.__name__))
- if not issubclass(iter_cls, bt2.notification_iterator._UserNotificationIterator):
- raise bt2.IncompleteUserClass("cannot create component class '{}': notification iterator class does not inherit bt2._UserNotificationIterator".format(cls.__name__))
+ if not issubclass(iter_cls, bt2.message_iterator._UserMessageIterator):
+ raise bt2.IncompleteUserClass("cannot create component class '{}': message iterator class does not inherit bt2._UserMessageIterator".format(cls.__name__))
if not hasattr(iter_cls, '__next__'):
- raise bt2.IncompleteUserClass("cannot create component class '{}': notification iterator class is missing a __next__() method".format(cls.__name__))
+ raise bt2.IncompleteUserClass("cannot create component class '{}': message iterator class is missing a __next__() method".format(cls.__name__))
cls._iter_cls = iter_cls
@property
def name(cls):
- return native_bt.component_class_get_name(cls._cc_ptr)
+ ptr = cls._as_component_class_ptr(cls._cc_ptr)
+ return native_bt.component_class_get_name(ptr)
@property
def description(cls):
- return native_bt.component_class_get_description(cls._cc_ptr)
+ ptr = cls._as_component_class_ptr(cls._cc_ptr)
+ return native_bt.component_class_get_description(ptr)
@property
def help(cls):
- return native_bt.component_class_get_help(cls._cc_ptr)
+ ptr = cls._as_component_class_ptr(cls._cc_ptr)
+ return native_bt.component_class_get_help(ptr)
@property
def addr(cls):
# this can raise, in which case the native call to
# bt_component_class_query() returns NULL
if params_ptr is not None:
- native_bt.get(params_ptr)
- params = bt2.values._create_from_ptr(params_ptr)
+ params = bt2.value._create_from_ptr_and_get_ref(params_ptr)
else:
params = None
- native_bt.get(query_exec_ptr)
- query_exec = bt2.QueryExecutor._create_from_ptr(query_exec_ptr)
+ query_exec = bt2.QueryExecutor._create_from_ptr_and_get_ref(
+ query_exec_ptr)
# this can raise, but the native side checks the exception
results = cls._query(query_exec, obj, params)
- if results is NotImplemented:
- return results
-
# this can raise, but the native side checks the exception
results = bt2.create_value(results)
results_addr = int(native_bt.value_null)
else:
# return new reference
- results._get()
- results_addr = int(results._ptr)
+ results_addr = int(results._release())
return results_addr
def _query(cls, query_executor, obj, params):
- # BT catches this and returns NULL to the user
- return NotImplemented
-
- def __eq__(self, other):
- if not hasattr(other, 'addr'):
- return False
+ raise NotImplementedError
- return self.addr == other.addr
+ def _component_class_ptr(self):
+ return self._as_component_class_ptr(self._cc_ptr)
def __del__(cls):
if hasattr(cls, '_cc_ptr'):
- native_bt.put(cls._cc_ptr)
+ cc_ptr = cls._as_component_class_ptr(cls._cc_ptr)
+ native_bt.component_class_put_ref(cc_ptr)
+# Subclasses must provide these methods or property:
+#
+# - _as_not_self_specific_component_ptr: static method, must return the passed
+# specialized self component pointer (e.g. 'bt_self_component_sink *') as a
+# specialized non-self pointer (e.g. 'bt_component_sink *').
+# - _borrow_component_class_ptr: static method, must return a pointer to the
+# specialized component class (e.g. 'bt_component_class_sink *') of the
+# passed specialized component pointer (e.g. 'bt_component_sink *').
+# - _comp_cls_type: property, one of the native_bt.COMPONENT_CLASS_TYPE_*
+# constants.
class _UserComponent(metaclass=_UserComponentType):
@property
def name(self):
- pub_ptr = native_bt.component_from_private_component(self._ptr)
- name = native_bt.component_get_name(pub_ptr)
- native_bt.put(pub_ptr)
- assert(name is not None)
+ ptr = self._as_not_self_specific_component_ptr(self._ptr)
+ ptr = self._as_component_ptr(ptr)
+ name = native_bt.component_get_name(ptr)
+ assert name is not None
return name
@property
- def graph(self):
- pub_ptr = native_bt.component_from_private_component(self._ptr)
- ptr = native_bt.component_get_graph(pub_ptr)
- native_bt.put(pub_ptr)
- assert(ptr)
- return bt2.Graph._create_from_ptr(ptr)
-
- @property
- def component_class(self):
- pub_ptr = native_bt.component_from_private_component(self._ptr)
- cc_ptr = native_bt.component_get_class(pub_ptr)
- native_bt.put(pub_ptr)
- assert(cc_ptr)
- return _create_generic_component_class_from_ptr(cc_ptr)
+ def cls(self):
+ comp_ptr = self._as_not_self_specific_component_ptr(self._ptr)
+ cc_ptr = self._borrow_component_class_ptr(comp_ptr)
+ return _create_component_class_from_ptr_and_get_ref(cc_ptr, self._comp_cls_type)
@property
def addr(self):
def _accept_port_connection(self, port, other_port):
return True
- def _accept_port_connection_from_native(self, port_ptr, other_port_ptr):
- native_bt.get(port_ptr)
- native_bt.get(other_port_ptr)
- port = bt2.port._create_private_from_ptr(port_ptr)
- other_port = bt2.port._create_from_ptr(other_port_ptr)
+ def _accept_port_connection_from_native(self, self_port_ptr, self_port_type, other_port_ptr):
+ port = bt2.port._create_self_from_ptr_and_get_ref(
+ self_port_ptr, self_port_type)
+
+ if self_port_type == native_bt.PORT_TYPE_OUTPUT:
+ other_port_type = native_bt.PORT_TYPE_INPUT
+ else:
+ other_port_type = native_bt.PORT_TYPE_OUTPUT
+
+ other_port = bt2.port._create_from_ptr_and_get_ref(
+ other_port_ptr, other_port_type)
res = self._accept_port_connection(port, other_port_ptr)
if type(res) is not bool:
def _port_connected(self, port, other_port):
pass
- def _port_connected_from_native(self, port_ptr, other_port_ptr):
- native_bt.get(port_ptr)
- native_bt.get(other_port_ptr)
- port = bt2.port._create_private_from_ptr(port_ptr)
- other_port = bt2.port._create_from_ptr(other_port_ptr)
+ def _port_connected_from_native(self, self_port_ptr, self_port_type, other_port_ptr):
+ port = bt2.port._create_self_from_ptr_and_get_ref(
+ self_port_ptr, self_port_type)
+
+ if self_port_type == native_bt.PORT_TYPE_OUTPUT:
+ other_port_type = native_bt.PORT_TYPE_INPUT
+ else:
+ other_port_type = native_bt.PORT_TYPE_OUTPUT
- try:
- self._port_connected(port, other_port_ptr)
- except:
- if not _NO_PRINT_TRACEBACK:
- traceback.print_exc()
+ other_port = bt2.port._create_from_ptr_and_get_ref(
+ other_port_ptr, other_port_type)
+ self._port_connected(port, other_port)
- def _port_disconnected(self, port):
- pass
+ def _graph_is_configured_from_native(self):
+ self._graph_is_configured()
+
+ def _create_trace_class(self, env=None, uuid=None,
+ assigns_automatic_stream_class_id=True):
+ ptr = self._as_self_component_ptr(self._ptr)
+ tc_ptr = native_bt.trace_class_create(ptr)
+
+ if tc_ptr is None:
+ raise bt2.CreationError('could not create trace class')
+
+ tc = bt2._TraceClass._create_from_ptr(tc_ptr)
+
+ if env is not None:
+ for key, value in env.items():
+ tc.env[key] = value
+
+ if uuid is not None:
+ tc._uuid = uuid
+
+ tc._assigns_automatic_stream_class_id = assigns_automatic_stream_class_id
+
+ return tc
+
+ def _create_clock_class(self, frequency=None, name=None, description=None,
+ precision=None, offset=None, origin_is_unix_epoch=True,
+ uuid=None):
+ ptr = self._as_self_component_ptr(self._ptr)
+ cc_ptr = native_bt.clock_class_create(ptr)
- def _port_disconnected_from_native(self, port_ptr):
- native_bt.get(port_ptr)
- port = bt2.port._create_private_from_ptr(port_ptr)
+ if cc_ptr is None:
+ raise bt2.CreationError('could not create clock class')
+
+ cc = bt2.clock_class._ClockClass._create_from_ptr(cc_ptr)
+
+ if frequency is not None:
+ cc._frequency = frequency
+
+ if name is not None:
+ cc._name = name
+
+ if description is not None:
+ cc._description = description
+
+ if precision is not None:
+ cc._precision = precision
+
+ if offset is not None:
+ cc._offset = offset
+
+ cc._origin_is_unix_epoch = origin_is_unix_epoch
- try:
- self._port_disconnected(port)
- except:
- if not _NO_PRINT_TRACEBACK:
- traceback.print_exc()
+ if uuid is not None:
+ cc._uuid = uuid
+
+ return cc
class _UserSourceComponent(_UserComponent, _SourceComponent):
+ _as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_source_as_component_source)
+ _as_self_component_ptr = staticmethod(native_bt.self_component_source_as_self_component)
+
@property
def _output_ports(self):
- return _ComponentPorts(True, self,
- native_bt.private_component_source_get_output_private_port_by_name,
- native_bt.private_component_source_get_output_private_port_by_index,
- native_bt.component_source_get_output_port_count)
+ def get_output_port_count(self_ptr):
+ ptr = self._as_not_self_specific_component_ptr(self_ptr)
+ return native_bt.component_source_get_output_port_count(ptr)
- def _add_output_port(self, name):
+ return _ComponentPorts(self._ptr,
+ native_bt.self_component_source_borrow_output_port_by_name,
+ native_bt.self_component_source_borrow_output_port_by_index,
+ get_output_port_count,
+ bt2.port._UserComponentOutputPort)
+
+ def _add_output_port(self, name, user_data=None):
utils._check_str(name)
- fn = native_bt.private_component_source_add_output_private_port
- comp_status, priv_port_ptr = fn(self._ptr, name, None)
+ fn = native_bt.self_component_source_add_output_port
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add output port to source component object')
- assert(priv_port_ptr)
- return bt2.port._create_private_from_ptr(priv_port_ptr)
+ assert self_port_ptr is not None
+ return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
class _UserFilterComponent(_UserComponent, _FilterComponent):
+ _as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_filter_as_component_filter)
+ _as_self_component_ptr = staticmethod(native_bt.self_component_filter_as_self_component)
+
@property
def _output_ports(self):
- return _ComponentPorts(True, self,
- native_bt.private_component_filter_get_output_private_port_by_name,
- native_bt.private_component_filter_get_output_private_port_by_index,
- native_bt.component_filter_get_output_port_count)
+ def get_output_port_count(self_ptr):
+ ptr = self._as_not_self_specific_component_ptr(self_ptr)
+ return native_bt.component_filter_get_output_port_count(ptr)
+
+ return _ComponentPorts(self._ptr,
+ native_bt.self_component_filter_borrow_output_port_by_name,
+ native_bt.self_component_filter_borrow_output_port_by_index,
+ get_output_port_count,
+ bt2.port._UserComponentOutputPort)
@property
def _input_ports(self):
- return _ComponentPorts(True, self,
- native_bt.private_component_filter_get_input_private_port_by_name,
- native_bt.private_component_filter_get_input_private_port_by_index,
- native_bt.component_filter_get_input_port_count)
+ def get_input_port_count(self_ptr):
+ ptr = self._as_not_self_specific_component_ptr(self_ptr)
+ return native_bt.component_filter_get_input_port_count(ptr)
+
+ return _ComponentPorts(self._ptr,
+ native_bt.self_component_filter_borrow_input_port_by_name,
+ native_bt.self_component_filter_borrow_input_port_by_index,
+ get_input_port_count,
+ bt2.port._UserComponentInputPort)
- def _add_output_port(self, name):
+ def _add_output_port(self, name, user_data=None):
utils._check_str(name)
- fn = native_bt.private_component_filter_add_output_private_port
- comp_status, priv_port_ptr = fn(self._ptr, name, None)
+ fn = native_bt.self_component_filter_add_output_port
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add output port to filter component object')
- assert(priv_port_ptr)
- return bt2.port._create_private_from_ptr(priv_port_ptr)
+ assert self_port_ptr
+ return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
- def _add_input_port(self, name):
+ def _add_input_port(self, name, user_data=None):
utils._check_str(name)
- fn = native_bt.private_component_filter_add_input_private_port
- comp_status, priv_port_ptr = fn(self._ptr, name, None)
+ fn = native_bt.self_component_filter_add_input_port
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add input port to filter component object')
- assert(priv_port_ptr)
- return bt2.port._create_private_from_ptr(priv_port_ptr)
+ assert self_port_ptr
+ return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)
class _UserSinkComponent(_UserComponent, _SinkComponent):
+ _as_not_self_specific_component_ptr = staticmethod(native_bt.self_component_sink_as_component_sink)
+ _as_self_component_ptr = staticmethod(native_bt.self_component_sink_as_self_component)
+
@property
def _input_ports(self):
- return _ComponentPorts(True, self,
- native_bt.private_component_sink_get_input_private_port_by_name,
- native_bt.private_component_sink_get_input_private_port_by_index,
- native_bt.component_sink_get_input_port_count)
+ def get_input_port_count(self_ptr):
+ ptr = self._as_not_self_specific_component_ptr(self_ptr)
+ return native_bt.component_sink_get_input_port_count(ptr)
+
+ return _ComponentPorts(self._ptr,
+ native_bt.self_component_sink_borrow_input_port_by_name,
+ native_bt.self_component_sink_borrow_input_port_by_index,
+ get_input_port_count,
+ bt2.port._UserComponentInputPort)
- def _add_input_port(self, name):
+ def _add_input_port(self, name, user_data=None):
utils._check_str(name)
- fn = native_bt.private_component_sink_add_input_private_port
- comp_status, priv_port_ptr = fn(self._ptr, name, None)
+ fn = native_bt.self_component_sink_add_input_port
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add input port to sink component object')
- assert(priv_port_ptr)
- return bt2.port._create_private_from_ptr(priv_port_ptr)
+ assert self_port_ptr
+ return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)
projects / babeltrace.git / blobdiff