+from bt2 import stream_class as bt2_stream_class
+from bt2 import event_class as bt2_event_class
+from bt2 import field_class as bt2_field_class
+from bt2 import value as bt2_value
+from utils import TestOutputPortMessageIterator
+
+
+def _create_const_event_class(tc, stream_class):
+ fc1 = tc.create_structure_field_class()
+ fc2 = tc.create_structure_field_class()
+ event_class = stream_class.create_event_class(
+ payload_field_class=fc1, specific_context_field_class=fc2
+ )
+
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self, config, self_port_output):
+ trace = tc()
+ stream = trace.create_stream(stream_class)
+ self._msgs = [
+ self._create_stream_beginning_message(stream),
+ self._create_event_message(event_class, stream),
+ ]
+
+ def __next__(self):
+ if len(self._msgs) == 0:
+ raise StopIteration
+
+ return self._msgs.pop(0)
+
+ class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, config, params, obj):
+ self._add_output_port("out", params)
+
+ graph = bt2.Graph()
+ src_comp = graph.add_component(MySrc, "my_source", None)
+ msg_iter = TestOutputPortMessageIterator(graph, src_comp.output_ports["out"])
+
+ # Ignore first message, stream beginning
+ _ = next(msg_iter)
+
+ event_msg = next(msg_iter)
+
+ return event_msg.event.cls