)
self._user_port_connected(port, other_port)
- def _create_trace_class(self, assigns_automatic_stream_class_id=True):
+ def _create_trace_class(
+ self, user_attributes=None, assigns_automatic_stream_class_id=True
+ ):
ptr = self._bt_as_self_component_ptr(self._bt_ptr)
tc_ptr = native_bt.trace_class_create(ptr)
tc = bt2_trace_class._TraceClass._create_from_ptr(tc_ptr)
tc._assigns_automatic_stream_class_id = assigns_automatic_stream_class_id
+ if user_attributes is not None:
+ tc._user_attributes = user_attributes
+
return tc
def _create_clock_class(
self,
frequency=None,
name=None,
+ user_attributes=None,
description=None,
precision=None,
offset=None,
if name is not None:
cc._name = name
+ if user_attributes is not None:
+ cc._user_attributes = user_attributes
+
if description is not None:
cc._description = description
def _create_input_port_message_iterator(self, input_port):
utils._check_type(input_port, bt2_port._UserComponentInputPort)
- msg_iter_ptr = native_bt.self_component_port_input_message_iterator_create_from_sink_component(
+ status, msg_iter_ptr = native_bt.bt2_self_component_port_input_message_iterator_create_from_sink_component(
self._bt_ptr, input_port._ptr
)
-
- if msg_iter_ptr is None:
- raise bt2._MemoryError('cannot create message iterator object')
+ utils._handle_func_status(status, 'cannot create message iterator object')
+ assert msg_iter_ptr is not None
return bt2_message_iterator._UserComponentInputPortMessageIterator(msg_iter_ptr)