__init__.py 3.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2014-2016 OpenMarket Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. from synapse.api.errors import SynapseError
  16. from synapse.util.logcontext import PreserveLoggingContext
  17. from twisted.internet import defer, reactor, task
  18. import time
  19. import logging
  20. logger = logging.getLogger(__name__)
  21. def unwrapFirstError(failure):
  22. # defer.gatherResults and DeferredLists wrap failures.
  23. failure.trap(defer.FirstError)
  24. return failure.value.subFailure
  25. class Clock(object):
  26. """A small utility that obtains current time-of-day so that time may be
  27. mocked during unit-tests.
  28. TODO(paul): Also move the sleep() functionallity into it
  29. """
  30. def time(self):
  31. """Returns the current system time in seconds since epoch."""
  32. return time.time()
  33. def time_msec(self):
  34. """Returns the current system time in miliseconds since epoch."""
  35. return int(self.time() * 1000)
  36. def looping_call(self, f, msec):
  37. l = task.LoopingCall(f)
  38. l.start(msec / 1000.0, now=False)
  39. return l
  40. def call_later(self, delay, callback, *args, **kwargs):
  41. """Call something later
  42. Args:
  43. delay(float): How long to wait in seconds.
  44. callback(function): Function to call
  45. *args: Postional arguments to pass to function.
  46. **kwargs: Key arguments to pass to function.
  47. """
  48. def wrapped_callback(*args, **kwargs):
  49. with PreserveLoggingContext():
  50. callback(*args, **kwargs)
  51. with PreserveLoggingContext():
  52. return reactor.callLater(delay, wrapped_callback, *args, **kwargs)
  53. def cancel_call_later(self, timer, ignore_errs=False):
  54. try:
  55. timer.cancel()
  56. except:
  57. if not ignore_errs:
  58. raise
  59. def time_bound_deferred(self, given_deferred, time_out):
  60. if given_deferred.called:
  61. return given_deferred
  62. ret_deferred = defer.Deferred()
  63. def timed_out_fn():
  64. try:
  65. ret_deferred.errback(SynapseError(504, "Timed out"))
  66. except:
  67. pass
  68. try:
  69. given_deferred.cancel()
  70. except:
  71. pass
  72. timer = None
  73. def cancel(res):
  74. try:
  75. self.cancel_call_later(timer)
  76. except:
  77. pass
  78. return res
  79. ret_deferred.addBoth(cancel)
  80. def sucess(res):
  81. try:
  82. ret_deferred.callback(res)
  83. except:
  84. pass
  85. return res
  86. def err(res):
  87. try:
  88. ret_deferred.errback(res)
  89. except:
  90. pass
  91. given_deferred.addCallbacks(callback=sucess, errback=err)
  92. timer = self.call_later(time_out, timed_out_fn)
  93. return ret_deferred