adding async handling in bulk send

Author: david-schrenker

socketlabs/injectionapi/core/retryhandler.py

@@ -2,14 +2,12 @@
 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):
 
+class RetryHandler(object):
     attempts = 0
     ErrorStatusCodes = [
         HTTPStatus.INTERNAL_SERVER_ERROR,
@@ -26,7 +24,7 @@ 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:
@@ -43,24 +41,24 @@ def send(self, body):
                     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):
+
+    def send_async(self, request: InjectionRequest, on_success_callback, on_error_callback):
 
         wait_interval = self.__retry_settings.get_next_wait_interval(self.attempts)
 
@@ -89,4 +87,3 @@ def on_error(exception):
                 on_error_callback(exception)
 
         self.__http_client.send_async_request(request, on_success, on_error)
-          

socketlabs/injectionapi/retrysettings.py

@@ -1,44 +1,46 @@
 from datetime import timedelta
 import random
 
-class RetrySettings(object):
 
+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):
+    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)
+                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)
-    
-    def get_retry_delta(self, 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))
+    @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))

socketlabs/injectionapi/socketlabsclient.py

@@ -122,9 +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()
-        retry_handler = RetryHandler(request, RetrySettings(self.number_of_retries))
-        result = retry_handler.send(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):
@@ -142,12 +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()
-        retry_handler = RetryHandler(request, RetrySettings(self.number_of_retries))
+        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):
@@ -184,16 +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()
-        retry_handler = RetryHandler(request, RetrySettings(self.number_of_retries))
-        def on_sucess_callback(response):
+        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_sucess_callback, on_error_callback)
+
+        retry_handler.send_async(body, on_success_callback, on_error_callback)
 
     def __send_bulk_message_async(self, message: BulkMessage, on_success, on_error):
         """
@@ -212,9 +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()
         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):
         """