- def _create_event_message(self, event_class, packet, default_clock_snapshot=None):
- utils._check_type(event_class, bt2.event_class._EventClass)
- utils._check_type(packet, bt2.packet._Packet)
+ 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_message_iterator(
+ self._bt_ptr, input_port._ptr
+ )
+
+ if msg_iter_ptr is None:
+ raise bt2._MemoryError('cannot create message iterator object')
+
+ return _UserComponentInputPortMessageIterator(msg_iter_ptr)
+
+ def _create_event_message(
+ self, event_class, parent=None, default_clock_snapshot=None
+ ):
+ utils._check_type(event_class, bt2_event_class._EventClass)
+
+ if event_class.stream_class.supports_packets:
+ utils._check_type(parent, bt2_packet._Packet)
+ else:
+ utils._check_type(parent, bt2_stream._Stream)