get_output_port_count,
bt2.port._UserComponentOutputPort)
- def _add_output_port(self, name):
+ def _add_output_port(self, name, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_source_add_output_port
- comp_status, self_port_ptr = fn(self._ptr, name, None)
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add output port to source component object')
assert self_port_ptr is not None
get_input_port_count,
bt2.port._UserComponentInputPort)
- def _add_output_port(self, name):
+ def _add_output_port(self, name, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_filter_add_output_port
- comp_status, self_port_ptr = fn(self._ptr, name, None)
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add output port to filter component object')
assert self_port_ptr
return bt2.port._UserComponentOutputPort._create_from_ptr(self_port_ptr)
- def _add_input_port(self, name):
+ def _add_input_port(self, name, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_filter_add_input_port
- comp_status, self_port_ptr = fn(self._ptr, name, None)
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add input port to filter component object')
assert self_port_ptr
get_input_port_count,
bt2.port._UserComponentInputPort)
- def _add_input_port(self, name):
+ def _add_input_port(self, name, user_data=None):
utils._check_str(name)
fn = native_bt.self_component_sink_add_input_port
- comp_status, self_port_ptr = fn(self._ptr, name, None)
+ comp_status, self_port_ptr = fn(self._ptr, name, user_data)
_handle_component_status(comp_status,
'cannot add input port to sink component object')
assert self_port_ptr
}
}
+/* Typemaps used for user data attached to self component ports. */
+
+/*
+ * The user data Python object is kept as the user data of the port, we pass
+ * the PyObject pointer directly to the port creation function.
+ */
+%typemap(in) void *PY_SELF_PORT_USER_DATA {
+ $1 = $input;
+}
+
+/*
+ * The port, if created successfully, now owns a reference to the Python object,
+ * we reflect that here.
+ */
+%typemap(argout) void *PY_SELF_PORT_USER_DATA {
+ if (PyLong_AsLong($result) == BT_SELF_COMPONENT_STATUS_OK) {
+ Py_INCREF($1);
+ }
+}
+
/* From component-const.h */
extern const char *bt_component_get_name(const bt_component *component);
extern bt_self_component_status
bt_self_component_source_add_output_port(
bt_self_component_source *self_component,
- const char *name, void *user_data,
+ const char *name, void *PY_SELF_PORT_USER_DATA,
bt_self_component_port_output **OUT);
/* From self-component-filter.h */
extern bt_self_component_status
bt_self_component_filter_add_output_port(
bt_self_component_filter *self_component,
- const char *name, void *data,
+ const char *name, void *PY_SELF_PORT_USER_DATA,
bt_self_component_port_output **OUT);
extern bt_self_component_port_input *
extern bt_self_component_status
bt_self_component_filter_add_input_port(
bt_self_component_filter *self_component,
- const char *name, void *data,
+ const char *name, void *PY_SELF_PORT_USER_DATA,
bt_self_component_port_input **OUT);
/* From self-component-sink.h */
extern bt_self_component_status
bt_self_component_sink_add_input_port(
bt_self_component_sink *self_component,
- const char *name, void *user_data,
+ const char *name, void *PY_SELF_PORT_USER_DATA,
bt_self_component_port_input **OUT);
* THE SOFTWARE.
*/
+/*
+ * Typemap for the user data attached to (and owned by) a self component port.
+ * The pointer saved as the port's user data is directly the PyObject *.
+ *
+ * As per the CPython calling convention, we need to return a new reference to
+ * the returned object, which will be transferred to the caller. The following
+ * typedef allows us to apply the typemap.
+ */
+%{
+typedef void *PY_SELF_PORT_USER_DATA;
+%}
+
+%typemap(out) PY_SELF_PORT_USER_DATA {
+ Py_INCREF($1);
+ $result = $1;
+}
+
/* From port-const.h */
typedef enum bt_port_type {
extern bt_self_component *bt_self_component_port_borrow_component(
bt_self_component_port *self_port);
-extern void *bt_self_component_port_get_data(
+extern PY_SELF_PORT_USER_DATA bt_self_component_port_get_data(
const bt_self_component_port *self_port);
/* From self-component-port-output.h */
return bt2.connection._Connection._create_from_ptr_and_get_ref(conn_ptr)
+ @property
+ def user_data(self):
+ ptr = self._as_self_port_ptr(self._ptr)
+ return native_bt.self_component_port_get_data(ptr)
+
class _UserComponentInputPort(_UserComponentPort, _InputPort):
_as_self_port_ptr = staticmethod(native_bt.self_component_port_input_as_self_component_port)
message_iterator_class=MyIter):
def __init__(self, params):
nonlocal the_output_port_from_source
- the_output_port_from_source = self._add_output_port('out')
+ the_output_port_from_source = self._add_output_port('out', 'user data')
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)
+ self.assertEqual(the_output_port_from_iter.user_data, 'user data')
def test_finalize(self):
class MyIter(bt2._UserMessageIterator):
pass
self._create_comp(MySink)
+
+ def test_source_self_port_user_data(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __next__(self):
+ raise bt2.Stop
+
+ class MySource(bt2._UserFilterComponent,
+ message_iterator_class=MyIter):
+ def __init__(comp_self, params):
+ nonlocal user_datas
+
+ p = comp_self._add_output_port('port1')
+ user_datas.append(p.user_data)
+ p = comp_self._add_output_port('port2', 2)
+ user_datas.append(p.user_data)
+
+ user_datas = []
+
+ comp = self._create_comp(MySource)
+ self.assertEqual(user_datas, [None, 2])
+
+ def test_filter_self_port_user_data(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __next__(self):
+ raise bt2.Stop
+
+ class MyFilter(bt2._UserFilterComponent,
+ message_iterator_class=MyIter):
+ def __init__(comp_self, params):
+ nonlocal user_datas
+
+ p = comp_self._add_output_port('port1')
+ user_datas.append(p.user_data)
+ p = comp_self._add_output_port('port2', 'user data string')
+ user_datas.append(p.user_data)
+
+ p = comp_self._add_input_port('port3')
+ user_datas.append(p.user_data)
+ p = comp_self._add_input_port('port4', user_data={'user data': 'dict'})
+ user_datas.append(p.user_data)
+
+ user_datas = []
+
+ comp = self._create_comp(MyFilter)
+ self.assertEqual(user_datas,
+ [None, 'user data string', None, {'user data': 'dict'}])
+
+ def test_sink_self_port_user_data(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __next__(self):
+ raise bt2.Stop
+
+ class MySink(bt2._UserFilterComponent,
+ message_iterator_class=MyIter):
+ def __init__(comp_self, params):
+ nonlocal user_datas
+
+ p = comp_self._add_input_port('port1')
+ user_datas.append(p.user_data)
+ p = comp_self._add_input_port('port2', set())
+ user_datas.append(p.user_data)
+
+ user_datas = []
+
+ comp = self._create_comp(MySink)
+ self.assertEqual(user_datas, [None, set()])