distributor.py 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142
  1. # Copyright 2014-2016 OpenMarket Ltd
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. import logging
  15. from typing import (
  16. Any,
  17. Awaitable,
  18. Callable,
  19. Dict,
  20. Generic,
  21. List,
  22. Optional,
  23. TypeVar,
  24. Union,
  25. )
  26. from typing_extensions import ParamSpec
  27. from twisted.internet import defer
  28. from synapse.logging.context import make_deferred_yieldable, run_in_background
  29. from synapse.metrics.background_process_metrics import run_as_background_process
  30. from synapse.types import UserID
  31. from synapse.util.async_helpers import maybe_awaitable
  32. logger = logging.getLogger(__name__)
  33. def user_left_room(distributor: "Distributor", user: UserID, room_id: str) -> None:
  34. distributor.fire("user_left_room", user=user, room_id=room_id)
  35. class Distributor:
  36. """A central dispatch point for loosely-connected pieces of code to
  37. register, observe, and fire signals.
  38. Signals are named simply by strings.
  39. TODO(paul): It would be nice to give signals stronger object identities,
  40. so we can attach metadata, docstrings, detect typos, etc... But this
  41. model will do for today.
  42. """
  43. def __init__(self) -> None:
  44. self.signals: Dict[str, Signal] = {}
  45. self.pre_registration: Dict[str, List[Callable]] = {}
  46. def declare(self, name: str) -> None:
  47. if name in self.signals:
  48. raise KeyError("%r already has a signal named %s" % (self, name))
  49. self.signals[name] = Signal(name)
  50. if name in self.pre_registration:
  51. signal = self.signals[name]
  52. for observer in self.pre_registration[name]:
  53. signal.observe(observer)
  54. def observe(self, name: str, observer: Callable) -> None:
  55. if name in self.signals:
  56. self.signals[name].observe(observer)
  57. else:
  58. # TODO: Avoid strong ordering dependency by allowing people to
  59. # pre-register observations on signals that don't exist yet.
  60. if name not in self.pre_registration:
  61. self.pre_registration[name] = []
  62. self.pre_registration[name].append(observer)
  63. def fire(self, name: str, *args: Any, **kwargs: Any) -> None:
  64. """Dispatches the given signal to the registered observers.
  65. Runs the observers as a background process. Does not return a deferred.
  66. """
  67. if name not in self.signals:
  68. raise KeyError("%r does not have a signal named %s" % (self, name))
  69. run_as_background_process(name, self.signals[name].fire, *args, **kwargs)
  70. P = ParamSpec("P")
  71. R = TypeVar("R")
  72. class Signal(Generic[P]):
  73. """A Signal is a dispatch point that stores a list of callables as
  74. observers of it.
  75. Signals can be "fired", meaning that every callable observing it is
  76. invoked. Firing a signal does not change its state; it can be fired again
  77. at any later point. Firing a signal passes any arguments from the fire
  78. method into all of the observers.
  79. """
  80. def __init__(self, name: str):
  81. self.name: str = name
  82. self.observers: List[Callable[P, Any]] = []
  83. def observe(self, observer: Callable[P, Any]) -> None:
  84. """Adds a new callable to the observer list which will be invoked by
  85. the 'fire' method.
  86. Each observer callable may return a Deferred."""
  87. self.observers.append(observer)
  88. def fire(self, *args: P.args, **kwargs: P.kwargs) -> "defer.Deferred[List[Any]]":
  89. """Invokes every callable in the observer list, passing in the args and
  90. kwargs. Exceptions thrown by observers are logged but ignored. It is
  91. not an error to fire a signal with no observers.
  92. Returns a Deferred that will complete when all the observers have
  93. completed."""
  94. async def do(observer: Callable[P, Union[R, Awaitable[R]]]) -> Optional[R]:
  95. try:
  96. return await maybe_awaitable(observer(*args, **kwargs))
  97. except Exception as e:
  98. logger.warning(
  99. "%s signal observer %s failed: %r",
  100. self.name,
  101. observer,
  102. e,
  103. )
  104. return None
  105. deferreds = [run_in_background(do, o) for o in self.observers]
  106. return make_deferred_yieldable(
  107. defer.gatherResults(deferreds, consumeErrors=True)
  108. )
  109. def __repr__(self) -> str:
  110. return "<Signal name=%r>" % (self.name,)