return bt2.message._create_from_ptr(msg_ptr)
+# This is created when a component wants to iterate on one of its input ports.
class _UserComponentInputPortMessageIterator(_GenericMessageIterator):
_get_msg_range = staticmethod(native_bt.py3_self_component_port_input_get_msg_range)
-
- @property
- def component(self):
- comp_ptr = native_bt.private_connection_message_iterator_get_component(self._ptr)
- assert(comp_ptr)
- return bt2.component._create_generic_component_from_ptr(comp_ptr)
+ _get_ref = staticmethod(native_bt.self_component_port_input_message_iterator_get_ref)
+ _put_ref = staticmethod(native_bt.self_component_port_input_message_iterator_put_ref)
+# This is created when the user wants to iterate on a component's output port,
+# from outside the graph.
class _OutputPortMessageIterator(_GenericMessageIterator):
_get_msg_range = staticmethod(native_bt.py3_port_output_get_msg_range)
_get_ref = staticmethod(native_bt.port_output_message_iterator_get_ref)
_put_ref = staticmethod(native_bt.port_output_message_iterator_put_ref)
+# This is extended by the user to implement component classes in Python. It
+# is created for a given output port when an input port message iterator is
+# created on the input port on the other side of the connection. It is also
+# created when an output port message iterator is created on this output port.
+#
+# Its purpose is to feed the messages that should go out through this output
+# port.
class _UserMessageIterator(_MessageIterator):
def __new__(cls, ptr):
# User iterator objects are always created by the native side,
self._ptr = ptr
return self
- def __init__(self):
+ def _init_from_native(self, self_output_port_ptr):
+ self_output_port = bt2.port._create_self_from_ptr_and_get_ref(
+ self_output_port_ptr, native_bt.PORT_TYPE_OUTPUT)
+ self.__init__(self_output_port)
+
+ def __init__(self, output_port):
pass
@property
import bt2
-@unittest.skip("this is broken")
class UserMessageIteratorTestCase(unittest.TestCase):
@staticmethod
def _create_graph(src_comp_cls):
def _consume(self):
next(self._msg_iter)
- def _port_connected(self, port, other_port):
- self._msg_iter = port.connection.create_message_iterator()
+ def _graph_is_configured(self):
+ self._msg_iter = self._input_ports['in'].create_message_iterator()
graph = bt2.Graph()
src_comp = graph.add_component(src_comp_cls, 'src')
return graph
def test_init(self):
+ the_output_port_from_source = None
+ the_output_port_from_iter = None
+
class MyIter(bt2._UserMessageIterator):
- def __init__(self):
+ def __init__(self, self_port_output):
nonlocal initialized
+ nonlocal the_output_port_from_iter
initialized = True
+ the_output_port_from_iter = self_port_output
class MySource(bt2._UserSourceComponent,
message_iterator_class=MyIter):
def __init__(self, params):
- self._add_output_port('out')
+ nonlocal the_output_port_from_source
+ the_output_port_from_source = self._add_output_port('out')
initialized = False
graph = self._create_graph(MySource)
+ graph.run()
self.assertTrue(initialized)
+ self.assertEqual(the_output_port_from_source.addr, the_output_port_from_iter.addr)
def test_finalize(self):
class MyIter(bt2._UserMessageIterator):
finalized = False
graph = self._create_graph(MySource)
+ graph.run()
del graph
self.assertTrue(finalized)
def test_component(self):
class MyIter(bt2._UserMessageIterator):
- def __init__(self):
+ def __init__(self, self_port_output):
nonlocal salut
salut = self._component._salut
salut = None
graph = self._create_graph(MySource)
+ graph.run()
self.assertEqual(salut, 23)
def test_addr(self):
class MyIter(bt2._UserMessageIterator):
- def __init__(self):
+ def __init__(self, self_port_output):
nonlocal addr
addr = self.addr
addr = None
graph = self._create_graph(MySource)
+ graph.run()
self.assertIsNotNone(addr)
self.assertNotEqual(addr, 0)
-@unittest.skip("this is broken")
-class PrivateConnectionMessageIteratorTestCase(unittest.TestCase):
- def test_component(self):
- class MyIter(bt2._UserMessageIterator):
- pass
-
- class MySource(bt2._UserSourceComponent,
- message_iterator_class=MyIter):
- def __init__(self, params):
- self._add_output_port('out')
-
- class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
- self._add_input_port('in')
-
- def _consume(self):
- next(self._msg_iter)
-
- def _port_connected(self, port, other_port):
- nonlocal upstream_comp
- self._msg_iter = port.connection.create_message_iterator()
- upstream_comp = self._msg_iter.component
-
- upstream_comp = None
- graph = bt2.Graph()
- src_comp = graph.add_component(MySource, 'src')
- sink_comp = graph.add_component(MySink, 'sink')
- graph.connect_ports(src_comp.output_ports['out'],
- sink_comp.input_ports['in'])
- self.assertEqual(src_comp, upstream_comp)
- del upstream_comp
-
-
-@unittest.skip("this is broken")
class OutputPortMessageIteratorTestCase(unittest.TestCase):
def test_component(self):
class MyIter(bt2._UserMessageIterator):
- def __init__(self):
- self._build_meta()
+ def __init__(self, self_port_output):
self._at = 0
- def _build_meta(self):
- self._trace = bt2.Trace()
- self._sc = bt2.StreamClass()
- self._ec = bt2.EventClass('salut')
- self._my_int_fc = bt2.IntegerFieldClass(32)
- self._ec.payload_field_class = bt2.StructureFieldClass()
- self._ec.payload_field_class += collections.OrderedDict([
- ('my_int', self._my_int_fc),
- ])
- self._sc.add_event_class(self._ec)
- self._trace.add_stream_class(self._sc)
- self._stream = self._sc()
- self._packet = self._stream.create_packet()
-
- def _create_event(self, value):
- ev = self._ec()
- ev.payload_field['my_int'] = value
- ev.packet = self._packet
- return ev
-
def __next__(self):
- if self._at == 5:
+ if self._at == 7:
raise bt2.Stop
- msg = bt2.EventMessage(self._create_event(self._at * 3))
+ if self._at == 0:
+ msg = self._create_stream_beginning_message(test_obj._stream)
+ elif self._at == 1:
+ msg = self._create_packet_beginning_message(test_obj._packet)
+ elif self._at == 5:
+ msg = self._create_packet_end_message(test_obj._packet)
+ elif self._at == 6:
+ msg = self._create_stream_end_message(test_obj._stream)
+ else:
+ msg = self._create_event_message(test_obj._event_class, test_obj._packet)
+ msg.event.payload_field['my_int'] = self._at * 3
+
self._at += 1
return msg
def __init__(self, params):
self._add_output_port('out')
+ trace_class = self._create_trace_class()
+ stream_class = trace_class.create_stream_class()
+
+ # Create payload field class
+ my_int_ft = trace_class.create_signed_integer_field_class(32)
+ payload_ft = trace_class.create_structure_field_class()
+ payload_ft += collections.OrderedDict([
+ ('my_int', my_int_ft),
+ ])
+
+ event_class = stream_class.create_event_class(name='salut', payload_field_class=payload_ft)
+
+ trace = trace_class()
+ stream = trace.create_stream(stream_class)
+ packet = stream.create_packet()
+
+ test_obj._event_class = event_class
+ test_obj._stream = stream
+ test_obj._packet = packet
+
+ test_obj = self
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
- types = [bt2.EventMessage]
- msg_iter = src.output_ports['out'].create_message_iterator(types)
+ msg_iter = graph.create_output_port_message_iterator(src.output_ports['out'])
for at, msg in enumerate(msg_iter):
- self.assertIsInstance(msg, bt2.EventMessage)
- self.assertEqual(msg.event.event_class.name, 'salut')
- field = msg.event.payload_field['my_int']
- self.assertEqual(field, at * 3)
+ if at == 0:
+ self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
+ elif at == 1:
+ self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
+ elif at == 5:
+ self.assertIsInstance(msg, bt2.message._PacketEndMessage)
+ elif at == 6:
+ self.assertIsInstance(msg, bt2.message._StreamEndMessage)
+ else:
+ self.assertIsInstance(msg, bt2.message._EventMessage)
+ self.assertEqual(msg.event.event_class.name, 'salut')
+ field = msg.event.payload_field['my_int']
+ self.assertEqual(field, at * 3)
projects / babeltrace.git / commitdiff
