Merge pull request #31 from DevoInc/release-next

Refactoring code, and fixes in API

Author: Worvast

CHANGELOG.md

@@ -4,6 +4,17 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [1.6.0] - 2019-01-09
+#### Changed
+ * Mild refactoring of Sender class
+ * Refactoring API Response processing
+ * Typos in docs
+ * API Socket recv size
+ 
+#### Fixed
+ * API responses blank lines and splitted lines
+ * Problems with API CLI and automatic shutdowns
+
 
 ## [1.5.1] - 2018-12-28
 #### Fixed

README.md

@@ -1,7 +1,7 @@
 
 [![master Build Status](https://travis-ci.com/DevoInc/python-sdk.svg?branch=master)](https://travis-ci.com/DevoInc/python-sdk) [![LICENSE](https://img.shields.io/dub/l/vibe-d.svg)](https://github.com/DevoInc/python-sdk/blob/master/LICENSE)
 
-[![wheel](https://img.shields.io/badge/wheel-yes-brightgreen.svg)](https://pypi.org/project/devo-sdk/) [![version](https://img.shields.io/badge/version-1.5.1-blue.svg)](https://pypi.org/project/devo-sdk/) [![python](https://img.shields.io/badge/python-2.7%20%7C%203.3%20%7C%203.4%20%7C%203.5%20%7C%203.6%20%7C%203.7-blue.svg)](https://pypi.org/project/devo-sdk/)
+[![wheel](https://img.shields.io/badge/wheel-yes-brightgreen.svg)](https://pypi.org/project/devo-sdk/) [![version](https://img.shields.io/badge/version-1.6.0-blue.svg)](https://pypi.org/project/devo-sdk/) [![python](https://img.shields.io/badge/python-2.7%20%7C%203.3%20%7C%203.4%20%7C%203.5%20%7C%203.6%20%7C%203.7-blue.svg)](https://pypi.org/project/devo-sdk/)
 
 
 # Devo Python SDK

devo/__version__.py

@@ -1,6 +1,6 @@
 __description__ = 'Devo Python Library.'
 __url__ = 'http://www.devo.com'
-__version__ = "1.5.1"
+__version__ = "1.6.0"
 __author__ = 'Devo'
 __author_email__ = 'support@devo.com'
 __license__ = 'MIT'

devo/api/client.py

@@ -157,10 +157,10 @@ def query(self, **kwargs):
         :param kwargs: query -> Query to perform
         :param kwargs: query_id -> Query ID to perform the query
         :param kwargs: dates -> Dict with "from" and "to" keys
-        :param kwargs: proccessor -> proccessor for each row of the response,
+        :param kwargs: processor -> processor for each row of the response,
         if stream
         :param kwargs: stream -> if stream or full response: Object with
-        options of query: proccessor, if stream
+        options of query: processor, if stream
         :param kwargs: response -> response format
         :return: Result of the query (dict) or Buffer object
         """
@@ -268,11 +268,11 @@ def _call_stream(self, payload=None):
             self.socket.send(self._get_stream_headers(payload))
         if not self.buffer.close and not self.buffer.error\
            and self.socket is not None:
-            result, data = self.buffer.proccess_first_line(
-                self.socket.recv(5000))
+            result, data = self.buffer.process_first_line(
+                self.socket.recv(4096))
             if result:
                 try:
-                    while self.buffer.proccess_recv(self.socket.recv(5000)):
+                    while self.buffer.decode(self.socket.recv(4096)):
                         pass
                 except socket.timeout:
                     while not self.buffer.is_empty() or self.buffer.close:
@@ -357,7 +357,7 @@ def _get_stream_headers(self, payload):
         """
         tstamp = str(int(time.time()) * 1000)
 
-        headers = ("POST /%s HTTP/1.1\r\n"
+        headers = ("POST /%s HTTP/2.0\r\n"
                    "Host: %s\r\n"
                    "Content-Type: application/json\r\n"
                    "Content-Length: %s \r\n"

devo/api/scripts/client_cli.py

@@ -3,6 +3,7 @@
 import sys
 import os
 import click
+import time
 from devo.common import Configuration
 from devo.api import Client, DevoClientException
 from devo.common.data.buffer import Buffer
@@ -42,11 +43,11 @@ def cli():
               help='Flag for make streaming query or full query with '
               'start and end. Default is true', default=True)
 @click.option('--proc', help='if flag exists, dont return raw query reply. In '
-                             'compact replies you receive proccessed lines.',
+                             'compact replies you receive processed lines.',
               is_flag=True)
 @click.option('--output', help='File path to store query response if not want '
                                'stdout')
-@click.option('--format', '-f', default="json/simple/compact",
+@click.option('--response', '-r', default="json/simple/compact",
               help='The output format. Default is json/simple/compact')
 @click.option('--from', default=None,
               help='From date, and time for the query (YYYY-MM-DD hh:mm:ss). '
@@ -65,7 +66,7 @@ def query(**kwargs):
                        dates={"from": config['from'],
                               "to": config['to'] if "to" in config.keys()
                                                     else None},
-                       format=config['format'],
+                       response=config['response'],
                        stream=config['stream'])
 
     process_response(buffer, config)
@@ -79,19 +80,19 @@ def identify_response(response, config):
     Identify what type of response are we received from Client API
     :param response: data received from Devo Client API
     :param config: array with launch options
-    :return: proccessed line (List or string normally)
+    :return: processed line (List or string normally)
     """
     return {
         'json': lambda x, y: proc_default(x) if y else x,
         'json/compact': lambda x, y: x,
         'json/simple': lambda x, y: list(x),
         'json/simple/compact': lambda x, y: list(x)
-    }[config['format']](response, config['proc'])
+    }[config['response']](response, config['proc'])
 
 
 def process_response(response, config):
     """
-    Proccess responses from Client API
+    process responses from Client API
     :param response: data received from Devo API
     :param config: array with launch options
     :return: None

devo/common/data/buffer.py

@@ -18,11 +18,13 @@ class Buffer(object):
     def __init__(self, buffer_max_size=1000, api_response="json/compact"):
         self.queue = Queue.Queue(maxsize=buffer_max_size)
         self.thread = None
-        self.temp = None
+        self.temp = ""
+        self.temp_event = ""
+        self.octet = 0
         self.error = None
         self.close = False
-        self.timeout = 0
-        self.response_split = "\n" if api_response is "csv" else "\r\n "
+        self.timeout = None
+        self.response_split = "\n" if api_response is "csv" else "\r\n"
 
     def is_alive(self):
         return self.thread.isAlive()
@@ -42,58 +44,74 @@ def start(self):
         """ Function for call the threat start """
         self.thread.start()
 
-    def get(self, proccessor=None, block=True, timeout=None):
-        """ Get one proccessed item from the queue """
+    def get(self, processor=None, block=True, timeout=None):
+        """ Get one processed item from the queue """
         if timeout is None:
             timeout = self.timeout
 
         if not self.error:
             try:
-                return proccessor(self.queue.get(block=block, timeout=timeout)) \
-                    if proccessor is not None \
+                return processor(self.queue.get(block=block, timeout=timeout)) \
+                    if processor is not None \
                     else self.queue.get(block=block, timeout=timeout)
             except Queue.Empty:
                 self.close = True
         else:
             raise DevoBufferException("Devo-Buffer|%s" % str(self.error))
 
-    def proccess_first_line(self, data):
-        """ Proccess first line of the Query call (For delete headers) """
+    def process_first_line(self, data):
+        """ process first line of the Query call (For delete headers) """
         if not isinstance(data, str):
             data = data.decode('utf8')
 
         if "200 OK" in data.split("\r\n\r\n")[0]:
-            self.proccess_recv(data[data.find("\r\n\r\n")+4:])
-            return True, None
+            return self.decode(data[data.find("\r\n\r\n")+4:]), None
 
         self.error = data
         return False, data
 
-    def size(self):
-        """ Verify queue size """
-        return self.queue.qsize()
-
-    def proccess_recv(self, data):
-        """ Proccess received data """
+    def decode(self, data):
         if not isinstance(data, str):
             data = data.decode('utf8')
+        return self.buffering(data)
 
-        data = data[data.find("\r\n") + 2:].split(self.response_split)
+    def buffering(self, data):
+        if not self.octet:
+            pointer = data.find("\r\n") + 2
+            size = int(data[:pointer], 16)
+            data = self.temp_event + data[pointer:]
+            self.octet = len(self.temp_event) + size
+            self.temp_event = ""
 
-        data_len = len(data)
-        if self.temp is not None:
-            data[0] = self.temp + data[0]
+        if len(self.temp + data) < self.octet or \
+                (not self.octet and not data.find("\r\n")):
+            self.temp += data
+            return not self.close
 
-        if len(data) > 1:
-            for aux in range(0, data_len - 1):
-                self.queue.put(data[aux].strip(), block=True)
+        data = self.temp + data
+        self.temp = data[self.octet + 2:]
+        return self.process_recv(data[:self.octet])
 
-        if data[data_len-1][-4:] == "\r\n\r\n":
-            self.queue.put(data[data_len-1][:-4].strip(), block=True)
-            self.temp = None
-        else:
-            self.temp = data[data_len-1][:-2].strip()
+    def size(self):
+        """ Verify queue size """
+        return self.queue.qsize()
 
+    def process_recv(self, data):
+        """ process received data """
+        data_list = data.strip().split(self.response_split)
+        if data[-(len(self.response_split)):] != self.response_split:
+            self.temp_event = data_list.pop()
+
+        for aux in range(0, len(data_list)):
+            self.queue.put(data_list[aux].strip(), block=True)
+
+        self.octet = 0
+        if len(self.temp.strip()):
+            data = self.temp
+            self.temp = ""
+            if data.find("\r\n") == 0:
+                data = data[2:]
+            return self.buffering(data)
         return not self.close
 
     def close(self):