fixes

Author: Ilija Vukotic

src/servicex_did_finder_lib/communication.py

@@ -58,9 +58,8 @@ def send_bulk(self, file_list: List[Dict[str, Any]]):
         'does a bulk put of files'
         for ifl in file_list:
             self._summary.add_file(ifl)
-            if self._summary.file_count == 1:
-                self._servicex.post_transform_start()
-        self._servicex.put_file_add(file_list)
+        self._servicex.put_file_add_bulk(file_list)
+        self._servicex.post_transform_start()
 
 
 async def run_file_fetch_loop(did: str, servicex: ServiceXAdapter, info: Dict[str, Any],

src/servicex_did_finder_lib/servicex_adaptor.py

@@ -34,6 +34,12 @@
 MAX_RETRIES = 3
 
 
+def chunks(lst, n):
+    """Yield successive n-sized chunks from lst."""
+    for i in range(0, len(lst), n):
+        yield lst[i:i + n]
+
+
 class ServiceXAdapter:
     def __init__(self, endpoint, file_prefix=None):
         self.endpoint = endpoint
@@ -88,6 +94,32 @@ def put_file_add(self, file_info):
             self.logger.error(f'After {attempts} tries, failed to send ServiceX App a put_file '
                               f'message: {str(file_info)} - Ignoring error.')
 
+    def put_file_add_bulk(self, file_list):
+        for chunk in chunks(file_list, 20):
+            success = False
+            attempts = 0
+            mesg = []
+            for fi in chunk:
+                mesg.append({
+                    "timestamp": datetime.now().isoformat(),
+                    "paths": [self._prefix_file(fp) for fp in fi['paths']],
+                    'adler32': fi['adler32'],
+                    'file_size': fi['file_size'],
+                    'file_events': fi['file_events']
+                })
+            while not success and attempts < MAX_RETRIES:
+                try:
+                    requests.put(self.endpoint + "/files", json=mesg)
+                    self.logger.info(f"Metric: {json.dumps(mesg)}")
+                    success = True
+                except requests.exceptions.ConnectionError:
+                    self.logger.exception(f'Connection error to ServiceX App. Will retry '
+                                          f'(try {attempts} out of {MAX_RETRIES}')
+                    attempts += 1
+            if not success:
+                self.logger.error(f'After {attempts} tries, failed to send ServiceX App a put_file_bulk '
+                                  f'message: {mesg} - Ignoring error.')
+
     def post_transform_start(self):
         success = False
         attempts = 0