cls._bt_cc_ptr = cc_ptr
- def _bt_init_from_native(cls, comp_ptr, params_ptr):
+ def _bt_init_from_native(cls, comp_ptr, params_ptr, obj):
# create instance, not user-initialized yet
self = cls.__new__(cls)
else:
params = None
- self.__init__(params)
+ self.__init__(params, obj)
return self
def __call__(cls, *args, **kwargs):
def addr(self):
return int(self._bt_ptr)
- def __init__(self, params=None):
+ def __init__(self, params=None, obj=None):
pass
def _user_finalize(self):
component_class,
name,
params=None,
+ obj=None,
logging_level=bt2_logging.LoggingLevel.NONE,
):
if isinstance(component_class, bt2_component._SourceComponentClass):
cc_ptr = component_class._ptr
- add_fn = native_bt.graph_add_source_component
+ add_fn = native_bt.bt2_graph_add_source_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
elif isinstance(component_class, bt2_component._FilterComponentClass):
cc_ptr = component_class._ptr
- add_fn = native_bt.graph_add_filter_component
+ add_fn = native_bt.bt2_graph_add_filter_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
elif isinstance(component_class, bt2_component._SinkComponentClass):
cc_ptr = component_class._ptr
- add_fn = native_bt.graph_add_sink_component
+ add_fn = native_bt.bt2_graph_add_sink_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_SINK
elif issubclass(component_class, bt2_component._UserSourceComponent):
cc_ptr = component_class._bt_cc_ptr
- add_fn = native_bt.graph_add_source_component
+ add_fn = native_bt.bt2_graph_add_source_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
elif issubclass(component_class, bt2_component._UserSinkComponent):
cc_ptr = component_class._bt_cc_ptr
- add_fn = native_bt.graph_add_sink_component
+ add_fn = native_bt.bt2_graph_add_sink_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_SINK
elif issubclass(component_class, bt2_component._UserFilterComponent):
cc_ptr = component_class._bt_cc_ptr
- add_fn = native_bt.graph_add_filter_component
+ add_fn = native_bt.bt2_graph_add_filter_component
cc_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
else:
raise TypeError(
utils._check_str(name)
utils._check_log_level(logging_level)
- params = bt2.create_value(params)
+ base_cc_ptr = component_class._bt_component_class_ptr()
+
+ if obj is not None and not native_bt.bt2_is_python_component_class(base_cc_ptr):
+ raise ValueError('cannot pass a Python object to a non-Python component')
+ params = bt2.create_value(params)
params_ptr = params._ptr if params is not None else None
- status, comp_ptr = add_fn(self._ptr, cc_ptr, name, params_ptr, logging_level)
+ status, comp_ptr = add_fn(
+ self._ptr, cc_ptr, name, params_ptr, obj, logging_level
+ )
utils._handle_func_status(status, 'cannot add component to graph')
assert comp_ptr
return bt2_component._create_component_from_ptr(comp_ptr, cc_type)
PyObject *py_cls, const char *name, const char *description,
const char *help);
void bt_bt2_unregister_cc_ptr_to_py_cls(const bt_component_class *comp_cls);
+bool bt_bt2_is_python_component_class(const bt_component_class *comp_cls);
*/
#include "logging/comp-logging.h"
+#include "compat/glib.h"
/*
* This hash table associates a BT component class object address to a
return py_exc_to_status(NULL, NULL, self_message_iterator, NULL, -1);
}
+static
+bool bt_bt2_is_python_component_class(const bt_component_class *comp_cls)
+{
+ return bt_g_hash_table_contains(bt_cc_ptr_to_py_cls, comp_cls);
+}
+
/* Component class proxy methods (delegate to the attached Python object) */
static
bt_logging_level log_level = get_self_component_log_level(
self_component);
- (void) init_method_data;
-
BT_ASSERT(self_component);
BT_ASSERT(self_component_v);
BT_ASSERT(self_comp_cls_type_swig_type);
+ /*
+ * If there's any `init_method_data`, assume this component is
+ * getting initialized from Python, so that `init_method_data`
+ * is a Python object to pass to the user's __init__() method.
+ */
+ BT_ASSERT(!init_method_data ||
+ bt_bt2_is_python_component_class(component_class));
+
/*
* Get the user-defined Python class which created this
* component's class in the first place (borrowed
/*
* Do the equivalent of this:
*
- * py_comp = py_cls._bt_init_from_native(py_comp_ptr, py_params_ptr)
+ * py_comp = py_cls._bt_init_from_native(py_comp_ptr,
+ * py_params_ptr, init_method_data ? init_method_data : Py_None)
*
* _UserComponentType._bt_init_from_native() calls the Python
* component object's __init__() function.
+ *
+ * We don't take any reference on `init_method_data` which, if
+ * not `NULL`, is assumed to be a `PyObject *`: the user's
+ * __init__() function will eventually take a reference if
+ * needed. If `init_method_data` is `NULL`, then we pass
+ * `Py_None` as the initialization's Python object.
*/
py_comp = PyObject_CallMethod(py_cls,
- "_bt_init_from_native", "(OO)", py_comp_ptr, py_params_ptr);
+ "_bt_init_from_native", "(OOO)", py_comp_ptr, py_params_ptr,
+ init_method_data ? init_method_data : Py_None);
if (!py_comp) {
BT_COMP_LOG_CUR_LVL(BT_LOG_WARNING, log_level, self_component,
"Failed to call Python class's _bt_init_from_native() method: "
PyObject *bt_bt2_graph_add_port_added_listener(struct bt_graph *graph,
PyObject *py_callable);
+
PyObject *bt_bt2_graph_add_ports_connected_listener(struct bt_graph *graph,
PyObject *py_callable);
+
+bt_graph_add_component_status
+bt_bt2_graph_add_source_component(
+ bt_graph *graph,
+ const bt_component_class_source *component_class,
+ const char *name, const bt_value *params,
+ PyObject *obj, bt_logging_level log_level,
+ const bt_component_source **component);
+
+bt_graph_add_component_status
+bt_bt2_graph_add_filter_component(
+ bt_graph *graph,
+ const bt_component_class_filter *component_class,
+ const char *name, const bt_value *params,
+ PyObject *obj, bt_logging_level log_level,
+ const bt_component_filter **component);
+
+bt_graph_add_component_status
+bt_bt2_graph_add_sink_component(
+ bt_graph *graph,
+ const bt_component_class_sink *component_class,
+ const char *name, const bt_value *params,
+ PyObject *obj, bt_logging_level log_level,
+ const bt_component_sink **component);
Py_XDECREF(py_listener_id);
return py_listener_ids;
}
+
+static
+bt_graph_add_component_status
+bt_bt2_graph_add_source_component(
+ bt_graph *graph,
+ const bt_component_class_source *component_class,
+ const char *name, const bt_value *params,
+ PyObject *obj, bt_logging_level log_level,
+ const bt_component_source **component)
+{
+ return bt_graph_add_source_component_with_init_method_data(graph,
+ component_class, name, params, obj == Py_None ? NULL : obj,
+ log_level, component);
+}
+
+static
+bt_graph_add_component_status
+bt_bt2_graph_add_filter_component(
+ bt_graph *graph,
+ const bt_component_class_filter *component_class,
+ const char *name, const bt_value *params,
+ PyObject *obj, bt_logging_level log_level,
+ const bt_component_filter **component)
+{
+ return bt_graph_add_filter_component_with_init_method_data(graph,
+ component_class, name, params, obj == Py_None ? NULL : obj,
+ log_level, component);
+}
+
+static
+bt_graph_add_component_status
+bt_bt2_graph_add_sink_component(
+ bt_graph *graph,
+ const bt_component_class_sink *component_class,
+ const char *name, const bt_value *params,
+ PyObject *obj, bt_logging_level log_level,
+ const bt_component_sink **component)
+{
+ return bt_graph_add_sink_component_with_init_method_data(graph,
+ component_class, name, params, obj == Py_None ? NULL : obj,
+ log_level, component);
+}
plugin_name,
class_name,
params=None,
+ obj=None,
logging_level=bt2_logging.LoggingLevel.NONE,
):
utils._check_str(plugin_name)
self._plugin_name = plugin_name
self._class_name = class_name
self._logging_level = logging_level
+ self._obj = obj
if type(params) is str:
self._params = bt2.create_value({'inputs': [params]})
def params(self):
return self._params
+ @property
+ def obj(self):
+ return self._obj
+
# datetime.datetime or integral to nanoseconds
def _get_ns(obj):
comp_cls = comp_classes[comp_spec.class_name]
name = self._get_unique_comp_name(comp_spec)
comp = self._graph.add_component(
- comp_cls, name, comp_spec.params, comp_spec.logging_level
+ comp_cls, name, comp_spec.params, comp_spec.obj, comp_spec.logging_level
)
return comp
return notif
class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
self._graph = bt2.Graph()
def test_name(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
self.assertEqual(comp_self.name, 'yaes')
def _user_consume(self):
def test_logging_level(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
self.assertEqual(comp_self.logging_level, bt2.LoggingLevel.INFO)
def _user_consume(self):
def test_class(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
self.assertEqual(comp_self.cls, MySink)
def _user_consume(self):
def test_addr(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
self.assertIsInstance(comp_self.addr, int)
self.assertNotEqual(comp_self.addr, 0)
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
class SourceWithFailingIter(
bt2._UserSourceComponent, message_iterator_class=FailingIter
):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class SourceWithFailingInit(
bt2._UserSourceComponent, message_iterator_class=FailingIter
):
- def __init__(self, params):
+ def __init__(self, params, obj):
raise ValueError('Source is failing')
class WorkingSink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._in = self._add_input_port('in')
def _user_graph_is_configured(self):
class SinkWithExceptionChaining(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._in = self._add_input_port('in')
def _user_graph_is_configured(self):
return msg
class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
tc = self._create_trace_class()
comp_params = None
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
nonlocal comp_params
comp_params = params
self.assertEqual(params, comp_params)
del comp_params
+ def test_add_component_obj_python_comp_cls(self):
+ comp_obj = None
+
+ class MySink(bt2._UserSinkComponent):
+ def __init__(self, params, obj):
+ nonlocal comp_obj
+ comp_obj = obj
+
+ def _user_consume(self):
+ pass
+
+ obj = object()
+ comp = self._graph.add_component(MySink, 'salut', obj=obj)
+ self.assertIs(comp_obj, obj)
+ del comp_obj
+
+ def test_add_component_obj_none_python_comp_cls(self):
+ comp_obj = None
+
+ class MySink(bt2._UserSinkComponent):
+ def __init__(self, params, obj):
+ nonlocal comp_obj
+ comp_obj = obj
+
+ def _user_consume(self):
+ pass
+
+ comp = self._graph.add_component(MySink, 'salut')
+ self.assertIsNone(comp_obj)
+ del comp_obj
+
+ def test_add_component_obj_non_python_comp_cls(self):
+ comp_obj = None
+
+ plugin = bt2.find_plugin('text', find_in_user_dir=False, find_in_sys_dir=False)
+ assert plugin is not None
+ cc = plugin.source_component_classes['dmesg']
+ assert cc is not None
+
+ with self.assertRaises(ValueError):
+ comp = self._graph.add_component(cc, 'salut', obj=57)
+
def test_add_component_invalid_cls_type(self):
with self.assertRaises(TypeError):
self._graph.add_component(int, 'salut')
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
raise TypeError
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
return self._create_stream_beginning_message(self._stream)
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
return msg
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._input_port = self._add_input_port('in')
self._at = 0
return msg
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._input_port = self._add_input_port('in')
self._at = 0
return msg
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._input_port = self._add_input_port('in')
self._at = 0
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
self._add_output_port('zero')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
self._add_output_port('zero')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
def test_raise_in_component_init(self):
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
raise ValueError('oops!')
def _user_consume(self):
def test_raise_in_port_added_listener(self):
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
return msg
class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out', params)
with_cc = bool(params['with_cc'])
@staticmethod
def _create_graph(src_comp_cls, flt_comp_cls=None):
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
def _user_consume(self):
the_output_port_from_iter = self_port_output
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
nonlocal the_output_port_from_source
the_output_port_from_source = self._add_output_port('out', 'user data')
src_iter_initialized = True
class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
class MyFilterIter(bt2._UserMessageIterator):
return next(self._up_iter)
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_input_port('in')
self._add_output_port('out')
finalized = True
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
finalized = False
salut = self._component._salut
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
self._salut = 23
addr = self.addr
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
addr = None
return self._msgs.pop(0)
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
tc = self._create_trace_class()
sc = tc.create_stream_class(supports_packets=True)
ec = sc.create_event_class()
raise StopIteration
class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
tc = self._create_trace_class()
sc = tc.create_stream_class(supports_packets=True)
ec = sc.create_event_class()
return self._upstream_iter.can_seek_beginning
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
input_port = self._add_input_port('in')
self._add_output_port('out', input_port)
raise bt2.TryAgain
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
graph = bt2.Graph()
return msg
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(self, params):
+ def __init__(self, params, obj):
self._add_output_port('out')
trace_class = self._create_trace_class()
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port = comp_self._add_output_port('out')
self.assertEqual(port.name, 'out')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port = comp_self._add_input_port('in')
self.assertEqual(port.name, 'in')
def test_sink_add_input_port(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port = comp_self._add_input_port('in')
self.assertEqual(port.name, 'in')
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
port3 = comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
port3 = comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
port3 = comp_self._add_input_port('insert')
def test_user_sink_input_ports_getitem(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
port3 = comp_self._add_input_port('insert')
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp_self._add_input_port('insert')
def test_user_sink_input_ports_getitem_invalid_key(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp_self._add_input_port('insert')
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp_self._add_input_port('insert')
def test_user_sink_input_ports_len(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp_self._add_input_port('insert')
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
port3 = comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
port3 = comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
port3 = comp_self._add_input_port('insert')
def test_user_sink_input_ports_iter(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
port3 = comp_self._add_input_port('insert')
port3 = None
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
port3 = None
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
port3 = None
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
port3 = None
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp_self._add_input_port('insert')
def test_gen_sink_input_ports_getitem_invalid_key(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp_self._add_input_port('insert')
raise bt2.Stop
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_output_port('clear')
comp_self._add_output_port('print')
comp_self._add_output_port('insert')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp_self._add_input_port('insert')
def test_gen_sink_input_ports_len(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
comp_self._add_input_port('print')
comp_self._add_input_port('insert')
port3 = None
class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
port3 = None
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_output_port('clear')
port2 = comp_self._add_output_port('print')
port3 = None
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
port3 = None
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal port1, port2, port3
port1 = comp_self._add_input_port('clear')
port2 = comp_self._add_input_port('print')
def test_name(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
def _user_consume(self):
def test_connection_none(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
def _user_consume(self):
def test_is_connected_false(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
comp_self._add_input_port('clear')
def _user_consume(self):
def test_self_name(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port = comp_self._add_input_port('clear')
self.assertEqual(port.name, 'clear')
def test_self_connection_none(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port = comp_self._add_input_port('clear')
self.assertIsNone(port.connection)
def test_self_is_connected_false(self):
class MySink(bt2._UserSinkComponent):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
port = comp_self._add_input_port('clear')
self.assertFalse(port.is_connected)
raise bt2.Stop
class MySource(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal user_datas
p = comp_self._add_output_port('port1')
raise bt2.Stop
class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal user_datas
p = comp_self._add_output_port('port1')
raise bt2.Stop
class MySink(bt2._UserFilterComponent, message_iterator_class=MyIter):
- def __init__(comp_self, params):
+ def __init__(comp_self, params, obj):
nonlocal user_datas
p = comp_self._add_input_port('port1')
def run_in_component_init(func):
class MySink(bt2._UserSinkComponent):
- def __init__(self, params):
+ def __init__(self, params, obj):
nonlocal res_bound
res_bound = func(self)
files are not grouped.
"""
- def __init__(self, params):
+ def __init__(self, params, obj):
self._print_params(params)
@staticmethod
directory "some-dir" won't be found by TestSourceExt, because we won't
recurse in "some-dir"."""
- def __init__(self, params):
+ def __init__(self, params, obj):
self._print_params(params)
@staticmethod
class TestSourceABCDE(Base, bt2._UserSourceComponent, message_iterator_class=TestIter):
"""A source that recognizes the arbitrary string input "ABCDE"."""
- def __init__(self, params):
+ def __init__(self, params, obj):
self._print_params(params)
@staticmethod
class TheSourceOfAllEvil(
bt2._UserSourceComponent, message_iterator_class=TheIteratorOfAllEvil
):
- def __init__(self, params):
+ def __init__(self, params, obj):
tc = self._create_trace_class()
# Use a clock class with an offset, so we can test with --begin or --end
projects / babeltrace.git / commitdiff
