Simulate interactivity boost for simulated user input events

telemetry/telemetry/internal/util/repeating_timer.py

+# Copyright 2017 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+import threading
+import time
+
+class RepeatingTimer(object):
+  """A simple repeating timer."""
+
+  def __init__(self, target, interval_s):
+    """Create a timer object which will repeatedly run the target function
+    with the specified interval. Interval should be specified in seconds.
+    """
+    self._target = target
+    self._interval = interval_s
+    self._stop = threading.Event()
+    self._stop.set()
+    self._timer_thread = None
+
+  def _BackgroundThread(self):
+    target_time = time.time() + self._interval
+    while not self._stop.wait(max(0, target_time - time.time())):
+      target_time = target_time + self._interval
+      self._target()
+
+  def Start(self):
+    """Start the timer. The target function will be called at regular intervals
+    after the given interval elapses. The timer may be stopped and restarted.
+    This must not be called if the timer is already running (Stop should be
+    called to stop it first).
+    """
+    # start new timer thread
+    self._stop.clear()
+    self._timer_thread = threading.Thread(target=self._BackgroundThread)
+    self._timer_thread.start()
+
+  def Stop(self):
+    """Stop the timer and block until it has stopped.
+    The timer must have been started before calling Stop."""
+    self._stop.set()
+    self._timer_thread.join()
+
+  def IsRunning(self):
+    """Return true if the timer is currently running."""
+    return not self._stop.is_set()