def _build_meta(self):
self._tc = self._component._create_trace_class()
self._t = self._tc()
- self._sc = self._tc.create_stream_class()
+ self._sc = self._tc.create_stream_class(supports_packets=True)
self._ec = self._sc.create_event_class(name='salut')
self._my_int_ft = self._tc.create_signed_integer_field_class(32)
payload_ft = self._tc.create_structure_field_class()
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
comp = self._graph.add_component(MySink, 'salut')
self.assertEqual(comp.name, 'salut')
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
comp = self._graph.add_component(MySink, 'salut')
assert comp
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
params = {'hello': 23, 'path': '/path/to/stuff'}
comp = self._graph.add_component(MySink, 'salut', params)
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
with self.assertRaises(TypeError):
self._graph.add_component(MySink, 'salut', logging_level='yo')
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
with self.assertRaises(ValueError):
self._graph.add_component(MySink, 'salut', logging_level=12345)
class MySink(bt2._UserSinkComponent):
def _consume(self):
pass
+ def _graph_is_configured(self):
+ pass
comp = self._graph.add_component(MySink, 'salut',
logging_level=bt2.LoggingLevel.DEBUG)
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
def _port_connected(self, port, other_port):
self._add_input_port('taste')
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
def _port_connected(self, port, other_port):
self._add_input_port('taste')
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
graph = bt2.Graph()
with self.assertRaises(bt2.Error):
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
def port_added_listener(component, port):
raise ValueError('oh noes!')
def _consume(self):
raise bt2.Stop
+ def _graph_is_configured(self):
+ pass
+
def ports_connected_listener(upstream_component, upstream_port,
downstream_component, downstream_port):
raise ValueError('oh noes!')