Merge pull request #4 from PraneethChandraThota/retry-logic

david-schrenker · web-flow · commit cd314ad952a8 · 2021-04-13T15:48:59.000-04:00
LU-49 Python Retry Logic
diff --git a/python-examples/basic/basic_send_with_retry.py b/python-examples/basic/basic_send_with_retry.py
@@ -0,0 +1,55 @@
+import json
+import os
+
+from socketlabs.injectionapi import SocketLabsClient
+from socketlabs.injectionapi.proxy import Proxy
+from socketlabs.injectionapi.message.__imports__ import BasicMessage, EmailAddress
+
+def on_success(response):
+    """
+    Handle the success response from the client
+    :param response: the SendResponse
+    :return: SendResponse
+    """
+    print(json.dumps(response.to_json(), indent=2))
+
+
+def on_error(exception):
+    """
+    Handle the error response from the client
+    :param exception: the Exception
+    :return: Exception
+    """
+    print(exception) 
+
+# build the message
+message = BasicMessage()
+
+message.subject = "Sending An Email Through A Proxy"
+message.html_body = "<html><body>" \
+                    "<h1>Sending An Email Through A Proxy</h1>" \
+                    "<p>This is the Html Body of my message.</p>" \
+                    "</body></html>"
+message.plain_text_body = "This is the Plain Text Body of my message."
+
+message.from_email_address = EmailAddress("from@example.com")
+message.add_to_email_address("recipient@example.com")
+
+
+# get credentials from environment variables
+server_id = int(os.environ.get('SOCKETLABS_SERVER_ID'))
+api_key = os.environ.get('SOCKETLABS_INJECTION_API_KEY')
+
+# create the proxy
+proxy = Proxy("127.0.0.1", 4433)
+
+# create the client
+client = SocketLabsClient(server_id, api_key, proxy)
+
+client.number_of_retries = 2
+
+# send the message
+response = client.send(message)
+client.send_async(message, on_success, on_error)
+
+print(json.dumps(response.to_json(), indent=2))
diff --git a/socketlabs/injectionapi/core/httprequest.py b/socketlabs/injectionapi/core/httprequest.py
@@ -106,26 +106,21 @@ def send_async_request(self, request: InjectionRequest, on_success_callback, on_
         :param on_error_callback: the callback method for error
         :type on_error_callback: method
         """
-        req_queue = queue.Queue()
+
         try:
 
             th = threading.Thread(target=self.__queue_request,
                                   kwargs={
                                       "request": request,
-                                      "out_queue": req_queue
+                                      "on_success_callback": on_success_callback,
+                                      "on_error_callback": on_error_callback
                                   })
             th.start()
-            th.join()
-
-            while not th.is_alive():
-                response = req_queue.get()
-                on_success_callback(response)
-                break
-
+            
         except Exception as e:
             on_error_callback(e)
 
-    def __queue_request(self, request: InjectionRequest, out_queue):
+    def __queue_request(self, request: InjectionRequest, on_success_callback, on_error_callback):
         """
         queue method for the threaded send request.
         :param request: the injection request to send
@@ -135,9 +130,10 @@ def __queue_request(self, request: InjectionRequest, out_queue):
         """
         try:
             response = self.send_request(request)
-            out_queue.put(response)
+            on_success_callback(response)
+
         except Exception:
-            raise
+            on_error_callback(sys.exc_info()[0])
 
     def send_request(self, request: InjectionRequest):
         """
@@ -155,15 +151,15 @@ def send_request(self, request: InjectionRequest):
 
         json_body = json.dumps(request.to_json())
 
-        connection = self.__get_connection()
-        connection.request("POST", self._endpoint.url, json_body, headers)
-        response = connection.getresponse()
+        try:
+            connection = self.__get_connection()
+            connection.request("POST", self._endpoint.url, json_body, headers)
+            response = connection.getresponse()
 
-        data = response.read().decode("utf-8")
-        response_code = response.status
+        except Exception as e:
+            raise e
 
-        result = InjectionResponseParser.parse(data, response_code)
-        return result
+        return response
 
     def __get_connection(self):
         """
diff --git a/socketlabs/injectionapi/core/retryhandler.py b/socketlabs/injectionapi/core/retryhandler.py
@@ -0,0 +1,89 @@
+from http import HTTPStatus
+from http.client import HTTPException
+from ..retrysettings import RetrySettings
+from .httprequest import HttpRequest
+from .serialization.injectionrequest import InjectionRequest
+import socket
+import time
+
+
+class RetryHandler(object):
+    attempts = 0
+    ErrorStatusCodes = [
+        HTTPStatus.INTERNAL_SERVER_ERROR,
+        HTTPStatus.BAD_GATEWAY,
+        HTTPStatus.SERVICE_UNAVAILABLE,
+        HTTPStatus.GATEWAY_TIMEOUT
+    ]
+    Exceptions = [
+        socket.timeout,
+        HTTPException
+    ]
+
+    def __init__(self, http_client: HttpRequest, settings: RetrySettings):
+
+        self.__http_client = http_client
+        self.__retry_settings = settings
+
+    def send(self, body):
+
+        if self.__retry_settings.maximum_number_of_retries == 0:
+            return self.__http_client.send_request(body)
+
+        while True:
+            wait_interval = self.__retry_settings.get_next_wait_interval(self.attempts)
+
+            try:
+
+                response = self.__http_client.send_request(body)
+
+                if response.status in self.ErrorStatusCodes:
+                    raise HTTPException("HttpStatusCode: {0}. Response contains server error.".format(response.status))
+
+                return response
+
+            except socket.timeout:
+
+                self.attempts += 1
+
+                if self.attempts > self.__retry_settings.maximum_number_of_retries:
+                    raise socket.timeout
+                time.sleep(wait_interval.seconds)
+
+            except HTTPException:
+
+                self.attempts += 1
+
+                if self.attempts > self.__retry_settings.maximum_number_of_retries:
+                    raise HTTPException
+                time.sleep(wait_interval.seconds)
+
+    def send_async(self, request: InjectionRequest, on_success_callback, on_error_callback):
+
+        wait_interval = self.__retry_settings.get_next_wait_interval(self.attempts)
+
+        def on_success(response):
+
+            if response.status in self.ErrorStatusCodes and self.attempts < self.__retry_settings.maximum_number_of_retries:
+
+                self.attempts += 1
+                time.sleep(wait_interval.seconds)
+                self.send_async(request, on_success_callback, on_error_callback)
+
+            else:
+                on_success_callback(response)
+
+        def on_error(exception):
+
+            if exception in self.Exceptions and self.attempts < self.__retry_settings.maximum_number_of_retries:
+
+                self.attempts += 1
+                time.sleep(wait_interval.seconds)
+                self.send_async(request, on_success_callback, on_error_callback)
+
+            else:
+
+                self.attempts = self.__retry_settings.maximum_number_of_retries + 1
+                on_error_callback(exception)
+
+        self.__http_client.send_async_request(request, on_success, on_error)
diff --git a/socketlabs/injectionapi/retrysettings.py b/socketlabs/injectionapi/retrysettings.py
@@ -0,0 +1,46 @@
+from datetime import timedelta
+import random
+
+
+class RetrySettings(object):
+    __default_number_of_retries = 0
+    __maximum_allowed_number_of_retries = 5
+    __minimum_retry_time = timedelta(seconds=1)
+    __maximum_retry_time = timedelta(seconds=10)
+
+    def __init__(self, maximum_retries=None):
+
+        if maximum_retries:
+
+            if maximum_retries < 0:
+                raise AttributeError("maximumNumberOfRetries must be greater than 0")
+
+            if maximum_retries > self.__maximum_allowed_number_of_retries:
+                raise AttributeError("The maximum number of allowed retries is ",
+                                     self.__maximum_allowed_number_of_retries)
+
+            self.__maximum_number_of_retries = maximum_retries
+
+        else:
+            self.__maximum_number_of_retries = self.__default_number_of_retries
+
+    @property
+    def maximum_number_of_retries(self):
+        return self.__maximum_number_of_retries
+
+    def get_next_wait_interval(self, number_of_attempts):
+
+        interval = int(min(
+            ((self.__minimum_retry_time.seconds * 1000) + self.get_retry_delta(number_of_attempts)),
+            (self.__maximum_retry_time.seconds * 1000)
+        ))
+
+        return timedelta(milliseconds=interval)
+
+    @staticmethod
+    def get_retry_delta(number_of_attempts):
+
+        minimum = int((timedelta(seconds=1).seconds * 1000) * 0.8)
+        maximum = int((timedelta(seconds=1).seconds * 1000) * 1.2)
+
+        return int((pow(2.0, number_of_attempts) - 1.0) * random.randint(minimum, maximum))
diff --git a/socketlabs/injectionapi/socketlabsclient.py b/socketlabs/injectionapi/socketlabsclient.py
@@ -5,7 +5,10 @@
 from .core.httpendpoint import HttpEndpoint
 from .core.httprequest import HttpRequest
 from .core.injectionrequestfactory import InjectionRequestFactory
+from .core.injectionresponseparser import InjectionResponseParser
 from .core.sendvalidator import SendValidator
+from .core.retryhandler import RetryHandler
+from .retrysettings import RetrySettings
 from .message.basicmessage import BasicMessage
 from .message.bulkmessage import BulkMessage
 from .proxy import Proxy
@@ -32,6 +35,7 @@ def __init__(self, server_id: int, api_key: str, proxy: Proxy = None):
         self._api_key = api_key
         self._http_proxy = proxy
         self._request_timeout = 120
+        self._number_of_retries = 0
 
     @property
     def __endpoint(self):
@@ -69,6 +73,14 @@ def request_timeout(self, timeout: int):
         """
         self._request_timeout = timeout
 
+    @property
+    def number_of_retries(self):
+        return self._number_of_retries
+    
+    @number_of_retries.setter
+    def number_of_retries(self, retries: int):
+        self._number_of_retries = retries
+
     def __build_http_request(self):
         """
         Build the HttpRequest. Will add the proxy, if set
@@ -110,8 +122,13 @@ def __send_basic_message(self, message: BasicMessage):
         req_factory = InjectionRequestFactory(self._server_id, self._api_key)
         body = req_factory.generate_request(message)
 
-        request = self.__build_http_request()
-        result = request.send_request(body)
+        retry_handler = RetryHandler(self.__build_http_request(), RetrySettings(self.number_of_retries))
+        response = retry_handler.send(body)
+
+        data = response.read().decode("utf-8")
+        response_code = response.status
+        result = InjectionResponseParser.parse(data, response_code)
+
         return result
 
     def __send_bulk_message(self, message: BulkMessage):
@@ -129,8 +146,13 @@ def __send_bulk_message(self, message: BulkMessage):
         req_factory = InjectionRequestFactory(self._server_id, self._api_key)
         body = req_factory.generate_request(message)
 
-        request = self.__build_http_request()
-        result = request.send_request(body)
+        retry_handler = RetryHandler(self.__build_http_request(), RetrySettings(self.number_of_retries))
+        response = retry_handler.send(body)
+
+        data = response.read().decode("utf-8")
+        response_code = response.status
+        result = InjectionResponseParser.parse(data, response_code)
+
         return result
 
     def send_async(self, message: BasicMessage, on_success, on_error):
@@ -167,8 +189,21 @@ def __send_basic_message_async(self, message: BasicMessage, on_success, on_error
         req_factory = InjectionRequestFactory(self._server_id, self._api_key)
         body = req_factory.generate_request(message)
 
-        request = self.__build_http_request()
-        request.send_async_request(body, on_success, on_error)
+        retry_handler = RetryHandler(self.__build_http_request(), RetrySettings(self.number_of_retries))
+
+        def on_success_callback(response):
+            response = retry_handler.send(body)
+
+            data = response.read().decode("utf-8")
+            response_code = response.status
+            result = InjectionResponseParser.parse(data, response_code)
+
+            on_success(result)
+
+        def on_error_callback(exception):
+            on_error(exception)
+
+        retry_handler.send_async(body, on_success_callback, on_error_callback)
 
     def __send_bulk_message_async(self, message: BulkMessage, on_success, on_error):
         """
@@ -187,8 +222,27 @@ def __send_bulk_message_async(self, message: BulkMessage, on_success, on_error):
         req_factory = InjectionRequestFactory(self._server_id, self._api_key)
         body = req_factory.generate_request(message)
 
+        retry_handler = RetryHandler(self.__build_http_request(), RetrySettings(self.number_of_retries))
+
+        def on_success_callback(response):
+            response = retry_handler.send(body)
+
+            data = response.read().decode("utf-8")
+            response_code = response.status
+            result = InjectionResponseParser.parse(data, response_code)
+
+            on_success(result)
+
+        def on_error_callback(exception):
+            on_error(exception)
+
+        retry_handler.send_async(body, on_success_callback, on_error_callback)
+
+        """
         request = self.__build_http_request()
-        request.send_async_request(body, on_success, on_error)
+        retry_handler = RetryHandler(request, RetrySettings(self.number_of_retries))
+        retry_handler.send_async(body, on_success, on_error)
+        """
 
     def __validate_basic_message(self, message: BasicMessage):
         """
