We read every piece of feedback, and take your input very seriously.
2 parents ae48aaa + 90425d3 commit 156b7afCopy full SHA for 156b7af
libraries/botbuilder-core/botbuilder/core/show_typing_middleware.py
@@ -1,8 +1,6 @@
1
# Copyright (c) Microsoft Corporation. All rights reserved.
2
# Licensed under the MIT License.
3
-
4
-import time
5
-from functools import wraps
+import asyncio
6
from typing import Awaitable, Callable
7
8
from botbuilder.schema import Activity, ActivityTypes
@@ -11,38 +9,38 @@
11
9
from .turn_context import TurnContext
12
10
13
14
-def delay(span=0.0):
15
- def wrap(func):
16
- @wraps(func)
17
- async def delayed():
18
- time.sleep(span)
19
- await func()
20
21
- return delayed
22
23
- return wrap
24
25
26
class Timer:
27
clear_timer = False
28
29
- async def set_timeout(self, func, time):
30
- is_invocation_cancelled = False
31
32
- @delay(time)
+ def set_timeout(self, func, span):
33
async def some_fn(): # pylint: disable=function-redefined
+ await asyncio.sleep(span)
34
if not self.clear_timer:
35
await func()
36
37
- await some_fn()
38
- return is_invocation_cancelled
+ asyncio.ensure_future(some_fn())
39
40
def set_clear_timer(self):
41
self.clear_timer = True
42
43
44
class ShowTypingMiddleware(Middleware):
+ """
+ When added, this middleware will send typing activities back to the user when a Message activity
+ is received to let them know that the bot has received the message and is working on the response.
+ You can specify a delay before the first typing activity is sent and then a frequency, which
+ determines how often another typing activity is sent. Typing activities will continue to be sent
+ until your bot sends another message back to the user.
+
45
def __init__(self, delay: float = 0.5, period: float = 2.0):
+ Initializes the middleware.
+ :param delay: Delay in seconds for the first typing indicator to be sent.
+ :param period: Delay in seconds for subsequent typing indicators.
46
if delay < 0:
47
raise ValueError("Delay must be greater than or equal to zero")
48
@@ -55,41 +53,43 @@ def __init__(self, delay: float = 0.5, period: float = 2.0):
55
53
async def on_turn(
56
54
self, context: TurnContext, logic: Callable[[TurnContext], Awaitable]
57
):
58
- finished = False
59
timer = Timer()
60
61
- async def start_interval(context: TurnContext, delay: int, period: int):
+ def start_interval(context: TurnContext, delay, period):
62
async def aux():
63
- if not finished:
64
- typing_activity = Activity(
65
- type=ActivityTypes.typing,
66
- relates_to=context.activity.relates_to,
67
- )
+ typing_activity = Activity(
+ type=ActivityTypes.typing, relates_to=context.activity.relates_to,
+ )
68
69
- conversation_reference = TurnContext.get_conversation_reference(
70
- context.activity
71
+ conversation_reference = TurnContext.get_conversation_reference(
+ context.activity
72
73
- typing_activity = TurnContext.apply_conversation_reference(
74
- typing_activity, conversation_reference
75
+ typing_activity = TurnContext.apply_conversation_reference(
+ typing_activity, conversation_reference
76
77
- await context.adapter.send_activities(context, [typing_activity])
+ asyncio.ensure_future(
+ context.adapter.send_activities(context, [typing_activity])
78
79
- start_interval(context, period, period)
+ # restart the timer, with the 'period' value for the delay
+ timer.set_timeout(aux, period)
80
81
- await timer.set_timeout(aux, delay)
+ # first time through we use the 'delay' value for the timer.
+ timer.set_timeout(aux, delay)
82
83
def stop_interval():
84
- nonlocal finished
85
- finished = True
86
timer.set_clear_timer()
87
+ # if it's a message, start sending typing activities until the
+ # bot logic is done.
88
if context.activity.type == ActivityTypes.message:
89
90
- await start_interval(context, self._delay, self._period)
+ start_interval(context, self._delay, self._period)
91
+ # call the bot logic
92
result = await logic()
93
stop_interval()
94
95
return result
libraries/botbuilder-core/tests/test_show_typing_middleware.py
@@ -1,68 +1,67 @@
-# Copyright (c) Microsoft Corporation. All rights reserved.
-# Licensed under the MIT License.
-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,
- ):
- await step1.assert_reply("echo:foo")
- async def test_should_not_immediately_respond_with_message(self):
- def assert_is_not_message(
- activity, description
49
- ): # pylint: disable=unused-argument
50
- assert activity.type != ActivityTypes.message
51
52
- await step1.assert_reply(assert_is_not_message)
- async def test_should_immediately_respond_with_message_if_no_middleware(self):
- def assert_is_message(activity, description): # pylint: disable=unused-argument
- assert activity.type == ActivityTypes.message
- await step1.assert_reply(assert_is_message)
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+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):
+ await asyncio.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,
+ ):
+ await step1.assert_reply("echo:foo")
+ async def test_should_not_immediately_respond_with_message(self):
+ def assert_is_not_message(
+ activity, description
+ ): # pylint: disable=unused-argument
+ assert activity.type != ActivityTypes.message
+ await step1.assert_reply(assert_is_not_message)
+ async def test_should_immediately_respond_with_message_if_no_middleware(self):
+ def assert_is_message(activity, description): # pylint: disable=unused-argument
+ assert activity.type == ActivityTypes.message
+ await step1.assert_reply(assert_is_message)