-static bt_self_component_status
-bt_py3_component_class_accept_port_connection(
- bt_self_component *self_component,
- bt_self_component_port *self_component_port,
- bt_port_type self_component_port_type,
- const bt_port *other_port)
-{
- enum bt_self_component_status status;
- PyObject *py_comp = NULL;
- PyObject *py_self_port_ptr = NULL;
- PyObject *py_other_port_ptr = NULL;
- PyObject *py_method_result = NULL;
-
- py_comp = bt_self_component_get_data(self_component);
- BT_ASSERT(py_comp);
-
- swig_type_info *self_component_port_swig_type = NULL;
- swig_type_info *other_port_swig_type = NULL;
- switch (self_component_port_type) {
- case BT_PORT_TYPE_INPUT:
- self_component_port_swig_type = SWIGTYPE_p_bt_self_component_port_input;
- other_port_swig_type = SWIGTYPE_p_bt_port_output;
- break;
- case BT_PORT_TYPE_OUTPUT:
- self_component_port_swig_type = SWIGTYPE_p_bt_self_component_port_output;
- other_port_swig_type = SWIGTYPE_p_bt_port_input;
- break;
- }
- BT_ASSERT(self_component_port_swig_type != NULL);
- BT_ASSERT(other_port_swig_type != NULL);
-
- py_self_port_ptr = SWIG_NewPointerObj(SWIG_as_voidptr(self_component_port),
- self_component_port_swig_type, 0);
- if (!py_self_port_ptr) {
- BT_LOGE_STR("Failed to create a SWIG pointer object.");
- goto error;
- }
-
- py_other_port_ptr = SWIG_NewPointerObj(SWIG_as_voidptr(other_port),
- other_port_swig_type, 0);
- if (!py_other_port_ptr) {
- BT_LOGE_STR("Failed to create a SWIG pointer object.");
- goto error;
- }
-
- py_method_result = PyObject_CallMethod(py_comp,
- "_accept_port_connection_from_native", "(OiO)", py_self_port_ptr,
- self_component_port_type, py_other_port_ptr);
-
- status = bt_py3_exc_to_self_component_status();
- if (!py_method_result && status == BT_SELF_COMPONENT_STATUS_OK) {
- /* Pretty sure this should never happen, but just in case */
- BT_LOGE("User's _accept_port_connection() method failed without raising an exception: "
- "status=%d", status);
- goto error;
- }
-
- if (status == BT_SELF_COMPONENT_STATUS_REFUSE_PORT_CONNECTION) {
- /*
- * Looks like the user method raised
- * PortConnectionRefused: accept this like if it
- * returned False.
- */
- goto end;
- } else if (status != BT_SELF_COMPONENT_STATUS_OK) {
- BT_LOGE("User's _accept_port_connection() raised an unexpected exception: "
- "status=%d", status);
- goto error;
- }
-
- BT_ASSERT(PyBool_Check(py_method_result));
-
- if (py_method_result == Py_True) {
- status = BT_SELF_COMPONENT_STATUS_OK;
- } else {
- status = BT_SELF_COMPONENT_STATUS_REFUSE_PORT_CONNECTION;
- }
-
- goto end;
-
-error:
- status = BT_SELF_COMPONENT_STATUS_ERROR;
-
- /*
- * Clear any exception: we're returning a bad status anyway. If
- * this call originated from Python, then the user gets an
- * appropriate error.
- */
- PyErr_Clear();
-
-end:
- Py_XDECREF(py_self_port_ptr);
- Py_XDECREF(py_other_port_ptr);
- Py_XDECREF(py_method_result);
- return status;
-}
-
-static bt_self_component_status
-bt_py3_component_class_source_accept_output_port_connection(bt_self_component_source *self_component_source,
- bt_self_component_port_output *self_component_port_output,
- const bt_port_input *other_port_input)
-{
- bt_self_component *self_component = bt_self_component_source_as_self_component(self_component_source);
- bt_self_component_port *self_component_port = bt_self_component_port_output_as_self_component_port(self_component_port_output);
- const bt_port *other_port = bt_port_input_as_port_const(other_port_input);
- return bt_py3_component_class_accept_port_connection(self_component, self_component_port, BT_PORT_TYPE_OUTPUT, other_port);
-}
-
-static bt_self_component_status
-bt_py3_component_class_filter_accept_input_port_connection(bt_self_component_filter *self_component_filter,
- bt_self_component_port_input *self_component_port_input,
- const bt_port_output *other_port_output)
-{
- bt_self_component *self_component = bt_self_component_filter_as_self_component(self_component_filter);
- bt_self_component_port *self_component_port = bt_self_component_port_input_as_self_component_port(self_component_port_input);
- const bt_port *other_port = bt_port_output_as_port_const(other_port_output);
- return bt_py3_component_class_accept_port_connection(self_component, self_component_port, BT_PORT_TYPE_INPUT, other_port);
-}
-
-static bt_self_component_status
-bt_py3_component_class_filter_accept_output_port_connection(bt_self_component_filter *self_component_filter,
- bt_self_component_port_output *self_component_port_output,
- const bt_port_input *other_port_input)
-{
- bt_self_component *self_component = bt_self_component_filter_as_self_component(self_component_filter);
- bt_self_component_port *self_component_port = bt_self_component_port_output_as_self_component_port(self_component_port_output);
- const bt_port *other_port = bt_port_input_as_port_const(other_port_input);
- return bt_py3_component_class_accept_port_connection(self_component, self_component_port, BT_PORT_TYPE_OUTPUT, other_port);
-}
-
-static bt_self_component_status
-bt_py3_component_class_sink_accept_input_port_connection(bt_self_component_sink *self_component_sink,
- bt_self_component_port_input *self_component_port_input,
- const bt_port_output *other_port_output)
-{
- bt_self_component *self_component = bt_self_component_sink_as_self_component(self_component_sink);
- bt_self_component_port *self_component_port = bt_self_component_port_input_as_self_component_port(self_component_port_input);
- const bt_port *other_port = bt_port_output_as_port_const(other_port_output);
- return bt_py3_component_class_accept_port_connection(self_component, self_component_port, BT_PORT_TYPE_INPUT, other_port);
-}
-