Datafeed Loop v2 implementation

In the BdkDatafeedConfig class, check if the `datafeed.version` configuration is
specified. If not, the datafeed version equals `v1`

In the ServiceFactory class, check if the datafeed version is configured as `v2`, return
the DatafeedLoopV2 instance. Otherwise, the DatafeedLoopV1 will be used by default.

Implementation the datafeed v2 loop service in DatafeedLoopV2 class

Author: symphony-hong

symphony/bdk/core/config/model/bdk_datafeed_config.py

@@ -15,7 +15,7 @@ def __init__(self, config):
 
         :param config: the dict containing the datafeed specific confguration.
         """
-        self.version = DF_V1
+        self.version = config.get(VERSION) if VERSION in config else DF_V1
         self.id_file_path = ""
         if config is not None and DF_ID_FILE_PATH in config:
             self.id_file_path = Path(config.get(DF_ID_FILE_PATH))

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

@@ -6,15 +6,21 @@
 from enum import Enum
 from typing import List
 
-from symphony.bdk.core.config.model.bdk_config import BdkConfig
 from symphony.bdk.core.auth.auth_session import AuthSession
-
-from symphony.bdk.gen.agent_api.datafeed_api import DatafeedApi
+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.gen.agent_api.datafeed_api import DatafeedApi
 from symphony.bdk.gen.agent_model.v4_event import V4Event
 
 logger = logging.getLogger(__name__)
 
+
+class DatafeedVersion(Enum):
+
+    V1 = "v1"
+    V2 = "v2"
+
+
 class RealTimeEvent(Enum):
     """This enum lists all possible
     `Real Time Events <https://docs.developers.symphony.com/building-bots-on-symphony/datafeed/real-time-events>`_

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

@@ -0,0 +1,90 @@
+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.gen import ApiException
+from symphony.bdk.gen.agent_api.datafeed_api import DatafeedApi
+from symphony.bdk.gen.agent_model.ack_id import AckId
+from symphony.bdk.gen.agent_model.v5_datafeed import V5Datafeed
+from typing import Optional
+
+
+class DatafeedLoopV2(AbstractDatafeedLoop):
+    """A class for implementing the datafeed v2 loop service.
+
+        This service will be started by calling :func:`~DatafeedLoopV2.start`.
+
+        On the very first run, the BDK bot will try to retrieve the list of datafeed to which it is listening.
+        Since each bot should only listening to just one datafeed, the first datafeed in the list will be used by
+        the bot to be listened to. If the retrieved list is empty, the BDK bot will create a new datafeed to listen.
+
+        The BDK bot will listen to this datafeed to get all the received real-time events.
+
+        If this datafeed becomes stale or faulty, the BDK bot will create the new one for listening.
+
+        This service will be stopped by calling :func:`~DatafeedLoopV1.stop`
+
+        If the datafeed service is stopped during a read datafeed call, it has to wait until the last read finish to be
+        really stopped
+        """
+
+    def __init__(self, datafeed_api: DatafeedApi, auth_session: AuthSession, config: BdkConfig):
+        super().__init__(datafeed_api, auth_session, config)
+        self._ack_id = ""
+        self.started = False
+        self.datafeed_id = None
+
+    async def start(self):
+        datafeed = await self._retrieve_datafeed()
+        if not datafeed:
+            datafeed = await self._create_datafeed()
+        self.datafeed_id = datafeed.id
+        self.started = True
+        while self.started:
+            try:
+                await self._read_datafeed()
+            except ApiException as e:
+                if e.status == 400:
+                    datafeed = await self._recreate_datafeed()
+                    self.datafeed_id = datafeed.id
+                else:
+                    raise e
+
+    async def stop(self):
+        self.started = False
+
+    async def _retrieve_datafeed(self) -> Optional[V5Datafeed]:
+        session_token = await self.auth_session.session_token
+        key_manager_token = await self.auth_session.key_manager_token
+        datafeeds = await self.datafeed_api.list_datafeed(session_token=session_token,
+                                                          key_manager_token=key_manager_token)
+        if datafeeds:
+            return datafeeds[0]
+        return None
+
+    async def _create_datafeed(self) -> V5Datafeed:
+        session_token = await self.auth_session.session_token
+        key_manager_token = await self.auth_session.key_manager_token
+        return await self.datafeed_api.create_datafeed(session_token=session_token, key_manager_token=key_manager_token)
+
+    async def _read_datafeed(self):
+        params = {
+            'session_token': await self.auth_session.session_token,
+            'key_manager_token': await self.auth_session.key_manager_token,
+            'datafeed_id': self.datafeed_id,
+            'ack_id': AckId(ack_id=self._ack_id)
+        }
+        event_list = await self.datafeed_api.read_datafeed(**params)
+        self._ack_id = event_list.ack_id
+        if event_list.events:
+            await self.handle_v4_event_list(events=event_list.events)
+
+    async def _delete_datafeed(self) -> None:
+        session_token = await self.auth_session.session_token
+        key_manager_token = await self.auth_session.key_manager_token
+        await self.datafeed_api.delete_datafeed(datafeed_id=self.datafeed_id,
+                                                session_token=session_token,
+                                                key_manager_token=key_manager_token)
+
+    async def _recreate_datafeed(self):
+        await self._delete_datafeed()
+        return await self._create_datafeed()

symphony/bdk/core/service_factory.py

@@ -2,7 +2,9 @@
 from symphony.bdk.core.client.api_client_factory import ApiClientFactory
 from symphony.bdk.core.config.model.bdk_config import BdkConfig
 from symphony.bdk.core.service.connection.connection_service import ConnectionService
+from symphony.bdk.core.service.datafeed.abstract_datafeed_loop import DatafeedVersion, AbstractDatafeedLoop
 from symphony.bdk.core.service.datafeed.datafeed_loop_v1 import DatafeedLoopV1
+from symphony.bdk.core.service.datafeed.datafeed_loop_v2 import DatafeedLoopV2
 from symphony.bdk.core.service.message.message_service import MessageService
 from symphony.bdk.core.service.message.multi_attachments_messages_api import MultiAttachmentsMessagesApi
 from symphony.bdk.core.service.stream.stream_service import StreamService
@@ -94,11 +96,18 @@ def get_stream_service(self) -> StreamService:
             ShareApi(self._agent_client),
             self._auth_session)
 
-    def get_datafeed_loop(self) -> DatafeedLoopV1:
+    def get_datafeed_loop(self) -> AbstractDatafeedLoop:
         """Returns a fully initialized DatafeedLoop
 
         :return: a new DatafeedLoop instance.
         """
+        df_version = self._config.datafeed.version
+        if df_version.lower() == DatafeedVersion.V2.value.lower():
+            return DatafeedLoopV2(
+                DatafeedApi(self._agent_client),
+                self._auth_session,
+                self._config
+            )
         return DatafeedLoopV1(
             DatafeedApi(self._agent_client),
             self._auth_session,

symphony/bdk/core/symphony_bdk.py

@@ -3,7 +3,7 @@
 from symphony.bdk.core.auth.exception import AuthInitializationException
 from symphony.bdk.core.client.api_client_factory import ApiClientFactory
 from symphony.bdk.core.service.connection.connection_service import ConnectionService
-from symphony.bdk.core.service.datafeed.datafeed_loop_v1 import DatafeedLoopV1
+from symphony.bdk.core.service.datafeed.abstract_datafeed_loop import AbstractDatafeedLoop
 from symphony.bdk.core.service.message.message_service import MessageService
 from symphony.bdk.core.service.stream.stream_service import StreamService
 from symphony.bdk.core.service.user.user_service import UserService
@@ -73,7 +73,7 @@ def streams(self) -> StreamService:
         """
         return self._stream_service
 
-    def datafeed(self) -> DatafeedLoopV1:
+    def datafeed(self) -> AbstractDatafeedLoop:
         """Get the Datafeed loop from the BDK entry point.
 
         :return: The Datafeed Loop instance.