Funktionierender Prototyp des Serious Games zur Vermittlung von Wissen zu Software-Engineering-Arbeitsmodellen.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

_buffer.py 1.5KB

1 year ago
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354
  1. # -*- test-case-name: twisted.logger.test.test_buffer -*-
  2. # Copyright (c) Twisted Matrix Laboratories.
  3. # See LICENSE for details.
  4. """
  5. Log observer that maintains a buffer.
  6. """
  7. from collections import deque
  8. from typing import Deque, Optional
  9. from zope.interface import implementer
  10. from ._interfaces import ILogObserver, LogEvent
  11. _DEFAULT_BUFFER_MAXIMUM = 64 * 1024
  12. @implementer(ILogObserver)
  13. class LimitedHistoryLogObserver:
  14. """
  15. L{ILogObserver} that stores events in a buffer of a fixed size::
  16. >>> from twisted.logger import LimitedHistoryLogObserver
  17. >>> history = LimitedHistoryLogObserver(5)
  18. >>> for n in range(10): history({'n': n})
  19. ...
  20. >>> repeats = []
  21. >>> history.replayTo(repeats.append)
  22. >>> len(repeats)
  23. 5
  24. >>> repeats
  25. [{'n': 5}, {'n': 6}, {'n': 7}, {'n': 8}, {'n': 9}]
  26. >>>
  27. """
  28. def __init__(self, size: Optional[int] = _DEFAULT_BUFFER_MAXIMUM) -> None:
  29. """
  30. @param size: The maximum number of events to buffer. If L{None}, the
  31. buffer is unbounded.
  32. """
  33. self._buffer: Deque[LogEvent] = deque(maxlen=size)
  34. def __call__(self, event: LogEvent) -> None:
  35. self._buffer.append(event)
  36. def replayTo(self, otherObserver: ILogObserver) -> None:
  37. """
  38. Re-play the buffered events to another log observer.
  39. @param otherObserver: An observer to replay events to.
  40. """
  41. for event in self._buffer:
  42. otherObserver(event)