Add StreamLineReader with unit tests

setup.py

@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
 
 setup(
     name="galaxy.plugin.api",
-    version="0.37",
+    version="0.39",
     description="GOG Galaxy Integrations Python API",
     author='Galaxy team',
     author_email='galaxy@gog.com',

src/galaxy/api/jsonrpc.py

@@ -5,6 +5,8 @@ import logging
 import inspect
 import json
 
+from galaxy.reader import StreamLineReader
+
 class JsonRpcError(Exception):
     def __init__(self, code, message, data=None):
         self.code = code
@@ -67,13 +69,12 @@ def anonymise_sensitive_params(params, sensitive_params):
 class Server():
     def __init__(self, reader, writer, encoder=json.JSONEncoder()):
         self._active = True
-        self._reader = reader
+        self._reader = StreamLineReader(reader)
         self._writer = writer
         self._encoder = encoder
         self._methods = {}
         self._notifications = {}
         self._eof_listeners = []
-        self._input_buffer = bytes()
 
     def register_method(self, name, callback, internal, sensitive_params=False):
         """
@@ -105,7 +106,7 @@ class Server():
     async def run(self):
         while self._active:
             try:
-                data = await self._readline()
+                data = await self._reader.readline()
                 if not data:
                     self._eof()
                     continue
@@ -116,21 +117,6 @@ class Server():
             logging.debug("Received %d bytes of data", len(data))
             self._handle_input(data)
 
-    async def _readline(self):
-        """Like StreamReader.readline but without limit"""
-        while True:
-            chunk = await self._reader.read(1024)
-            if not chunk:
-                return chunk
-            previous_size = len(self._input_buffer)
-            self._input_buffer += chunk
-            it = self._input_buffer.find(b"\n", previous_size)
-            if it < 0:
-                continue
-            line = self._input_buffer[:it]
-            self._input_buffer = self._input_buffer[it+1:]
-            return line
-
     def stop(self):
         self._active = False
 

src/galaxy/reader.py

@@ -0,0 +1,28 @@
+from asyncio import StreamReader
+
+
+class StreamLineReader:
+    """Handles StreamReader readline without buffer limit"""
+    def __init__(self, reader: StreamReader):
+        self._reader = reader
+        self._buffer = bytes()
+        self._processed_buffer_it = 0
+
+    async def readline(self):
+        while True:
+            # check if there is no unprocessed data in the buffer
+            if not self._buffer or self._processed_buffer_it != 0:
+                chunk = await self._reader.read(1024)
+                if not chunk:
+                    return bytes() # EOF
+                self._buffer += chunk
+
+            it = self._buffer.find(b"\n", self._processed_buffer_it)
+            if it < 0:
+                self._processed_buffer_it = len(self._buffer)
+                continue
+
+            line = self._buffer[:it]
+            self._buffer = self._buffer[it+1:]
+            self._processed_buffer_it = 0
+            return line

tests/test_chunk_messages.py

@@ -12,6 +12,43 @@ def test_chunked_messages(plugin, read):
 
     message = json.dumps(request).encode() + b"\n"
     read.side_effect = [message[:5], message[5:], b""]
-    plugin.get_owned_games.return_value = None
     asyncio.run(plugin.run())
     plugin.install_game.assert_called_with(game_id="3")
+
+def test_joined_messages(plugin, read):
+    requests = [
+        {
+            "jsonrpc": "2.0",
+            "method": "install_game",
+            "params": {
+                "game_id": "3"
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "launch_game",
+            "params": {
+                "game_id": "3"
+            }
+        }
+    ]
+    data = b"".join([json.dumps(request).encode() + b"\n" for request in requests])
+
+    read.side_effect = [data, b""]
+    asyncio.run(plugin.run())
+    plugin.install_game.assert_called_with(game_id="3")
+    plugin.launch_game.assert_called_with(game_id="3")
+
+def test_not_finished(plugin, read):
+    request = {
+        "jsonrpc": "2.0",
+        "method": "install_game",
+        "params": {
+            "game_id": "3"
+        }
+    }
+
+    message = json.dumps(request).encode() # no new line
+    read.side_effect = [message, b""]
+    asyncio.run(plugin.run())
+    plugin.install_game.assert_not_called()

tests/test_stream_line_reader.py

@@ -0,0 +1,52 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from galaxy.reader import StreamLineReader
+from galaxy.unittest.mock import AsyncMock
+
+@pytest.fixture()
+def stream_reader():
+    reader = MagicMock()
+    reader.read = AsyncMock()
+    return reader
+
+@pytest.fixture()
+def read(stream_reader):
+    return stream_reader.read
+
+@pytest.fixture()
+def reader(stream_reader):
+    return StreamLineReader(stream_reader)
+
+@pytest.mark.asyncio
+async def test_message(reader, read):
+    read.return_value = b"a\n"
+    assert await reader.readline() == b"a"
+    read.assert_called_once()
+
+@pytest.mark.asyncio
+async def test_separate_messages(reader, read):
+    read.side_effect = [b"a\n", b"b\n"]
+    assert await reader.readline() == b"a"
+    assert await reader.readline() == b"b"
+    assert read.call_count == 2
+
+@pytest.mark.asyncio
+async def test_connected_messages(reader, read):
+    read.return_value = b"a\nb\n"
+    assert await reader.readline() == b"a"
+    assert await reader.readline() == b"b"
+    read.assert_called_once()
+
+@pytest.mark.asyncio
+async def test_cut_message(reader, read):
+    read.side_effect = [b"a", b"b\n"]
+    assert await reader.readline() == b"ab"
+    assert read.call_count == 2
+
+@pytest.mark.asyncio
+async def test_half_message(reader, read):
+    read.side_effect = [b"a", b""]
+    assert await reader.readline() == b""
+    assert read.call_count == 2