Reached functional file download

amqp/adapter/cleverthis_service_adapter.py

@@ -126,7 +126,7 @@ async def convert_file_response(
     Converts a FileResponse object to a JSON string containing the filename and stream.
     """
     async def wrap_connection_channel(connection: AbstractRobustConnection) -> tuple[AbstractRobustChannel, AbstractExchange, str]:
-        """ Wrap call to channel() to make it Awaitable """
+        """ Wrap calls to RAbbitMQ API to make it Awaitable, so it can be run thread-safe with correct event loop """
         _channel = await connection.channel()
         exchange_name: str = AMQ_REPLY_TO_EXCHANGE
         _exchange = await _channel.get_exchange(name=exchange_name, ensure=True)
@@ -140,12 +140,16 @@ async def convert_file_response(
         logging.debug(f" [*] Publishing to exchange: {exchange_name}, q: {_queue_name}")
         return _channel, _exchange, _queue_name
 
+    # All calls to RabbittMQ API hav eto be called threadsafe, because those appear to be attached to parent
+    # event loop, and need to execute at that event loop, not the current one
     _future = asyncio.run_coroutine_threadsafe(wrap_connection_channel(connection), loop)
     while not _future.done():
-        await asyncio.sleep(0.1) # allow coroutine to execute
+        await asyncio.sleep(0.1) # allow coroutine to execute. calling _future.result() will block the entire thread and the coroutine will never execute
     _channel, _exchange, _routing_key = _future.result()
+    # publish_file also needs RabbitMQ API, ensure the loop is passed as argument
     (_stream_name, _file_size) = await publish_file_to_rabbitmq(_exchange, obj.path, _routing_key, loop)
-    await _channel.close()
+    # as above, but luckily this time we don't need to wait for the result
+    _future = asyncio.run_coroutine_threadsafe(_channel.close(), loop)
     return json.dumps({'files': [
         {'filename': obj.filename,
          'mediaType': 'application/octet-stream',
@@ -200,7 +204,7 @@ async def call_cleverthis_api(
         ) else _json_result.body if hasattr(
             _json_result, 'body'
         ) else _json_result.read() if hasattr(_json_result, 'read') else _json_result
-
+        logging.info(f" [x] ALL DONE in call_cleverthis_api id:{message.id}, resp code:{success_code}")
         return DataResponseFactory.of(message.id, success_code, _content_type, _binary_result, message.trace_info)
     except HTTPException as http_error:
         _tb = traceback.extract_tb(sys.exc_info()[2])[-1]
@@ -320,6 +324,7 @@ class CleverThisServiceAdapter:
                 raise ValueError(f"Unexpected HTTP method: {message.method}")
         else:
             amq_response = DataResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
+        logging.info(f" [x] ALL DONE in on_message id:{message.id}, resp code:{amq_response.response_code}")
         return amq_response
 
     # ==================================================================================================

amqp/adapter/file_uploader.py

@@ -63,10 +63,13 @@ async def publish_file_to_rabbitmq(
                     'eof': END_OF_FILE if max_chunk_size + _offset >= _file_size else IN_PROGRESS
                 }
             )
+            logging.debug(f"Going to Publish chunk {_offset}/{_file_size} to {exchange.name}:{routing_key}")
             _future = asyncio.run_coroutine_threadsafe(exchange.publish(message, routing_key=routing_key), loop)
             while not _future.done():
                 # sleep allows coroutine to execute. must wait for completion to ensure all chunks written in right order
                 await asyncio.sleep(0.02)
+                logging.debug(f"Waiting for chunk {_offset}/{_file_size} future= {_future}")
+
             _offset += len(_chunk)
             logging.debug(f"Published chunk {_offset}/{_file_size} to {exchange.name}:{routing_key}")
 

amqp/service/amq_message_handler.py

@@ -232,7 +232,8 @@ class DataMessageHandler:
             current_parallel_executions -= 1
             return
         except Exception as e:
-            print(f"Error in thread: {e}")
+            _tb = traceback.extract_tb(sys.exc_info()[2])[-1]
+            logging.error(f" [E] Error in thread, {_tb.filename}, line {_tb.lineno}, function '{_tb.name}': {e}")
             return