Spaces:
Build error
Build error
File size: 3,088 Bytes
0827183 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 |
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
import asyncio
from uuid import UUID, uuid4
from botframework.streaming.payloads import (
PayloadAssemblerManager,
RequestManager,
SendOperations,
StreamManager,
)
from botframework.streaming.payloads.assemblers import PayloadStreamAssembler
from botframework.streaming.payload_transport import PayloadSender, PayloadReceiver
from .receive_request import ReceiveRequest
from .receive_response import ReceiveResponse
from .request_handler import RequestHandler
from .streaming_request import StreamingRequest
class ProtocolAdapter:
def __init__(
self,
request_handler: RequestHandler,
request_manager: RequestManager,
payload_sender: PayloadSender,
payload_receiver: PayloadReceiver,
handler_context: object = None,
):
self._request_handler = request_handler
self._request_manager = request_manager
self._payload_sender = payload_sender
self._payload_receiver = payload_receiver
self._handler_context = handler_context
self._send_operations = SendOperations(self._payload_sender)
# TODO: might be able to remove
self._stream_manager = StreamManager(self._on_cancel_stream)
self._assembler_manager = PayloadAssemblerManager(
self._stream_manager, self._on_receive_request, self._on_receive_response
)
self._payload_receiver.subscribe(
self._assembler_manager.get_payload_stream,
self._assembler_manager.on_receive,
)
async def send_request(self, request: StreamingRequest) -> ReceiveResponse:
if not request:
raise TypeError(
f"'request: {request.__class__.__name__}' argument can't be None"
)
request_id = uuid4()
response_task = self._request_manager.get_response(request_id)
request_task = self._send_operations.send_request(request_id, request)
[_, response] = await asyncio.gather(request_task, response_task)
return response
async def _on_receive_request(self, identifier: UUID, request: ReceiveRequest):
# request is done, we can handle it
if self._request_handler:
response = await self._request_handler.process_request(
request, None, self._handler_context
)
if response:
await self._send_operations.send_response(identifier, response)
async def _on_receive_response(self, identifier: UUID, response: ReceiveResponse):
# we received the response to something, signal it
await self._request_manager.signal_response(identifier, response)
def _on_cancel_stream(self, content_stream_assembler: PayloadStreamAssembler):
# TODO: on original C# code content_stream_assembler is typed as IAssembler
task = asyncio.create_task(
self._send_operations.send_cancel_stream(
content_stream_assembler.identifier
)
)
|