Skip to content

Add connect_named_pipe Method to CloudAdapter for DirectLine Speech Support #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cosineai wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add connect_named_pipe Method to CloudAdapter for DirectLine Speech Support #1

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 50 additions & 3 deletions libraries/botbuilder-integration-aiohttp/botbuilder/integration/aiohttp/cloud_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,16 +25,24 @@
StreamingRequestHandler,
)
from botbuilder.schema import Activity
from botbuilder.integration.aiohttp.streaming import AiohttpWebSocket
from botframework.connector import AsyncBfPipeline, BotFrameworkConnectorConfiguration
from botframework.connector.aio import ConnectorClient
from botbuilder.integration.aiohttp.streaming import AiohttpWebSocket
from botframework.connector import (
AsyncBfPipeline,
BotFrameworkConnectorConfiguration,
ConnectorClient,
)
from botframework.connector.auth import (
AuthenticateRequestResult,
BotFrameworkAuthentication,
BotFrameworkAuthenticationFactory,
ConnectorFactory,
MicrosoftAppCredentials,
)
from botframework.streaming import (
NamedPipeClient,
PayloadTypes,
StreamingRequest,
)

from .bot_framework_http_adapter_integration_base import (
BotFrameworkHttpAdapterIntegrationBase,
Expand Down Expand Up @@ -138,6 +146,38 @@ async def _connect(
await streaming_activity_processor.listen()


class _NamedPipeClient(NamedPipeClient):
def __init__(self, request_handler: StreamingRequestHandler) -> None:
super().__init__()
self._request_handler = request_handler

async def connect(self, server_name: str, pipe_name: str):
if not server_name:
raise TypeError("server_name can't be None")
if not pipe_name:
raise TypeError("pipe_name can't be None")

await self.start_async(server_name, pipe_name)

async def send(self, request: StreamingRequest):
if not request:
raise TypeError("request can't be None")

response = await self._request_handler.process_request(request)

if response.streams:
response_streams = response.streams
content_stream = response_streams[0]

if content_stream.content_type == PayloadTypes.response:
assembler = self._payload_assembler_manager.get_payload_assembler(
content_stream.id
)
response = await assembler.get_stream().read_as_json()
return response

return None

class _StreamingActivityProcessor(StreamingActivityProcessor):
def __init__(
self,
Expand All @@ -157,6 +197,13 @@ def __init__(
_StreamingConnectorFactory(self._request_handler)
)

async def connect_named_pipe(self, pipe_name: str):
if not pipe_name:
raise TypeError("pipe_name can't be None")

named_pipe_client = _NamedPipeClient(self._request_handler)
await named_pipe_client.connect("", pipe_name)

async def listen(self):
await self._request_handler.listen()

Expand Down