diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 45caf3b..4c7c040 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -8,6 +8,7 @@ import json import logging import sys import traceback +import pathlib from asyncio import Future from typing import Dict, List, Tuple, Callable, Coroutine, Any from urllib.parse import parse_qs, urlparse @@ -118,13 +119,12 @@ async def convert_file_response(obj: FileResponse, channel: AbstractRobustChanne """ Converts a FileResponse object to a JSON string containing the filename and stream. """ - _future: Future[str] = asyncio.Future(loop=channel.channel.loop) - asyncio.run_coroutine_threadsafe( - publish_file_to_rabbitmq(channel, obj.path, _future), loop=channel.channel.loop) + (_stream_name, _file_size) = await publish_file_to_rabbitmq(channel, obj.path) return json.dumps({'files': [ {'filename': obj.filename, 'mediaType': 'application/octet-stream', - 'streamName': _future.result() + 'streamName': _stream_name, + 'fileSize': _file_size }] }) diff --git a/amqp/adapter/file_uploader.py b/amqp/adapter/file_uploader.py index d8ce6d8..3bd729a 100644 --- a/amqp/adapter/file_uploader.py +++ b/amqp/adapter/file_uploader.py @@ -6,18 +6,19 @@ import os from pathlib import Path from aio_pika.abc import AbstractRobustChannel from aio_pika import Message, DeliveryMode + +from amqp.adapter.file_downloader import IN_PROGRESS, END_OF_FILE from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE async def publish_file_to_rabbitmq( channel: AbstractRobustChannel, file_path: Path, - future: asyncio.Future[str] = None, max_chunk_size: int = 2**20, # 1MB default chunk size exchange_name: str = AMQ_REPLY_TO_EXCHANGE, content_type: str = 'application/octet-stream', # Default content type delivery_mode: DeliveryMode = DeliveryMode.PERSISTENT, -) -> str: +) -> tuple[str, int]: """ Reads a file from the file system and publishes its content to a RabbitMQ exchange in chunks. It declares a queue with a random name, binds it to the specified @@ -63,7 +64,7 @@ async def publish_file_to_rabbitmq( 'offset': _offset, 'total_size': _file_size, 'chunk_size': len(_chunk), - 'last_chunk': max_chunk_size + _offset >= _file_size + 'eof': END_OF_FILE if max_chunk_size + _offset >= _file_size else IN_PROGRESS } ) await _exchange.publish(message, routing_key=_queue_name) @@ -71,6 +72,4 @@ async def publish_file_to_rabbitmq( logging.debug(f"Published chunk {_offset}/{_file_size} to {exchange_name}:{_queue_name}") print(f"File {file_path} published to RabbitMQ exchange {exchange_name}, queue {_queue_name}") - if future: - future.set_result(_queue_name) - return _queue_name + return _queue_name, _file_size