Axsuarez/streaming extensions (microsoft#986)

* Initial commit for POC. This is WIP

* Updates for POC. This is WIP

* Pylint: POC updates

* Updates on POC, protocol adapter pending

* Updates on POC, protocol adapter in progress

* POC almost ready for testing, changes on BFAdapter pending

* POC waiting on client injection in connector

* black: POC waiting on client injection in connector

* POC for http client injection in connector

* got rid of importing errors when loading libraries. Currently in the process of testing.

* Fix couple of errors, still in debugging phase. Initial receive doesnt work with current structure.

* Several fixes including deadlock in threading, serialization and minor logic bugs. PayloadStream logic pending.

* More errors fixed, trying to fit websocket into ms rest pipeline. Receiving is working with some bugs.

* Disassembler fixes, sender struggling to send through socket

* changes on disassembler and receiver

* adding streaming to ci pipeline

* Pylint fixes

* updated streaming setup.py

* Removing 3.6

* Changing all concurrent mechanisms in streaming to asyncio

* Added validation for abrupt closing of websocket, added tracebacks and async validation for disconnecting callbacks

* UnblockActivityProcessorThread

* Header serialization fix and stream serialization fix.

* Parity change in the internal buffer structure of the payload_stream object, fixes on stream writing behavior in web_socket_transport and send response instead of request fix in protocol adapter.

* Fixes on the RecieveResponse path

* PayloadStream length fix

* Grouping related imports

* payload receiver unit test (microsoft#1664)

* Payload sender unit test (microsoft#1666)

* payload receiver unit test

* senderTest

* disconnect

* fixWarning

Co-authored-by: Axel Suarez <[email protected]>

* blackcheck

* pylintfix

* test_req_processor (microsoft#1668)

* Axsuarez/streaming receive loop unittest (microsoft#1667)

* payload receiver unit test

* StreamingRequestHandler test listen

* renaming straming to botframework scope

* Updating pipeline

* Removing sleep() safety measure

* Remove unused import

Co-authored-by: msomanathan <[email protected]>
Co-authored-by: Muthuveer Somanathan <[email protected]>

Author: axelsrz

.pylintrc

@@ -158,7 +158,9 @@ disable=print-statement,
         too-many-return-statements,
         import-error,
         no-name-in-module,
-        too-many-branches
+        too-many-branches,
+        too-many-ancestors,
+        too-many-nested-blocks
 
 # Enable the message, report, category or checker with the given id(s). You can
 # either give multiple identifier separated by comma (,) or put this option

libraries/botbuilder-core/botbuilder/core/bot_framework_adapter.py

@@ -1282,6 +1282,16 @@ async def exchange_token_from_credentials(
             )
         raise TypeError(f"exchange token returned improper result: {type(result)}")
 
+    def can_process_outgoing_activity(
+        self, activity: Activity  # pylint: disable=unused-argument
+    ) -> bool:
+        return False
+
+    async def process_outgoing_activity(
+        self, turn_context: TurnContext, activity: Activity
+    ) -> ResourceResponse:
+        raise Exception("NotImplemented")
+
     @staticmethod
     def key_for_connector_client(service_url: str, app_id: str, scope: str):
         return f"{service_url if service_url else ''}:{app_id if app_id else ''}:{scope if scope else ''}"

libraries/botbuilder-core/botbuilder/core/integration/aiohttp_channel_service_exception_middleware.py

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
+import traceback
+
 from aiohttp.web import (
     middleware,
     HTTPNotImplemented,
@@ -26,4 +28,5 @@ async def aiohttp_error_middleware(request, handler):
     except KeyError:
         raise HTTPNotFound()
     except Exception:
+        traceback.print_exc()
         raise HTTPInternalServerError()

libraries/botbuilder-core/botbuilder/core/streaming/__init__.py

@@ -0,0 +1,14 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from .bot_framework_http_adapter_base import BotFrameworkHttpAdapterBase
+from .streaming_activity_processor import StreamingActivityProcessor
+from .streaming_request_handler import StreamingRequestHandler
+from .version_info import VersionInfo
+
+__all__ = [
+    "BotFrameworkHttpAdapterBase",
+    "StreamingActivityProcessor",
+    "StreamingRequestHandler",
+    "VersionInfo",
+]

libraries/botbuilder-core/botbuilder/core/streaming/bot_framework_http_adapter_base.py

@@ -0,0 +1,125 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from http import HTTPStatus
+from typing import Awaitable, Callable, List
+
+from botbuilder.core import (
+    Bot,
+    BotFrameworkAdapter,
+    BotFrameworkAdapterSettings,
+    InvokeResponse,
+    TurnContext,
+)
+from botbuilder.schema import Activity, ActivityTypes, ResourceResponse
+from botframework.connector import AsyncBfPipeline, BotFrameworkConnectorConfiguration
+from botframework.connector.aio import ConnectorClient
+from botframework.connector.auth import (
+    ClaimsIdentity,
+    MicrosoftAppCredentials,
+    MicrosoftGovernmentAppCredentials,
+)
+
+from .streaming_activity_processor import StreamingActivityProcessor
+from .streaming_request_handler import StreamingRequestHandler
+from .streaming_http_client import StreamingHttpDriver
+
+
+class BotFrameworkHttpAdapterBase(BotFrameworkAdapter, StreamingActivityProcessor):
+    # pylint: disable=pointless-string-statement
+    def __init__(self, settings: BotFrameworkAdapterSettings):
+        super().__init__(settings)
+
+        self.connected_bot: Bot = None
+        self.claims_identity: ClaimsIdentity = None
+        self.request_handlers: List[StreamingRequestHandler] = None
+
+    async def process_streaming_activity(
+        self,
+        activity: Activity,
+        bot_callback_handler: Callable[[TurnContext], Awaitable],
+    ) -> InvokeResponse:
+        if not activity:
+            raise TypeError(
+                f"'activity: {activity.__class__.__name__}' argument can't be None"
+            )
+
+        """
+        If a conversation has moved from one connection to another for the same Channel or Skill and
+        hasn't been forgotten by the previous StreamingRequestHandler. The last requestHandler
+        the conversation has been associated with should always be the active connection.
+        """
+        request_handler = [
+            handler
+            for handler in self.request_handlers
+            if handler.service_url == activity.service_url
+            and handler.has_conversation(activity.conversation.id)
+        ]
+        request_handler = request_handler[-1] if request_handler else None
+        context = TurnContext(self, activity)
+
+        if self.claims_identity:
+            context.turn_state[self.BOT_IDENTITY_KEY] = self.claims_identity
+
+        connector_client = self._create_streaming_connector_client(
+            activity, request_handler
+        )
+        context.turn_state[self.BOT_CONNECTOR_CLIENT_KEY] = connector_client
+
+        await self.run_pipeline(context, bot_callback_handler)
+
+        if activity.type == ActivityTypes.invoke:
+            activity_invoke_response = context.turn_state.get(self._INVOKE_RESPONSE_KEY)
+
+            if not activity_invoke_response:
+                return InvokeResponse(status=HTTPStatus.NOT_IMPLEMENTED)
+            return activity_invoke_response.value
+
+        return None
+
+    async def send_streaming_activity(self, activity: Activity) -> ResourceResponse:
+        raise NotImplementedError()
+
+    def can_process_outgoing_activity(self, activity: Activity) -> bool:
+        if not activity:
+            raise TypeError(
+                f"'activity: {activity.__class__.__name__}' argument can't be None"
+            )
+
+        return not activity.service_url.startswith("https")
+
+    async def process_outgoing_activity(
+        self, turn_context: TurnContext, activity: Activity
+    ) -> ResourceResponse:
+        if not activity:
+            raise TypeError(
+                f"'activity: {activity.__class__.__name__}' argument can't be None"
+            )
+
+        # TODO: Check if we have token responses from OAuth cards.
+
+        # The ServiceUrl for streaming channels begins with the string "urn" and contains
+        # information unique to streaming connections. Now that we know that this is a streaming
+        # activity, process it in the streaming pipeline.
+        # Process streaming activity.
+        return await self.send_streaming_activity(activity)
+
+    def _create_streaming_connector_client(
+        self, activity: Activity, request_handler: StreamingRequestHandler
+    ) -> ConnectorClient:
+        empty_credentials = (
+            MicrosoftAppCredentials.empty()
+            if self._channel_provider and self._channel_provider.is_government()
+            else MicrosoftGovernmentAppCredentials.empty()
+        )
+        streaming_driver = StreamingHttpDriver(request_handler)
+        config = BotFrameworkConnectorConfiguration(
+            empty_credentials,
+            activity.service_url,
+            pipeline_type=AsyncBfPipeline,
+            driver=streaming_driver,
+        )
+        streaming_driver.config = config
+        connector_client = ConnectorClient(None, custom_configuration=config)
+
+        return connector_client

libraries/botbuilder-core/botbuilder/core/streaming/streaming_activity_processor.py

@@ -0,0 +1,21 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from abc import ABC
+from typing import Awaitable, Callable
+
+from botbuilder.core import TurnContext, InvokeResponse
+from botbuilder.schema import Activity
+
+
+class StreamingActivityProcessor(ABC):
+    """
+    Process streaming activities.
+    """
+
+    async def process_streaming_activity(
+        self,
+        activity: Activity,
+        bot_callback_handler: Callable[[TurnContext], Awaitable],
+    ) -> InvokeResponse:
+        raise NotImplementedError()

libraries/botbuilder-core/botbuilder/core/streaming/streaming_http_client.py

@@ -0,0 +1,94 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from http import HTTPStatus
+from logging import Logger
+from typing import Any
+
+from msrest.universal_http import ClientRequest
+from msrest.universal_http.async_abc import AsyncClientResponse
+from msrest.universal_http.async_requests import (
+    AsyncRequestsHTTPSender as AsyncRequestsHTTPDriver,
+)
+from botframework.streaming import StreamingRequest, ReceiveResponse
+
+from .streaming_request_handler import StreamingRequestHandler
+
+
+class StreamingProtocolClientResponse(AsyncClientResponse):
+    def __init__(
+        self, request: StreamingRequest, streaming_response: ReceiveResponse
+    ) -> None:
+        super(StreamingProtocolClientResponse, self).__init__(
+            request, streaming_response
+        )
+        # https://aiohttp.readthedocs.io/en/stable/client_reference.html#aiohttp.ClientResponse
+        self.status_code = streaming_response.status_code
+        # self.headers = streaming_response.headers
+        # self.reason = streaming_response.reason
+        self._body = None
+
+    def body(self) -> bytes:
+        """Return the whole body as bytes in memory.
+        """
+        if not self._body:
+            return bytes([])
+        return self._body
+
+    async def load_body(self) -> None:
+        """Load in memory the body, so it could be accessible from sync methods."""
+        self._body: ReceiveResponse
+        self._body = self.internal_response.read_body()
+
+    def raise_for_status(self):
+        if 400 <= self.internal_response.status_code <= 599:
+            raise Exception(f"Http error: {self.internal_response.status_code}")
+
+
+class StreamingHttpDriver(AsyncRequestsHTTPDriver):
+    def __init__(
+        self,
+        request_handler: StreamingRequestHandler,
+        *,
+        config=None,
+        logger: Logger = None,
+    ):
+        super().__init__(config)
+        if not request_handler:
+            raise TypeError(
+                f"'request_handler: {request_handler.__class__.__name__}' argument can't be None"
+            )
+        self._request_handler = request_handler
+        self._logger = logger
+
+    async def send(
+        self, request: ClientRequest, **config: Any  # pylint: disable=unused-argument
+    ) -> AsyncClientResponse:
+        # TODO: validate form of request to perform operations
+        streaming_request = StreamingRequest(
+            path=request.url[request.url.index("v3/") :], verb=request.method
+        )
+        streaming_request.set_body(request.data)
+
+        return await self._send_request(streaming_request)
+
+    async def _send_request(
+        self, request: StreamingRequest
+    ) -> StreamingProtocolClientResponse:
+        try:
+            server_response = await self._request_handler.send_streaming_request(
+                request
+            )
+
+            if not server_response:
+                raise Exception("Server response from streaming request is None")
+
+            if server_response.status_code == HTTPStatus.OK:
+                # TODO: this should be an object read from json
+
+                return StreamingProtocolClientResponse(request, server_response)
+        except Exception as error:
+            # TODO: log error
+            raise error
+
+        return None