Wait for drain before writing

.gitignore

@@ -7,3 +7,4 @@ docs/build/
 Pipfile
 .idea
 docs/source/_build
+.mypy_cache

setup.py

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

src/galaxy/api/consts.py

@@ -113,6 +113,7 @@ class Feature(Enum):
     LaunchPlatformClient = "LaunchPlatformClient"
     ImportGameLibrarySettings = "ImportGameLibrarySettings"
     ImportOSCompatibility = "ImportOSCompatibility"
+    ImportUserPresence = "ImportUserPresence"
 
 
 class LicenseType(Enum):
@@ -140,3 +141,11 @@ class OSCompatibility(Flag):
     Windows = 0b001
     MacOS   = 0b010
     Linux   = 0b100
+
+
+class PresenceState(Enum):
+    """"Possible states of a user."""
+    Unknown = "unknown"
+    Online = "online"
+    Offline = "offline"
+    Away = "away"

src/galaxy/api/jsonrpc.py

@@ -88,6 +88,7 @@ class Server():
         self._methods = {}
         self._notifications = {}
         self._task_manager = TaskManager("jsonrpc server")
+        self._write_lock = asyncio.Lock()
 
     def register_method(self, name, callback, immediate, sensitive_params=False):
         """
@@ -223,12 +224,16 @@ class Server():
             raise InvalidRequest()
 
     def _send(self, data):
+        async def send_task(data_):
+            async with self._write_lock:
+                self._writer.write(data_)
+                await self._writer.drain()
+
         try:
             line = self._encoder.encode(data)
             logging.debug("Sending data: %s", line)
             data = (line + "\n").encode("utf-8")
-            self._writer.write(data)
-            self._task_manager.create_task(self._writer.drain(), "drain")
+            self._task_manager.create_task(send_task(data), "send")
         except TypeError as error:
             logging.error(str(error))
 
@@ -263,6 +268,7 @@ class NotificationClient():
         self._encoder = encoder
         self._methods = {}
         self._task_manager = TaskManager("notification client")
+        self._write_lock = asyncio.Lock()
 
     def notify(self, method, params, sensitive_params=False):
         """
@@ -286,12 +292,16 @@ class NotificationClient():
         await self._task_manager.wait()
 
     def _send(self, data):
+        async def send_task(data_):
+            async with self._write_lock:
+                self._writer.write(data_)
+                await self._writer.drain()
+
         try:
             line = self._encoder.encode(data)
             data = (line + "\n").encode("utf-8")
             logging.debug("Sending %d byte of data", len(data))
-            self._writer.write(data)
-            self._task_manager.create_task(self._writer.drain(), "drain")
+            self._task_manager.create_task(send_task(data), "send")
         except TypeError as error:
             logging.error("Failed to parse outgoing message: %s", str(error))
 

src/galaxy/api/plugin.py

@@ -10,7 +10,9 @@ from typing import Any, Dict, List, Optional, Set, Union
 from galaxy.api.consts import Feature, OSCompatibility
 from galaxy.api.errors import ImportInProgress, UnknownError
 from galaxy.api.jsonrpc import ApplicationError, NotificationClient, Server
-from galaxy.api.types import Achievement, Authentication, FriendInfo, Game, GameTime, LocalGame, NextStep, GameLibrarySettings
+from galaxy.api.types import (
+    Achievement, Authentication, FriendInfo, Game, GameLibrarySettings, GameTime, LocalGame, NextStep, UserPresence
+)
 from galaxy.task_manager import TaskManager
 
 
@@ -49,6 +51,7 @@ class Plugin:
         self._game_times_import_in_progress = False
         self._game_library_settings_import_in_progress = False
         self._os_compatibility_import_in_progress = False
+        self._user_presence_import_in_progress = False
 
         self._persistent_cache = dict()
 
@@ -118,6 +121,9 @@ class Plugin:
         self._register_method("start_os_compatibility_import", self._start_os_compatibility_import)
         self._detect_feature(Feature.ImportOSCompatibility, ["get_os_compatibility"])
 
+        self._register_method("start_user_presence_import", self._start_user_presence_import)
+        self._detect_feature(Feature.ImportUserPresence, ["get_user_presence"])
+
     async def __aenter__(self):
         return self
 
@@ -265,7 +271,7 @@ class Plugin:
 
         """
         # temporary solution for persistent_cache vs credentials issue
-        self.persistent_cache['credentials'] = credentials  # type: ignore
+        self.persistent_cache["credentials"] = credentials  # type: ignore
 
         self._notification_client.notify("store_credentials", credentials, sensitive_params=True)
 
@@ -450,6 +456,27 @@ class Plugin:
     def _os_compatibility_import_finished(self) -> None:
         self._notification_client.notify("os_compatibility_import_finished", None)
 
+    def _user_presence_import_success(self, user_id: str, user_presence: UserPresence) -> None:
+        self._notification_client.notify(
+            "user_presence_import_success",
+            {
+                "user_id": user_id,
+                "presence": user_presence
+            }
+        )
+
+    def _user_presence_import_failure(self, user_id: str, error: ApplicationError) -> None:
+        self._notification_client.notify(
+            "user_presence_import_failure",
+            {
+                "user_id": user_id,
+                "error": error.json()
+            }
+        )
+
+    def _user_presence_import_finished(self) -> None:
+        self._notification_client.notify("user_presence_import_finished", None)
+
     def lost_authentication(self) -> None:
         """Notify the client that integration has lost authentication for the
          current user and is unable to perform actions which would require it.
@@ -912,6 +939,62 @@ class Plugin:
     def os_compatibility_import_complete(self) -> None:
         """Override this method to handle operations after OS compatibility import is finished (like updating cache)."""
 
+    async def _start_user_presence_import(self, user_ids: List[str]) -> None:
+        if self._user_presence_import_in_progress:
+            raise ImportInProgress()
+
+        context = await self.prepare_user_presence_context(user_ids)
+
+        async def import_user_presence(user_id, context_) -> None:
+            try:
+                self._user_presence_import_success(user_id, await self.get_user_presence(user_id, context_))
+            except ApplicationError as error:
+                self._user_presence_import_failure(user_id, error)
+            except Exception:
+                logging.exception("Unexpected exception raised in import_user_presence")
+                self._user_presence_import_failure(user_id, UnknownError())
+
+        async def import_user_presence_set(user_ids_, context_) -> None:
+            try:
+                await asyncio.gather(*[
+                    import_user_presence(user_id, context_)
+                    for user_id in user_ids_
+                ])
+            finally:
+                self._user_presence_import_finished()
+                self._user_presence_import_in_progress = False
+                self.user_presence_import_complete()
+
+        self._external_task_manager.create_task(
+            import_user_presence_set(user_ids, context),
+            "user presence import",
+            handle_exceptions=False
+        )
+        self._user_presence_import_in_progress = True
+
+    async def prepare_user_presence_context(self, user_ids: List[str]) -> Any:
+        """Override this method to prepare context for get_user_presence.
+        This allows for optimizations like batch requests to platform API.
+        Default implementation returns None.
+
+        :param user_ids: the ids of the users for whom presence information is imported
+        :return: context
+        """
+        return None
+
+    async def get_user_presence(self, user_id: str, context: Any) -> UserPresence:
+        """Override this method to return presence information for the user with the provided user_id.
+        This method is called by import task initialized by GOG Galaxy Client.
+
+        :param user_id: the id of the user for whom presence information is imported
+        :param context: the value returned from :meth:`prepare_user_presence_context`
+        :return: UserPresence presence information of the provided user
+        """
+        raise NotImplementedError()
+
+    def user_presence_import_complete(self) -> None:
+        """Override this method to handle operations after presence import is finished (like updating cache)."""
+
 
 def create_and_run_plugin(plugin_class, argv):
     """Call this method as an entry point for the implemented integration.

src/galaxy/api/types.py

@@ -1,7 +1,7 @@
 from dataclasses import dataclass
 from typing import Dict, List, Optional
 
-from galaxy.api.consts import LicenseType, LocalGameState
+from galaxy.api.consts import LicenseType, LocalGameState, PresenceState
 
 
 @dataclass
@@ -174,3 +174,18 @@ class GameLibrarySettings:
     game_id: str
     tags: Optional[List[str]]
     hidden: Optional[bool]
+
+
+@dataclass
+class UserPresence:
+    """Presence information of a user.
+
+    :param presence_state: the state of the user
+    :param game_id: id of the game a user is currently in
+    :param game_title: name of the game a user is currently in
+    :param presence_status: detailed user's presence description
+    """
+    presence_state: PresenceState
+    game_id: Optional[str] = None
+    game_title: Optional[str] = None
+    presence_status: Optional[str] = None

tests/conftest.py

@@ -1,33 +1,38 @@
-from contextlib import ExitStack
 import logging
-from unittest.mock import patch, MagicMock
+from contextlib import ExitStack
+from unittest.mock import MagicMock, patch
 
 import pytest
 
-from galaxy.api.plugin import Plugin
 from galaxy.api.consts import Platform
+from galaxy.api.plugin import Plugin
 from galaxy.unittest.mock import async_return_value
 
+
 @pytest.fixture()
 def reader():
     stream = MagicMock(name="stream_reader")
     stream.read = MagicMock()
     yield stream
 
+
 @pytest.fixture()
 async def writer():
     stream = MagicMock(name="stream_writer")
     stream.drain.side_effect = lambda: async_return_value(None)
     yield stream
 
+
 @pytest.fixture()
 def read(reader):
     yield reader.read
 
+
 @pytest.fixture()
 def write(writer):
     yield writer.write
 
+
 @pytest.fixture()
 async def plugin(reader, writer):
     """Return plugin instance with all feature methods mocked"""
@@ -56,6 +61,9 @@ async def plugin(reader, writer):
         "get_os_compatibility",
         "prepare_os_compatibility_context",
         "os_compatibility_import_complete",
+        "get_user_presence",
+        "prepare_user_presence_context",
+        "user_presence_import_complete",
     )
 
     with ExitStack() as stack:

tests/test_achievements.py

@@ -5,7 +5,7 @@ from pytest import raises
 
 from galaxy.api.types import Achievement
 from galaxy.api.errors import BackendError
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -201,6 +201,7 @@ async def test_import_in_progress(plugin, read, write):
 async def test_unlock_achievement(plugin, write):
     achievement = Achievement(achievement_id="lvl20", unlock_time=1548422395)
     plugin.unlock_achievement("14", achievement)
+    await skip_loop()
     response = json.loads(write.call_args[0][0])
 
     assert response == {

tests/test_authenticate.py

@@ -5,7 +5,7 @@ from galaxy.api.errors import (
     UnknownError, InvalidCredentials, NetworkError, LoggedInElsewhere, ProtocolError,
     BackendNotAvailable, BackendTimeout, BackendError, TemporaryBlocked, Banned, AccessDenied
 )
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -97,6 +97,7 @@ async def test_store_credentials(plugin, write):
         "token": "ABC"
     }
     plugin.store_credentials(credentials)
+    await skip_loop()
 
     assert get_messages(write) == [
         {
@@ -110,6 +111,7 @@ async def test_store_credentials(plugin, write):
 @pytest.mark.asyncio
 async def test_lost_authentication(plugin, write):
     plugin.lost_authentication()
+    await skip_loop()
 
     assert get_messages(write) == [
         {

tests/test_features.py

@@ -16,7 +16,8 @@ def test_base_class():
         Feature.ShutdownPlatformClient,
         Feature.LaunchPlatformClient,
         Feature.ImportGameLibrarySettings,
-        Feature.ImportOSCompatibility
+        Feature.ImportOSCompatibility,
+        Feature.ImportUserPresence
     }
 
 

tests/test_friends.py

@@ -1,6 +1,6 @@
 from galaxy.api.types import FriendInfo
 from galaxy.api.errors import UnknownError
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 import pytest
 
@@ -67,6 +67,7 @@ async def test_add_friend(plugin, write):
     friend = FriendInfo("7", "Kuba")
 
     plugin.add_friend(friend)
+    await skip_loop()
 
     assert get_messages(write) == [
         {
@@ -82,6 +83,7 @@ async def test_add_friend(plugin, write):
 @pytest.mark.asyncio
 async def test_remove_friend(plugin, write):
     plugin.remove_friend("5")
+    await skip_loop()
 
     assert get_messages(write) == [
         {

tests/test_game_times.py

@@ -3,7 +3,7 @@ from unittest.mock import call
 import pytest
 from galaxy.api.types import GameTime
 from galaxy.api.errors import BackendError
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -199,6 +199,7 @@ async def test_import_in_progress(plugin, read, write):
 async def test_update_game(plugin, write):
     game_time = GameTime("3", 60, 1549550504)
     plugin.update_game_time(game_time)
+    await skip_loop()
 
     assert get_messages(write) == [
         {

tests/test_local_games.py

@@ -3,7 +3,7 @@ import pytest
 from galaxy.api.types import LocalGame
 from galaxy.api.consts import LocalGameState
 from galaxy.api.errors import UnknownError, FailedParsingManifest
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -83,6 +83,7 @@ async def test_failure(plugin, read, write, error, code, message):
 async def test_local_game_state_update(plugin, write):
     game = LocalGame("1", LocalGameState.Running)
     plugin.update_local_game_status(game)
+    await skip_loop()
 
     assert get_messages(write) == [
         {

tests/test_owned_games.py

@@ -3,7 +3,7 @@ import pytest
 from galaxy.api.types import Game, Dlc, LicenseInfo
 from galaxy.api.consts import LicenseType
 from galaxy.api.errors import UnknownError
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -100,6 +100,7 @@ async def test_failure(plugin, read, write):
 async def test_add_game(plugin, write):
     game = Game("3", "Doom", None, LicenseInfo(LicenseType.SinglePurchase, None))
     plugin.add_game(game)
+    await skip_loop()
     assert get_messages(write) == [
         {
             "jsonrpc": "2.0",
@@ -120,6 +121,7 @@ async def test_add_game(plugin, write):
 @pytest.mark.asyncio
 async def test_remove_game(plugin, write):
     plugin.remove_game("5")
+    await skip_loop()
     assert get_messages(write) == [
         {
             "jsonrpc": "2.0",
@@ -135,6 +137,7 @@ async def test_remove_game(plugin, write):
 async def test_update_game(plugin, write):
     game = Game("3", "Doom", None, LicenseInfo(LicenseType.SinglePurchase, None))
     plugin.update_game(game)
+    await skip_loop()
     assert get_messages(write) == [
         {
             "jsonrpc": "2.0",

tests/test_persistent_cache.py

@@ -1,6 +1,6 @@
 import pytest
 
-from galaxy.unittest.mock import async_return_value
+from galaxy.unittest.mock import async_return_value, skip_loop
 
 from tests import create_message, get_messages
 
@@ -57,6 +57,7 @@ async def test_set_cache(plugin, write, cache_data):
 
     plugin.persistent_cache.update(cache_data)
     plugin.push_cache()
+    await skip_loop()
 
     assert_rpc_request(write, "push_cache", cache_data)
     assert cache_data == plugin.persistent_cache
@@ -68,6 +69,7 @@ async def test_clear_cache(plugin, write, cache_data):
 
     plugin.persistent_cache.clear()
     plugin.push_cache()
+    await skip_loop()
 
     assert_rpc_request(write, "push_cache", {})
     assert {} == plugin.persistent_cache

tests/test_user_presence.py

@@ -0,0 +1,231 @@
+from unittest.mock import call
+
+import pytest
+
+from galaxy.api.consts import PresenceState
+from galaxy.api.errors import BackendError
+from galaxy.api.types import UserPresence
+from galaxy.unittest.mock import async_return_value
+from tests import create_message, get_messages
+
+
+@pytest.mark.asyncio
+async def test_get_user_presence_success(plugin, read, write):
+    context = "abc"
+    user_ids = ["666", "13", "42", "69"]
+    plugin.prepare_user_presence_context.return_value = async_return_value(context)
+    request = {
+        "jsonrpc": "2.0",
+        "id": "11",
+        "method": "start_user_presence_import",
+        "params": {"user_ids": user_ids}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_user_presence.side_effect = [
+        async_return_value(UserPresence(
+            PresenceState.Unknown,
+            "game-id1",
+            None,
+            "unknown state"
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Offline,
+            None,
+            None,
+            "Going to grandma's house"
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Online,
+            "game-id3",
+            "game-title3",
+            "Pew pew"
+        )),
+        async_return_value(UserPresence(
+            PresenceState.Away,
+            None,
+            "game-title4",
+            "AFKKTHXBY"
+        )),
+    ]
+    await plugin.run()
+    plugin.get_user_presence.assert_has_calls([
+        call(user_id, context) for user_id in user_ids
+    ])
+    plugin.user_presence_import_complete.assert_called_once_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": "11",
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "666",
+                "presence": {
+                    "presence_state": PresenceState.Unknown.value,
+                    "game_id": "game-id1",
+                    "presence_status": "unknown state"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "13",
+                "presence": {
+                    "presence_state": PresenceState.Offline.value,
+                    "presence_status": "Going to grandma's house"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "42",
+                "presence": {
+                    "presence_state": PresenceState.Online.value,
+                    "game_id": "game-id3",
+                    "game_title": "game-title3",
+                    "presence_status": "Pew pew"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_success",
+            "params": {
+                "user_id": "69",
+                "presence": {
+                    "presence_state": PresenceState.Away.value,
+                    "game_title": "game-title4",
+                    "presence_status": "AFKKTHXBY"
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("exception,code,message", [
+    (BackendError, 4, "Backend error"),
+    (KeyError, 0, "Unknown error")
+])
+async def test_get_user_presence_error(exception, code, message, plugin, read, write):
+    user_id = "69"
+    request_id = "55"
+    plugin.prepare_user_presence_context.return_value = async_return_value(None)
+    request = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "method": "start_user_presence_import",
+        "params": {"user_ids": [user_id]}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    plugin.get_user_presence.side_effect = exception
+    await plugin.run()
+    plugin.get_user_presence.assert_called()
+    plugin.user_presence_import_complete.assert_called_once_with()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "result": None
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_failure",
+            "params": {
+                "user_id": user_id,
+                "error": {
+                    "code": code,
+                    "message": message
+                }
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "method": "user_presence_import_finished",
+            "params": None
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_prepare_get_user_presence_context_error(plugin, read, write):
+    request_id = "31415"
+    plugin.prepare_user_presence_context.side_effect = BackendError()
+    request = {
+        "jsonrpc": "2.0",
+        "id": request_id,
+        "method": "start_user_presence_import",
+        "params": {"user_ids": ["6"]}
+    }
+    read.side_effect = [async_return_value(create_message(request)), async_return_value(b"", 10)]
+    await plugin.run()
+
+    assert get_messages(write) == [
+        {
+            "jsonrpc": "2.0",
+            "id": request_id,
+            "error": {
+                "code": 4,
+                "message": "Backend error"
+            }
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_import_already_in_progress_error(plugin, read, write):
+    plugin.prepare_user_presence_context.return_value = async_return_value(None)
+    requests = [
+        {
+            "jsonrpc": "2.0",
+            "id": "3",
+            "method": "start_user_presence_import",
+            "params": {
+                "user_ids": ["42"]
+            }
+        },
+        {
+            "jsonrpc": "2.0",
+            "id": "4",
+            "method": "start_user_presence_import",
+            "params": {
+                "user_ids": ["666"]
+            }
+        }
+    ]
+    read.side_effect = [
+        async_return_value(create_message(requests[0])),
+        async_return_value(create_message(requests[1])),
+        async_return_value(b"", 10)
+    ]
+
+    await plugin.run()
+
+    responses = get_messages(write)
+    assert {
+        "jsonrpc": "2.0",
+        "id": "3",
+        "result": None
+    } in responses
+    assert {
+         "jsonrpc": "2.0",
+         "id": "4",
+         "error": {
+             "code": 600,
+             "message": "Import already in progress"
+         }
+    } in responses