CAIP-119 Fix datahose loop (#376)

* Fix datahose implementation: use datahose api in datahose loop replacing the datafeed api

Author: catalinsymphony

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "symphony_bdk_python"
-version = "2.11.0"
+version = "2.11.1"
 license = "Apache-2.0"
 description = "Symphony Bot Development Kit for Python"
 readme = "README.md"

symphony/bdk/core/service/datafeed/abstract_ackId_event_loop.py

@@ -1,13 +1,15 @@
 import logging
 import time
 from abc import ABC, abstractmethod
+from typing import Union
 
 from symphony.bdk.core.auth.auth_session import AuthSession
 from symphony.bdk.core.config.model.bdk_config import BdkConfig
 from symphony.bdk.core.service.datafeed.abstract_datafeed_loop import AbstractDatafeedLoop
 from symphony.bdk.core.service.datafeed.exception import EventError
 from symphony.bdk.core.service.session.session_service import SessionService
 from symphony.bdk.gen.agent_api.datafeed_api import DatafeedApi
+from symphony.bdk.gen.agent_api.datahose_api import DatahoseApi
 
 EVENT_PROCESSING_MAX_DURATION_SECONDS = 30
 
@@ -19,14 +21,14 @@ class AbstractAckIdEventLoop(AbstractDatafeedLoop, ABC):
 
     def __init__(
         self,
-        datafeed_api: DatafeedApi,
+        datafeed_api: Union[DatafeedApi, DatahoseApi],
         session_service: SessionService,
         auth_session: AuthSession,
         config: BdkConfig,
     ):
         """
 
-        :param datafeed_api: DatafeedApi to request the service
+        :param datafeed_api: DatafeedApi or DatahoseApi to request the service
         :param session_service: the SessionService to get user session information
         :param auth_session: the AuthSession instance used to get session and key manager tokens
         :param config: the bot configuration

symphony/bdk/core/service/datafeed/abstract_datafeed_loop.py

@@ -6,13 +6,14 @@
 from asyncio import Task
 from contextvars import ContextVar
 from enum import Enum
-from typing import List
+from typing import List, Union
 
 from symphony.bdk.core.auth.auth_session import AuthSession
 from symphony.bdk.core.config.model.bdk_config import BdkConfig
 from symphony.bdk.core.service.datafeed.real_time_event_listener import RealTimeEventListener
 from symphony.bdk.core.service.session.session_service import SessionService
 from symphony.bdk.gen.agent_api.datafeed_api import DatafeedApi
+from symphony.bdk.gen.agent_api.datahose_api import DatahoseApi
 from symphony.bdk.gen.agent_model.v4_event import V4Event
 
 logger = logging.getLogger(__name__)
@@ -72,14 +73,14 @@ class AbstractDatafeedLoop(ABC):
 
     def __init__(
         self,
-        datafeed_api: DatafeedApi,
+        datafeed_api: Union[DatafeedApi, DatahoseApi],
         session_service: SessionService,
         auth_session: AuthSession,
         config: BdkConfig,
     ):
         """
 
-        :param datafeed_api: DatafeedApi to request the service
+        :param datafeed_api: DatafeedApi or DatahoseApi to request the service
         :param session_service: the SessionService to get user session information
         :param auth_session: the AuthSession instance used to get session and key manager tokens
         :param config: the bot configuration

symphony/bdk/core/service/datafeed/datahose_loop.py

@@ -7,7 +7,7 @@
 from symphony.bdk.core.service.datafeed.abstract_ackId_event_loop import AbstractAckIdEventLoop
 from symphony.bdk.core.service.datafeed.abstract_datahose_loop import AbstractDatahoseLoop
 from symphony.bdk.core.service.session.session_service import SessionService
-from symphony.bdk.gen.agent_api.datafeed_api import DatafeedApi
+from symphony.bdk.gen.agent_api.datahose_api import DatahoseApi
 from symphony.bdk.gen.agent_model.v5_events_read_body import V5EventsReadBody
 
 # DFv2 API authorizes a maximum length for the tag parameter
@@ -30,12 +30,12 @@ class DatahoseLoop(AbstractAckIdEventLoop, AbstractDatahoseLoop):
 
     def __init__(
         self,
-        datafeed_api: DatafeedApi,
+        datahose_api: DatahoseApi,
         session_service: SessionService,
         auth_session: AuthSession,
         config: BdkConfig,
     ):
-        super().__init__(datafeed_api, session_service, auth_session, config)
+        super().__init__(datahose_api, session_service, auth_session, config)
         if config.datahose is not None:
             not_truncated_tag = (
                 config.datahose.tag

symphony/bdk/core/service_factory.py

@@ -30,6 +30,7 @@
 from symphony.bdk.gen.agent_api.attachments_api import AttachmentsApi
 from symphony.bdk.gen.agent_api.audit_trail_api import AuditTrailApi
 from symphony.bdk.gen.agent_api.datafeed_api import DatafeedApi
+from symphony.bdk.gen.agent_api.datahose_api import DatahoseApi
 from symphony.bdk.gen.agent_api.share_api import ShareApi
 from symphony.bdk.gen.agent_api.signals_api import SignalsApi
 from symphony.bdk.gen.agent_api.system_api import SystemApi as AgentSystemApi
@@ -175,7 +176,7 @@ def get_datahose_loop(self) -> AbstractDatahoseLoop:
         """
         if self._config.datahose is not None:
             return DatahoseLoop(
-                DatafeedApi(self._agent_client),
+                DatahoseApi(self._agent_client),
                 self._session_service,
                 self._auth_session,
                 self._config,