TypingMiddleware

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

@@ -25,6 +25,7 @@
 from .null_telemetry_client import NullTelemetryClient
 from .recognizer import Recognizer
 from .recognizer_result import RecognizerResult, TopIntent
+from .show_typing_middleware import ShowTypingMiddleware
 from .state_property_accessor import StatePropertyAccessor
 from .state_property_info import StatePropertyInfo
 from .storage import Storage, StoreItem, calculate_change_hash
@@ -56,6 +57,7 @@
     "NullTelemetryClient",
     "Recognizer",
     "RecognizerResult",
+    "ShowTypingMiddleware",
     "StatePropertyAccessor",
     "StatePropertyInfo",
     "Storage",

libraries/botbuilder-core/botbuilder/core/show_typing_middleware.py

@@ -0,0 +1,92 @@
+import time
+from functools import wraps
+from typing import Awaitable, Callable
+
+from botbuilder.schema import Activity, ActivityTypes
+
+from .middleware_set import Middleware
+from .turn_context import TurnContext
+
+
+def delay(span=0.0):
+    def wrap(func):
+        @wraps(func)
+        async def delayed():
+            time.sleep(span)
+            await func()
+
+        return delayed
+
+    return wrap
+
+
+class Timer:
+    clear_timer = False
+
+    async def set_timeout(self, func, time):
+        is_invocation_cancelled = False
+
+        @delay(time)
+        async def some_fn():  # pylint: disable=function-redefined
+            if not self.clear_timer:
+                await func()
+
+        await some_fn()
+        return is_invocation_cancelled
+
+    def set_clear_timer(self):
+        self.clear_timer = True
+
+
+class ShowTypingMiddleware(Middleware):
+    def __init__(self, delay: float = 0.5, period: float = 2.0):
+        if delay < 0:
+            raise ValueError("Delay must be greater than or equal to zero")
+
+        if period <= 0:
+            raise ValueError("Repeat period must be greater than zero")
+
+        self._delay = delay
+        self._period = period
+
+    async def on_process_request(
+        self, context: TurnContext, logic: Callable[[TurnContext], Awaitable]
+    ):
+        finished = False
+        timer = Timer()
+
+        async def start_interval(context: TurnContext, delay: int, period: int):
+            async def aux():
+                if not finished:
+                    typing_activity = Activity(
+                        type=ActivityTypes.typing,
+                        relates_to=context.activity.relates_to,
+                    )
+
+                    conversation_reference = TurnContext.get_conversation_reference(
+                        context.activity
+                    )
+
+                    typing_activity = TurnContext.apply_conversation_reference(
+                        typing_activity, conversation_reference
+                    )
+
+                    await context.adapter.send_activities(context, [typing_activity])
+
+                    start_interval(context, period, period)
+
+            await timer.set_timeout(aux, delay)
+
+        def stop_interval():
+            nonlocal finished
+            finished = True
+            timer.set_clear_timer()
+
+        if context.activity.type == ActivityTypes.message:
+            finished = False
+            await start_interval(context, self._delay, self._period)
+
+        result = await logic()
+        stop_interval()
+
+        return result

libraries/botbuilder-core/tests/test_show_typing_middleware.py

@@ -0,0 +1,65 @@
+import time
+import aiounittest
+
+from botbuilder.core import ShowTypingMiddleware
+from botbuilder.core.adapters import TestAdapter
+from botbuilder.schema import ActivityTypes
+
+
+class TestShowTypingMiddleware(aiounittest.AsyncTestCase):
+    async def test_should_automatically_send_a_typing_indicator(self):
+        async def aux(context):
+            time.sleep(0.600)
+            await context.send_activity(f"echo:{context.activity.text}")
+
+        def assert_is_typing(activity, description):  # pylint: disable=unused-argument
+            assert activity.type == ActivityTypes.typing
+
+        adapter = TestAdapter(aux)
+        adapter.use(ShowTypingMiddleware())
+
+        step1 = await adapter.send("foo")
+        step2 = await step1.assert_reply(assert_is_typing)
+        step3 = await step2.assert_reply("echo:foo")
+        step4 = await step3.send("bar")
+        step5 = await step4.assert_reply(assert_is_typing)
+        await step5.assert_reply("echo:bar")
+
+    async def test_should_not_automatically_send_a_typing_indicator_if_no_middleware(
+        self
+    ):
+        async def aux(context):
+            await context.send_activity(f"echo:{context.activity.text}")
+
+        adapter = TestAdapter(aux)
+
+        step1 = await adapter.send("foo")
+        await step1.assert_reply("echo:foo")
+
+    async def test_should_not_immediately_respond_with_message(self):
+        async def aux(context):
+            time.sleep(0.600)
+            await context.send_activity(f"echo:{context.activity.text}")
+
+        def assert_is_not_message(
+            activity, description
+        ):  # pylint: disable=unused-argument
+            assert activity.type != ActivityTypes.message
+
+        adapter = TestAdapter(aux)
+        adapter.use(ShowTypingMiddleware())
+
+        step1 = await adapter.send("foo")
+        await step1.assert_reply(assert_is_not_message)
+
+    async def test_should_immediately_respond_with_message_if_no_middleware(self):
+        async def aux(context):
+            await context.send_activity(f"echo:{context.activity.text}")
+
+        def assert_is_message(activity, description):  # pylint: disable=unused-argument
+            assert activity.type == ActivityTypes.message
+
+        adapter = TestAdapter(aux)
+
+        step1 = await adapter.send("foo")
+        await step1.assert_reply(assert_is_message)