diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index b48dc96..a6cd00a 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/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 # ================================================================================================== diff --git a/amqp/adapter/file_uploader.py b/amqp/adapter/file_uploader.py index 19148b8..12a2eae 100644 --- a/amqp/adapter/file_uploader.py +++ b/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}") diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 1a3887b..1024916 100644 --- a/amqp/service/amq_message_handler.py +++ b/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