openstack/oslo.messaging
Code Issues Proposed changes

Add restart() method to DecayingTimer

Browse Source 
This merely provides a restart() method that passes through to the existing
restart() method on the StopWatch used as the internal for DecayingTimer so
that we can reset it.

Change-Id: Ie6b607ec588db94e2c768bd22ae736a05ab484c1
This commit is contained in:
Dan Smith 2018-03-03 14:00:35 -08:00
parent 6bdae93658
commit 68c48ad0bb
2 changed files with 17 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
oslo_messaging/_drivers/common.py
View File

@ -347,6 +347,9 @@ class DecayingTimer(object):
def start(self): def start(self):
self._watch.start() self._watch.start()
def restart(self):
self._watch.restart()
def check_return(self, timeout_callback=None, *args, **kwargs): def check_return(self, timeout_callback=None, *args, **kwargs):
maximum = kwargs.pop('maximum', None) maximum = kwargs.pop('maximum', None)
left = self._watch.leftover(return_none=True) left = self._watch.leftover(return_none=True)

14
oslo_messaging/tests/test_utils.py
View File

@ -97,3 +97,17 @@ class TimerTestCase(test_utils.BaseTestCase):
remaining = t.check_return(callback, 1, a='b') remaining = t.check_return(callback, 1, a='b')
self.assertEqual(0, remaining) self.assertEqual(0, remaining)
callback.assert_called_once_with(1, a='b') callback.assert_called_once_with(1, a='b')
@mock.patch('oslo_utils.timeutils.now')
def test_reset(self, now):
now.return_value = 0
t = common.DecayingTimer(3)
t.start()
now.return_value = 1
remaining = t.check_return()
self.assertEqual(2, remaining)
t.restart()
remaining = t.check_return()
self.assertEqual(3, remaining)