def _component(self):
return native_bt.bt2_get_user_component_from_user_msg_iter(self._bt_ptr)
+ @property
+ def _port(self):
+ port_ptr = native_bt.self_message_iterator_borrow_port(self._bt_ptr)
+ assert port_ptr is not None
+ return bt2_port._create_self_from_ptr_and_get_ref(
+ port_ptr, native_bt.PORT_TYPE_OUTPUT
+ )
+
@property
def addr(self):
return int(self._bt_ptr)
import bt2
import sys
from utils import TestOutputPortMessageIterator
+from bt2 import port as bt2_port
class UserMessageIteratorTestCase(unittest.TestCase):
graph.run()
self.assertEqual(salut, 23)
+ def test_port(self):
+ class MyIter(bt2._UserMessageIterator):
+ def __init__(self_iter, self_port_output):
+ nonlocal called
+ called = True
+ port = self_iter._port
+ self.assertIs(type(self_port_output), bt2_port._UserComponentOutputPort)
+ self.assertIs(type(port), bt2_port._UserComponentOutputPort)
+ self.assertEqual(self_port_output.addr, port.addr)
+
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
+ def __init__(self, params, obj):
+ self._add_output_port('out')
+
+ called = False
+ graph = self._create_graph(MySource)
+ graph.run()
+ self.assertTrue(called)
+
def test_addr(self):
class MyIter(bt2._UserMessageIterator):
def __init__(self, self_port_output):