pulser: Pulse when threads are started

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>

ui/pulser.py

@@ -1,8 +1,44 @@
 # Copyright 2021 (c) Anna Schumaker.
-from gi.repository import Gtk
+from gi.repository import Gtk, GLib
+import lib
+import threading
 
 Bar = Gtk.ProgressBar()
+Threads = [ ]
+Lock = threading.Lock()
+Timeout = None
 
 Bar.set_valign(Gtk.Align.CENTER)
 Bar.set_margin_start(10)
 Bar.set_margin_end(10)
+
+def do_pulses():
+    global Timeout
+    Bar.pulse()
+    with Lock:
+        if not Threads[0].running():
+            Threads.pop(0)
+
+        if len(Threads) == 0:
+            Timeout = None
+            return GLib.SOURCE_REMOVE
+        return GLib.SOURCE_CONTINUE
+
+def start_pulsing(thread):
+    global Timeout
+    with Lock:
+        Threads.append(thread)
+        if Timeout == None:
+            Timeout = GLib.timeout_add(100, do_pulses)
+
+def initialize():
+    lib.thread.Start.register(start_pulsing)
+initialize()
+
+def reset():
+    global Timeout
+    with Lock:
+        Threads.clear()
+        if Timeout != None:
+            GLib.source_remove(Timeout)
+            Timeout = None

ui/test_pulser.py

@@ -1,12 +1,49 @@
 # Copyright 2021 (c) Anna Schumaker.
 from . import pulser
-from gi.repository import Gtk
+from gi.repository import Gtk, GLib
+import lib
+import threading
 import unittest
 
+class TestThread:
+    def __init__(self): pass
+    def running(self): return False
+
+
 class TestUIPulser(unittest.TestCase):
+    def setUp(self):
+        pulser.initialize()
+        pulser.reset()
+
     def test_pulser_init(self):
         self.assertIsInstance(pulser.Bar, Gtk.ProgressBar)
+        self.assertIsInstance(pulser.Lock, type(threading.Lock()))
+        self.assertIsNone(pulser.Timeout)
 
+        self.assertEqual(pulser.Threads, [ ])
         self.assertEqual(pulser.Bar.get_valign(), Gtk.Align.CENTER)
         self.assertEqual(pulser.Bar.get_margin_start(), 10)
         self.assertEqual(pulser.Bar.get_margin_end(), 10)
+
+        self.assertIn(pulser.start_pulsing, lib.thread.Start.subscribers)
+
+    def test_pulser_pulse(self):
+        t1 = TestThread()
+        t2 = TestThread()
+
+        pulser.start_pulsing(t1)
+        self.assertEqual(pulser.Threads, [ t1 ])
+        tid = pulser.Timeout
+        self.assertIsNotNone(tid)
+
+        pulser.start_pulsing(t2)
+        self.assertEqual(pulser.Threads, [ t1, t2 ])
+        self.assertEqual(pulser.Timeout, tid)
+
+        pulser.do_pulses()
+        self.assertEqual(pulser.Threads, [ t2 ])
+        pulser.do_pulses()
+        self.assertEqual(pulser.Threads, [ ])
+        self.assertIsNone(pulser.Timeout)
+
+        GLib.source_remove(tid)