model/boxes: Add functionality to send typing event to server for PMs.

tests/model/test_model.py

@@ -434,6 +434,32 @@ def test_react_to_message_for_not_thumbs_up(self, model):
         with pytest.raises(AssertionError):
             model.react_to_message(dict(), 'x')
 
+    @pytest.mark.parametrize('recipient_user_ids', [[5140], [5140, 5179]])
+    @pytest.mark.parametrize('status', ['start', 'stop'])
+    def test_send_typing_status_by_user_ids(self, mocker, model, status,
+                                            recipient_user_ids):
+        response = mocker.Mock()
+        mock_api_query = mocker.patch('zulipterminal.core.Controller'
+                                      '.client.set_typing_status',
+                                      return_value=response)
+
+        model.send_typing_status_by_user_ids(recipient_user_ids,
+                                             status=status)
+
+        mock_api_query.assert_called_once_with(
+            {'to': recipient_user_ids, 'op': status},
+        )
+
+        self.display_error_if_present.assert_called_once_with(response,
+                                                              self.controller)
+
+    @pytest.mark.parametrize('status', ['start', 'stop'])
+    def test_send_typing_status_with_no_recipients(self, model, status,
+                                                   recipient_user_ids=[]):
+        with pytest.raises(RuntimeError):
+            model.send_typing_status_by_user_ids(recipient_user_ids,
+                                                 status=status)
+
     @pytest.mark.parametrize('response, return_value', [
         ({'result': 'success'}, True),
         ({'result': 'some_failure'}, False),

tests/ui_tools/test_boxes.py

@@ -32,6 +32,8 @@ def write_box(self, mocker, users_fixture, user_groups_fixture,
             {'name': stream['name']} for stream in
             streams_fixture], key=lambda stream: stream['name'].lower())
 
+        write_box.to_write_box = None
+
         return write_box
 
     def test_init(self, write_box):

zulipterminal/model.py

@@ -333,6 +333,20 @@ def mark_message_ids_as_read(self, id_list: List[int]) -> None:
         })
         display_error_if_present(response, self.controller)
 
+    @asynch
+    def send_typing_status_by_user_ids(self, recipient_user_ids: List[int],
+                                       *, status: Literal['start', 'stop']
+                                       ) -> None:
+        if recipient_user_ids:
+            request = {
+                'to': recipient_user_ids,
+                'op': status
+            }
+            response = self.client.set_typing_status(request)
+            display_error_if_present(response, self.controller)
+        else:
+            raise RuntimeError('Empty recipient list.')
+
     def send_private_message(self, recipients: List[str],
                              content: str) -> bool:
         if recipients:

zulipterminal/ui_tools/boxes.py

@@ -1,9 +1,9 @@
 import re
 import unicodedata
 from collections import OrderedDict, defaultdict
-from datetime import date, datetime
+from datetime import date, datetime, timedelta
 from sys import platform
-from time import ctime, time
+from time import ctime, sleep, time
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 from urllib.parse import urljoin, urlparse
 
@@ -23,7 +23,7 @@
     STREAM_TOPIC_SEPARATOR, TIME_MENTION_MARKER,
 )
 from zulipterminal.helper import (
-    Message, format_string, get_unused_fence, match_emoji, match_group,
+    Message, asynch, format_string, get_unused_fence, match_emoji, match_group,
     match_stream, match_topics, match_user,
 )
 from zulipterminal.server_url import near_message_url
@@ -42,6 +42,9 @@ def __init__(self, view: Any) -> None:
         self.stream_id = None  # type: Optional[int]
         self.recipient_user_ids = []  # type: List[int]
         self.msg_body_edit_enabled = True
+        self.send_next_typing_update = datetime.now()
+        self.last_key_update = datetime.now()
+        self.idle_status_tracking = False
         self.FOCUS_CONTAINER_HEADER = 0
         self.FOCUS_HEADER_BOX_RECIPIENT = 0
         self.FOCUS_HEADER_BOX_STREAM = 1
@@ -63,13 +66,24 @@ def main_view(self, new: bool) -> Any:
     def set_editor_mode(self) -> None:
         self.view.controller.enter_editor_mode_with(self)
 
+    def send_stop_typing_status(self) -> None:
+        # Send 'stop' updates only for PM narrows.
+        if self.to_write_box:
+            self.model.send_typing_status_by_user_ids(
+                self.recipient_user_ids,
+                status='stop'
+            )
+            self.send_next_typing_update = datetime.now()
+            self.idle_status_tracking = False
+
     def private_box_view(self, button: Any=None, email: str='',
                          recipient_user_ids: Optional[List[int]]=None) -> None:
         self.set_editor_mode()
         if recipient_user_ids:
             self.recipient_user_ids = recipient_user_ids
         if email == '' and button is not None:
             email = button.email
+        self.send_next_typing_update = datetime.now()
         self.to_write_box = ReadlineEdit("To: ", edit_text=email)
         self.msg_write_box = ReadlineEdit(multiline=True)
         self.msg_write_box.enable_autocomplete(
@@ -89,6 +103,40 @@ def private_box_view(self, button: Any=None, email: str='',
         ]
         self.focus_position = self.FOCUS_CONTAINER_MESSAGE
 
+        # Typing status is sent in regular intervals to limit the number of
+        # notifications sent. Idleness should also prompt a notification.
+        # Refer to https://zulip.com/api/set-typing-status for the protocol
+        # on typing notifications sent by clients.
+        TYPING_STARTED_WAIT_PERIOD = 10
+        TYPING_STOPPED_WAIT_PERIOD = 5
+
+        start_period_delta = timedelta(seconds=TYPING_STARTED_WAIT_PERIOD)
+        stop_period_delta = timedelta(seconds=TYPING_STOPPED_WAIT_PERIOD)
+
+        def on_type_send_status(edit: object, new_edit_text: str) -> None:
+            if new_edit_text:
+                self.last_key_update = datetime.now()
+                if self.last_key_update > self.send_next_typing_update:
+                    self.model.send_typing_status_by_user_ids(
+                                self.recipient_user_ids, status='start')
+                    self.send_next_typing_update += start_period_delta
+                    # Initiate tracker function only if it isn't already
+                    # initiated.
+                    if not self.idle_status_tracking:
+                        self.idle_status_tracking = True
+                        track_idleness_and_update_status()
+
+        @asynch
+        def track_idleness_and_update_status() -> None:
+            while datetime.now() < self.last_key_update + stop_period_delta:
+                idle_check_time = (self.last_key_update
+                                   + stop_period_delta
+                                   - datetime.now())
+                sleep(idle_check_time.total_seconds())
+            self.send_stop_typing_status()
+
+        urwid.connect_signal(self.msg_write_box, 'change', on_type_send_status)
+
     def stream_box_view(self, stream_id: int, caption: str='', title: str='',
                         ) -> None:
         self.set_editor_mode()
@@ -352,6 +400,7 @@ def keypress(self, size: urwid_Size, key: str) -> Optional[str]:
                 self.view.set_footer_text()
 
         if is_command_key('SEND_MESSAGE', key):
+            self.send_stop_typing_status()
             if not self.to_write_box:
                 if re.fullmatch(r'\s*', self.title_write_box.edit_text):
                     topic = '(no topic)'
@@ -393,6 +442,7 @@ def keypress(self, size: urwid_Size, key: str) -> Optional[str]:
         elif is_command_key('GO_BACK', key):
             self.msg_edit_id = None
             self.msg_body_edit_enabled = True
+            self.send_stop_typing_status()
             self.view.controller.exit_editor_mode()
             self.main_view(False)
             self.view.middle_column.set_focus('body')

zulipterminal/ui_tools/buttons.py

@@ -233,7 +233,10 @@ def _narrow_with_compose(self, button: Any) -> None:
         # FIXME should we just narrow?
         self.controller.narrow_to_user(self)
         self._view.body.focus.original_widget.set_focus('footer')
-        self._view.write_box.private_box_view(self)
+        self._view.write_box.private_box_view(
+            self,
+            recipient_user_ids=[self.user_id]
+        )
 
 
 class TopicButton(TopButton):