refactoring

pyinfra/rest.py

@@ -16,28 +16,23 @@ def pack(data: bytes, metadata: dict):
 
 def make_processor(operation, buffer_size=3):
     def processor_fn(request: flask.Request, final=False):
-
-        data = get_data_from_request(request)
-
-        buffer.append(data)
-
-        items = repeatedly(buffer.popleft, len(buffer)) if consume_buffer_now(buffer, data, final) else []
-
-        result = lmap(compose(bytes_to_string, operation), items)
-
+        buffer.append(get_data_from_request(request))
+        result = lmap(operation, repeatedly(buffer.popleft, n_items_to_pop(buffer, final)))
         response_payload = {"data": result}
         return response_payload
 
-    def consume_buffer_now(buffer, data, final):
-        current_buffer_size = len(buffer)
+    def buffer_full(current_buffer_size):
         assert current_buffer_size <= buffer_size
+        return current_buffer_size == buffer_size
 
-        buffer_is_full = current_buffer_size >= buffer_size
-        last_request_has_been_sent = not data
-        return final or buffer_is_full or last_request_has_been_sent
+    def n_items_to_pop(buffer, final):
+        current_buffer_size = len(buffer)
+        return (final or buffer_full(current_buffer_size)) * current_buffer_size
 
     get_data_from_request = compose(string_to_bytes, itemgetter("data"), json.loads, lambda p: p.data.decode())
 
+    operation = compose(bytes_to_string, operation)
+
     buffer = deque()
 
     return processor_fn