Class Event

src/debugpy/common/messaging.py:529–601

Represents an incoming event. https://microsoft.github.io/debug-adapter-protocol/specification#event It is guaranteed that body is a MessageDict associated with this Event, and so are all the nested dicts in it. If "body" was missing or null in JSON, body is an empty dict. To

class Event(Message):
    """Represents an incoming event.

    https://microsoft.github.io/debug-adapter-protocol/specification#event

    It is guaranteed that body is a MessageDict associated with this Event, and so
    are all the nested dicts in it. If "body" was missing or null in JSON, body is
    an empty dict.

    To handle the event, JsonMessageChannel tries to find a handler for this event in
    JsonMessageChannel.handlers. Given event="X", if handlers.X_event exists, then it
    is the specific handler for this event. Otherwise, handlers.event must exist, and
    it is the generic handler for this event. A missing handler is a fatal error.

    No further incoming messages are processed until the handler returns, except for
    responses to requests that have wait_for_response() invoked on them.

    To report failure to handle the event, the handler must raise an instance of
    MessageHandlingError that applies_to() the Event object it was handling. Any such
    failure is logged, after which the message loop moves on to the next message.

    Helper methods Message.isnt_valid() and Message.cant_handle() can be used to raise
    the appropriate exception type that applies_to() the Event object.
    """

    def __init__(self, channel, seq, event, body, json=None):
        super().__init__(channel, seq, json)

        self.event = event

        if isinstance(body, MessageDict) and hasattr(body, "associate_with"):
            body.associate_with(self)
        self.body = body

    def describe(self):
        return f"#{self.seq} event {json.repr(self.event)} from {self.channel}"

    @property
    def payload(self):
        return self.body

    @staticmethod
    def _parse(channel, message_dict):
        seq = message_dict("seq", int)
        event = message_dict("event", str)
        body = message_dict("body", _payload)
        message = Event(channel, seq, event, body, json=message_dict)
        channel._enqueue_handlers(message, message._handle)

    def _handle(self):
        channel = self.channel
        handler = channel._get_handler_for("event", self.event)
        try:
            try:
                result = handler(self)
                assert (
                    result is None
                ), f"Handler {util.srcnameof(handler)} tried to respond to {self.describe()}."