Optimize the number of messages sent between the remote inputs and the message handler

beat/core/message_handler.py

@@ -81,10 +81,12 @@ class ProxyMessageHandler(MessageHandler):
         return retval
 
 
-    def write(self, name, packed):
+    def write(self, name, end_data_index, packed):
         """Syntax: wrt output data"""
 
-        logger.debug('recv: wrt %s <bin> (size=%d)', name, len(packed))
+        end_data_index = int(end_data_index)
+
+        logger.debug('recv: wrt %s %d <bin> (size=%d)', name, end_data_index, len(packed))
 
         # Get output object
         output_candidate = self._get_output_candidate(name)
@@ -93,7 +95,7 @@ class ProxyMessageHandler(MessageHandler):
 
         data = output_candidate.data_sink.dataformat.type()
         data.unpack(packed)
-        output_candidate.write(data)
+        output_candidate.write(data, end_data_index=end_data_index)
 
         logger.debug('send: ack')
         self.socket.send('ack')

beat/core/test/test_message_handler.py

@@ -99,6 +99,90 @@ class TestMessageHandlerBase(unittest.TestCase):
 #----------------------------------------------------------
 
 
+class TestOneInput(TestMessageHandlerBase):
+
+    def setUp(self):
+        dataformat = DataFormat(prefix, 'user/single_integer/1')
+
+        self.create_remote_inputs(
+            DataFormat(prefix, 'user/single_integer/1'),
+            dict(
+                a = MockDataSource([
+                        dataformat.type(value=10),
+                        dataformat.type(value=20),
+                    ],
+                    [
+                        (0, 0),
+                        (1, 1),
+                    ]
+                )
+            )
+        )
+
+        self.remote_input_a = self.remote_input_list['a']
+
+
+    def test_input_has_more_data(self):
+        self.assertTrue(self.remote_input_a.hasMoreData())
+
+
+    def test_input_next(self):
+        self.remote_input_a.next()
+        self.assertEqual(self.remote_input_a.data.value, 10)
+
+
+    def test_input_full_cycle(self):
+        self.assertTrue(self.remote_input_a.hasMoreData())
+        self.remote_input_a.next()
+        self.assertEqual(self.remote_input_a.data.value, 10)
+
+        self.assertTrue(self.remote_input_a.hasDataChanged())
+        self.assertTrue(self.remote_input_a.hasMoreData())
+        self.assertTrue(self.remote_input_a.isDataUnitDone())
+        self.remote_input_a.next()
+        self.assertEqual(self.remote_input_a.data.value, 20)
+
+        self.assertTrue(self.remote_input_a.hasDataChanged())
+        self.assertTrue(not self.remote_input_a.hasMoreData())
+        self.assertTrue(self.remote_input_a.isDataUnitDone())
+
+
+    def test_group_has_more_data(self):
+        self.assertTrue(self.remote_group.hasMoreData())
+
+
+    def test_group_next(self):
+        self.remote_group.next()
+        self.assertEqual(self.remote_input_a.data.value, 10)
+
+        self.assertTrue(self.remote_input_a.hasDataChanged())
+        self.assertTrue(self.remote_input_a.hasMoreData())
+        self.assertTrue(self.remote_input_a.isDataUnitDone())
+
+
+    def test_group_full_cycle(self):
+        self.assertTrue(self.remote_group.hasMoreData())
+        self.remote_group.next()
+        self.assertEqual(self.remote_input_a.data.value, 10)
+
+        self.assertTrue(self.remote_input_a.hasDataChanged())
+        self.assertTrue(self.remote_input_a.hasMoreData())
+        self.assertTrue(self.remote_input_a.isDataUnitDone())
+
+        self.assertTrue(self.remote_group.hasMoreData())
+        self.remote_group.next()
+        self.assertEqual(self.remote_input_a.data.value, 20)
+
+        self.assertTrue(self.remote_input_a.hasDataChanged())
+        self.assertFalse(self.remote_input_a.hasMoreData())
+        self.assertTrue(self.remote_input_a.isDataUnitDone())
+
+        self.assertFalse(self.remote_group.hasMoreData())
+
+
+#----------------------------------------------------------
+
+
 class TestSameFrequencyInputs(TestMessageHandlerBase):
 
     def setUp(self):