def _add_output_port(self, name, user_data=None):
utils._check_str(name)
+
+ if name in self._output_ports:
+ raise ValueError(
+ 'source component `{}` already contains an output port named `{}`'.format(
+ self.name, name
+ )
+ )
+
fn = native_bt.self_component_source_add_output_port
comp_status, self_port_ptr = fn(self._bt_ptr, name, user_data)
utils._handle_func_status(
def _add_output_port(self, name, user_data=None):
utils._check_str(name)
+
+ if name in self._output_ports:
+ raise ValueError(
+ 'filter component `{}` already contains an output port named `{}`'.format(
+ self.name, name
+ )
+ )
+
fn = native_bt.self_component_filter_add_output_port
comp_status, self_port_ptr = fn(self._bt_ptr, name, user_data)
utils._handle_func_status(
def _add_input_port(self, name, user_data=None):
utils._check_str(name)
+
+ if name in self._input_ports:
+ raise ValueError(
+ 'filter component `{}` already contains an input port named `{}`'.format(
+ self.name, name
+ )
+ )
+
fn = native_bt.self_component_filter_add_input_port
comp_status, self_port_ptr = fn(self._bt_ptr, name, user_data)
utils._handle_func_status(
def _add_input_port(self, name, user_data=None):
utils._check_str(name)
+
+ if name in self._input_ports:
+ raise ValueError(
+ 'sink component `{}` already contains an input port named `{}`'.format(
+ self.name, name
+ )
+ )
+
fn = native_bt.self_component_sink_add_input_port
comp_status, self_port_ptr = fn(self._bt_ptr, name, user_data)
utils._handle_func_status(
def _create_message_iterator(self, input_port):
utils._check_type(input_port, bt2_port._UserComponentInputPort)
+ if not input_port.is_connected:
+ raise ValueError('input port is not connected')
+
(
status,
msg_iter_ptr,