gateway.py (6114B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\publicGateway\gateway.py 2 import blue 3 from eve.common.script.util.retry import retry_with_exponential_backoff, STATUS_CODES_THAT_TRIGGER_RETRIES 4 from logging import getLogger 5 from stackless_response_router.exceptions import TimeoutException 6 from uthread2 import Sleep 7 logger = getLogger('PublicGatewaySvc') 8 9 class PublicGateway(object): 10 11 def __init__(self, grpc_module, connection_config, event_publisher, notice_signal_registry, notice_consumer, requests_broker): 12 self.notice_signal_registry = notice_signal_registry 13 self.grpc_module = grpc_module 14 self.connection_config = connection_config 15 self.grpc_event_publisher = event_publisher 16 self.grpc_notice_consumer = notice_consumer 17 self.grpc_requests_broker = requests_broker 18 self.instantiationTime = blue.os.GetWallclockTimeNow() 19 20 def get_uptime_ms(self): 21 return blue.os.TimeDiffInMs(self.instantiationTime, blue.os.GetWallclockTimeNow()) 22 23 def get_event_publisher_rtt(self): 24 if self.grpc_event_publisher is None: 25 return -1 26 return self.grpc_event_publisher.get_rtt() 27 28 def get_event_publisher_details(self): 29 if self.grpc_event_publisher is None: 30 return 31 return self.grpc_event_publisher.get_details() 32 33 def publish_event(self, event_payload, journey_id = None): 34 self.grpc_event_publisher.publish(event_payload, journey_id=journey_id) 35 36 def publish_event_payload(self, event_payload, journey_id = None): 37 self.grpc_event_publisher.publish_event_payload(event_payload, journey_id=journey_id) 38 39 def get_notice_consumer_details(self): 40 if self.grpc_notice_consumer is None: 41 return 42 return self.grpc_notice_consumer.get_details() 43 44 def get_request_broker_rtt(self): 45 if self.grpc_requests_broker is None: 46 return -1 47 return self.grpc_requests_broker.get_rtt() 48 49 def get_request_broker_details(self): 50 if self.grpc_requests_broker is None: 51 return 52 return self.grpc_requests_broker.get_details() 53 54 def send_user_request(self, request_payload, expected_response_class, timeout_seconds): 55 request_primitive, response_channel = self.grpc_requests_broker.send_user_request(request_payload, expected_response_class, timeout_seconds) 56 logger.info('Request %s sent for user', self._get_request_id(request_payload)) 57 return (request_primitive, response_channel) 58 59 def send_character_request(self, request_payload, expected_response_class, timeout_seconds): 60 request_primitive, response_channel = self.grpc_requests_broker.send_character_request(request_payload, expected_response_class, timeout_seconds) 61 logger.info('Request %s sent for character', self._get_request_id(request_payload)) 62 return (request_primitive, response_channel) 63 64 def send_blocking_user_request_and_receive_response(self, request_payload, expected_response_class, timeout_seconds, max_attempts, retry_delay_in_seconds): 65 sender = self.grpc_requests_broker.send_user_request 66 response_primitive, response_payload = self._retry_on_timeout(max_attempts, sender, request_payload, expected_response_class, timeout_seconds, retry_delay_in_seconds) 67 logger.info('Request %s sent for user', self._get_request_id(request_payload)) 68 return (response_primitive, response_payload) 69 70 def send_blocking_character_request_and_receive_response(self, request_payload, expected_response_class, timeout_seconds, max_attempts, retry_delay_in_seconds): 71 sender = self.grpc_requests_broker.send_character_request 72 response_primitive, response_payload = self._retry_on_timeout(max_attempts, sender, request_payload, expected_response_class, timeout_seconds, retry_delay_in_seconds) 73 logger.info('Request %s sent for character', self._get_request_id(request_payload)) 74 return (response_primitive, response_payload) 75 76 def set_active_character_id(self, character_id): 77 self.grpc_event_publisher.set_active_character_id(character_id) 78 self.grpc_requests_broker.set_active_character_id(character_id) 79 80 def set_authenticated_user_id(self, user_id): 81 self.grpc_event_publisher.set_authenticated_user_id(user_id) 82 self.grpc_requests_broker.set_authenticated_user_id(user_id) 83 84 def set_auth_token(self, auth_token): 85 if not auth_token: 86 return 87 self.grpc_module.set_auth_token(auth_token) 88 89 def connect(self): 90 self.connection_config.connect() 91 92 def subscribe_to_notice(self, notice_payload, callback): 93 self.notice_signal_registry.subscribe_to_notice(notice_payload, callback) 94 95 def session_changed(self, change): 96 self.grpc_notice_consumer.session_changed(change) 97 98 def _retry_on_timeout(self, max_attempts, sender, request_payload, expected_response_class, timeout_seconds, retry_delay_in_seconds): 99 100 def send_request(): 101 request_primitive, response_channel = sender(request_payload, expected_response_class, timeout_seconds) 102 response_primitive, response_payload = response_channel.receive() 103 return (response_primitive, response_payload) 104 105 def should_retry(result_or_exception): 106 if isinstance(result_or_exception, TimeoutException): 107 return True 108 if isinstance(result_or_exception, tuple): 109 response_primitive, response_payload = result_or_exception 110 if response_primitive.status_code in STATUS_CODES_THAT_TRIGGER_RETRIES: 111 return True 112 return False 113 114 return retry_with_exponential_backoff(func=send_request, max_attempts=max_attempts, retry_delay_in_seconds=retry_delay_in_seconds, jitter=True, should_retry=should_retry, waiter=self._wait) 115 116 def _get_request_id(self, request_payload): 117 return '{module}.{name}'.format(module=request_payload.__module__, name=type(request_payload).__name__) 118 119 def _wait(self, delay): 120 Sleep(delay)