First version of async printing example

1 files changed, 121 insertions, 0 deletions

diff --git a/examples/async_printing.py b/examples/async_printing.py
new file mode 100755
index 00000000..65e83bbe
--- /dev/null
+++ b/examples/async_printing.py
@@ -0,0 +1,121 @@
+#!/usr/bin/env python
+# coding=utf-8
+"""A simple example demonstrating an application that asynchronously prints alerts and updates the prompt"""
+
+import random
+import threading
+import time
+from typing import List
+
+import cmd2
+
+ALERTS = ["Mail server is down",
+          "Blockage detected in sector 4",
+          "Ants have overrun the break room",
+          "Your next appointment has arrived",
+          "Christmas bonuses are cancelled",
+          "Mandatory overtime this weekend",
+          "Jimmy quit",
+          "Jody got married",
+          "Is there anyone on board who knows how to fly a plane?",
+          "Gentlemen, you can't fight in here. This is the War Room!",
+          "Your mom goes to college!"]
+
+
+def get_alerts() -> List[str]:
+    """ Randomly generates alerts for testing purposes """
+
+    rand_num = random.randint(1, 20)
+    if rand_num > 3:
+        return []
+
+    alerts = []
+    for i in range(0, rand_num):
+        cur_alert = random.choice(ALERTS)
+        if cur_alert in alerts:
+            i -= 1
+        else:
+            alerts.append(cur_alert)
+
+    return alerts
+
+
+def get_alert_str() -> str:
+    """
+    Combines alerts into one string that can be printed to the terminal
+    :return: the alert string
+    """
+    alert_str = ''
+    alerts = get_alerts()
+
+    longest_alert = max(ALERTS, key=len)
+    num_astericks = len(longest_alert) + 8
+
+    for i, cur_alert in enumerate(alerts):
+        # Use padding to center the alert
+        padding = ' ' * int((num_astericks - len(cur_alert)) / 2)
+
+        if i > 0:
+            alert_str += '\n'
+        alert_str += '*' * num_astericks + '\n'
+        alert_str += padding + cur_alert + padding + '\n'
+        alert_str += '*' * num_astericks + '\n'
+
+    return alert_str
+
+
+class AlerterApp(cmd2.Cmd):
+    """ An app that shows off async_alert() and async_update_prompt() """
+
+    def __init__(self, *args, **kwargs) -> None:
+        """ Initializer """
+
+        super().__init__(*args, **kwargs)
+
+        self.prompt = "(APR)> "
+
+        # The thread that will asynchronously alert the user of events
+        self._stop_thread = False
+        self._alerter_thread = threading.Thread(name='alerter', target=self._alerter_thread_func)
+
+        # Create some hooks to handle the starting and stopping of our thread
+        self.register_preloop_hook(self._preloop_hook)
+        self.register_postloop_hook(self._postloop_hook)
+
+    def _preloop_hook(self) -> None:
+        """ Start the alerter thread """
+
+        # This function runs after cmdloop() locks _terminal_lock, which will be locked until the prompt appears.
+        # Therefore it is safe to start our thread since there is no risk of it alerting before the prompt is displayed.
+        self._stop_thread = False
+        self._alerter_thread.start()
+
+    def _postloop_hook(self) -> None:
+        """ Stops the alerter thread """
+        self._stop_thread = True
+        if self._alerter_thread.is_alive():
+            self._alerter_thread.join()
+
+    def _alerter_thread_func(self) -> None:
+        """ Prints alerts and updates the prompt any time the prompt is showing """
+
+        while not self._stop_thread:
+            # Always acquire _terminal_lock before printing alerts or updating the prompt
+            # To keep the app responsive, do not block on this call
+            if self._terminal_lock.acquire(blocking=False):
+
+                # We have the terminal lock. See if any alerts need to be printed.
+                alert_str = get_alert_str()
+
+                if alert_str:
+                    self.async_alert(alert_str)
+
+                # Don't forget to release the lock
+                self._terminal_lock.release()
+
+            time.sleep(0.5)
+
+
+if __name__ == '__main__':
+    app = AlerterApp()
+    app.cmdloop()