refactoring

test/fixtures/server.py

@@ -1,12 +1,13 @@
 import json
 import socket
+from collections import deque
 from multiprocessing import Process
 from operator import itemgetter
 
 import flask
 import pytest
 import requests
-from funcy import retry, compose
+from funcy import retry, compose, repeatedly
 from waitress import serve
 
 from test.server import set_up_processing_server
@@ -42,30 +43,31 @@ def server(processor_fn):
 @pytest.fixture
 def processor_fn(item_type, data_items):
     if item_type == "string":
-        return make_string_processor()
+        return make_processor(operation=lambda x: x.decode().upper().encode())
 
 
-def make_string_processor():
-    def processor_fn(request: flask.Request, buffer_size=3):
+def make_processor(operation, buffer_size=3):
+    def processor_fn(request: flask.Request):
 
         data = get_data_from_request(request)
 
-        if not data or len(buffer) == buffer_size:
-            result = [bytes_to_string(itm.decode().upper().encode()) for itm in buffer]
-            buffer.clear()
+        items = repeatedly(buffer.pop, len(buffer)) if consume_buffer_now(buffer, data) else []
 
-        else:
-            result = []
+        result = list(map(compose(bytes_to_string, operation), items))
 
-        if data:
-            buffer.append(data)
+        buffer.append(data)
 
         response_payload = {"data": result}
         return response_payload
 
+    def consume_buffer_now(buffer, data):
+        buffer_is_full = len(buffer) == buffer_size
+        last_request_has_been_sent = not data
+        return buffer_is_full or last_request_has_been_sent
+
     get_data_from_request = compose(string_to_bytes, itemgetter("data"), json.loads, lambda p: p.data.decode())
 
-    buffer = []
+    buffer = deque()
 
     return processor_fn