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.

_baseprocess.py 2.0KB

1 year ago
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768
  1. # -*- test-case-name: twisted.test.test_process -*-
  2. # Copyright (c) Twisted Matrix Laboratories.
  3. # See LICENSE for details.
  4. """
  5. Cross-platform process-related functionality used by different
  6. L{IReactorProcess} implementations.
  7. """
  8. from typing import Optional
  9. from twisted.python.deprecate import getWarningMethod
  10. from twisted.python.failure import Failure
  11. from twisted.python.log import err
  12. from twisted.python.reflect import qual
  13. _missingProcessExited = (
  14. "Since Twisted 8.2, IProcessProtocol.processExited "
  15. "is required. %s must implement it."
  16. )
  17. class BaseProcess:
  18. pid: Optional[int] = None
  19. status: Optional[int] = None
  20. lostProcess = 0
  21. proto = None
  22. def __init__(self, protocol):
  23. self.proto = protocol
  24. def _callProcessExited(self, reason):
  25. default = object()
  26. processExited = getattr(self.proto, "processExited", default)
  27. if processExited is default:
  28. getWarningMethod()(
  29. _missingProcessExited % (qual(self.proto.__class__),),
  30. DeprecationWarning,
  31. stacklevel=0,
  32. )
  33. else:
  34. try:
  35. processExited(Failure(reason))
  36. except BaseException:
  37. err(None, "unexpected error in processExited")
  38. def processEnded(self, status):
  39. """
  40. This is called when the child terminates.
  41. """
  42. self.status = status
  43. self.lostProcess += 1
  44. self.pid = None
  45. self._callProcessExited(self._getReason(status))
  46. self.maybeCallProcessEnded()
  47. def maybeCallProcessEnded(self):
  48. """
  49. Call processEnded on protocol after final cleanup.
  50. """
  51. if self.proto is not None:
  52. reason = self._getReason(self.status)
  53. proto = self.proto
  54. self.proto = None
  55. try:
  56. proto.processEnded(Failure(reason))
  57. except BaseException:
  58. err(None, "unexpected error in processEnded")