updated retry in send async

Author: PraneethChandraThota

python-examples/basic/basic_send_with_retry.py

@@ -5,6 +5,23 @@
 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()
 
@@ -33,5 +50,6 @@
 
 # send the message
 response = client.send(message)
+client.send_async(message, on_success, on_error)
 
 print(json.dumps(response.to_json(), indent=2))

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,14 +151,14 @@ 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 response
 
     def __get_connection(self):

socketlabs/injectionapi/core/retryhandler.py

@@ -1,19 +1,26 @@
 from http import HTTPStatus
+from http.client import HTTPException
 from ..retrysettings import RetrySettings
 from .httprequest import HttpRequest
 from .injectionresponseparser import InjectionResponseParser
+from .serialization.injectionrequest import InjectionRequest
 import sys
 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):
 
@@ -23,17 +30,23 @@ def __init__(self, http_client: HttpRequest, settings: RetrySettings):
     def send(self, body):
 
         if self.__retry_settings.maximum_number_of_retries == 0:
-            return self.__http_client.send_request(body)
 
-        attempts = 0
+            response = self.__http_client.send_request(body)
+            data = response.read().decode("utf-8")
+            response_code = response.status
+            result = InjectionResponseParser.parse(data, response_code)
+            
+            return result
+
         while True:
-            wait_interval = self.__retry_settings.get_next_wait_interval(attempts)
+            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 Exception("Server Error")
+                    raise HTTPException
 
                 data = response.read().decode("utf-8")
                 response_code = response.status
@@ -42,9 +55,52 @@ def send(self, body):
                 return result
 
             except socket.timeout:
-                attempts += 1
-                print("Retry Attempt : ", sys.exc_info()[0])
-                if attempts > self.__retry_settings.maximum_number_of_retries:
+
+                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:
+
+                data = response.read().decode("utf-8")
+                response_code = response.status
+                result = InjectionResponseParser.parse(data, response_code)
+                on_success_callback(result)
+
+        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)
+          

socketlabs/injectionapi/socketlabsclient.py

@@ -181,7 +181,8 @@ def __send_basic_message_async(self, message: BasicMessage, on_success, on_error
         body = req_factory.generate_request(message)
 
         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 __send_bulk_message_async(self, message: BulkMessage, on_success, on_error):
         """
@@ -201,7 +202,8 @@ def __send_bulk_message_async(self, message: BulkMessage, on_success, on_error):
         body = req_factory.generate_request(message)
 
         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):
         """