NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

stackless_response_router.py (3743B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\stackless_response_router\stackless_response_router.py
      2 from __future__ import absolute_import
      3 import logging
      4 import sentry_sdk
      5 import time
      6 import uthread2
      7 import uuid
      8 from .exceptions import TimeoutException, UnpackException
      9 DEFAULT_TIMEOUT_SECONDS = 10
     10 MIN_TIMEOUT_SECONDS = 0.1
     11 logger = logging.getLogger(__name__)
     12 
     13 class StacklessResponseRouter(object):
     14 
     15     def __init__(self, span_manager):
     16         self.response_handlers = {}
     17         self.span_manager = span_manager
     18 
     19     def add_response_handler(self, request, expected_response_class, timeout_seconds = None):
     20         if timeout_seconds is None:
     21             timeout_seconds = DEFAULT_TIMEOUT_SECONDS
     22         elif timeout_seconds < MIN_TIMEOUT_SECONDS:
     23             timeout_seconds = MIN_TIMEOUT_SECONDS
     24         rh = ResponseHandler(request, expected_response_class, timeout_seconds)
     25         self.response_handlers[request.correlation_uuid] = rh
     26         return rh.response_channel
     27 
     28     def handle_response(self, response_id_bytes, response):
     29         try:
     30             rh = self.response_handlers.pop(response_id_bytes)
     31             if rh is None:
     32                 return
     33             rh.handle(response)
     34         except KeyError:
     35             return
     36 
     37     def handle_request_timeouts(self):
     38         try:
     39             timed_out_request_ids = []
     40             now = time.time()
     41             for request_id, rh in self.response_handlers.items():
     42                 if now > rh.deadline:
     43                     timed_out_request_ids.append(request_id)
     44 
     45             for request_id_bytes in timed_out_request_ids:
     46                 rh = self.response_handlers.pop(request_id_bytes)
     47                 rh.timeout()
     48                 self.span_manager.timeout_request_trace(request_id_bytes)
     49 
     50         except Exception as e:
     51             logger.error(e)
     52 
     53     @staticmethod
     54     def report_timeout_exception(request_id_bytes, details):
     55         with sentry_sdk.push_scope() as scope:
     56             request_uuid = uuid.UUID(bytes=request_id_bytes)
     57             scope.set_tag('request_uuid', request_uuid)
     58             for key, val in details.items():
     59                 scope.set_tag(key, val)
     60 
     61             sentry_sdk.capture_message('request timed out', 'error')
     62 
     63 
     64 class ResponseHandler(object):
     65     __slots__ = ('request', 'response_channel', 'deadline', 'expected_response_class')
     66 
     67     def __init__(self, request, expected_response_class, timeout_seconds):
     68         self.request = request
     69         self.response_channel = uthread2.queue_channel()
     70         self.deadline = time.time() + timeout_seconds
     71         self.expected_response_class = expected_response_class
     72 
     73     def unpack_failure(self, payload_type_url, exception):
     74         msg = 'Delivered Response Type: {} Expected Response Class: {} Exception: {}'.format(payload_type_url, self.expected_response_class, exception)
     75         self.response_channel.send_exception(UnpackException, msg)
     76         self.response_channel.close()
     77 
     78     def handle(self, response_primitive):
     79         payload = None
     80         if response_primitive.payload.type_url:
     81             try:
     82                 payload = self.expected_response_class()
     83                 unpacked = response_primitive.payload.Unpack(payload)
     84                 if not unpacked:
     85                     self.unpack_failure(response_primitive.payload.type_url, None)
     86                     return
     87             except Exception as e:
     88                 self.unpack_failure(response_primitive.payload.type_url, e)
     89                 return
     90 
     91         self.response_channel.send((response_primitive, payload))
     92         self.response_channel.close()
     93 
     94     def timeout(self):
     95         self.response_channel.send_exception(TimeoutException)
     96         self.response_channel.close()