Skip to content

Add message queue #38

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Testiot539 wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Add message queue #38

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e9fd3f1
Add message queue
Testiot539 Dec 5, 2023
5a2d429
Change call function
Testiot539 Dec 7, 2023
0a94e24
Change ipfs and rabbit
Testiot539 Dec 10, 2023
f054b87
Add rabbit module and aio_pika
Testiot539 Dec 10, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 34 additions & 5 deletions docker-compose.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,11 @@
version: "3"
version: "3.9"
services:
feecc_workbench_daemon:
environment:
# Use these environment variables to configure your deployment
LANGUAGE_MESSAGE: "ru"
MONGODB_URI: "" # Your MongoDB connection URI
MONGODB_DB_NAME: "" # Your MongoDB DB name
LANGUAGE_MESSAGE: "ru" # Choose message language
MONGODB_URI: "mongodb://root:pass@localhost:27017/?authMechanism=DEFAULT" # Your MongoDB connection URI
MONGODB_DB_NAME: "FEECC-Academy" # Your MongoDB DB name
ROBONOMICS_ENABLE_DATALOG: false # Whether to enable datalog posting or not
ROBONOMICS_ACCOUNT_SEED: "" # Your Robonomics network account seed phrase
ROBONOMICS_SUBSTRATE_NODE_URI: "" # Robonomics network node URI
Expand All @@ -21,7 +21,7 @@ services:
CAMERA_ENABLE: false # Whether to enable Cameraman or not
CAMERA_FFMPEG_COMMAND: ""
WORKBENCH_NUMBER: 1 # Workbench number
HID_DEVICES_RFID_READER: "" # RFID reader device name
HID_DEVICES_RFID_READER: "Sample RFID Scanner" # RFID reader device name
HID_DEVICES_BARCODE_READER: "" # Barcode reader device name
build:
context: ./
Expand All @@ -35,3 +35,32 @@ services:
- "./workbench.pem:/src/workbench.pem:ro"
network_mode: host
restart: always

hid-emulator:
image: nyurik/alpine-python3-requests
container_name: feecc_academy_hid_emulator
network_mode: host
build:
context: hid-emulator
dockerfile: Dockerfile

mongodb:
image: mongo:jammy
container_name: feecc_academy_mongoDB
network_mode: host
environment:
MONGO_INITDB_DATABASE: FEECC-Academy
MONGO_INITDB_ROOT_USERNAME: root
MONGO_INITDB_ROOT_PASSWORD: pass
volumes:
- ./mongodb-local:/docker-entrypoint-initdb.d:ro

rabbitmq:
image: rabbitmq:3.12-management
netwok_mode: host
ports:
- 5672:5672
- 15672:15672
restart: always


1 change: 1 addition & 0 deletions pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ robonomics-interface = "^1.2.2"
ecs-logging = "^2.0.0"
aioprometheus = "^22.5.0"
pycups = "^2.0.1"
aio-pika = "^9.3.1"

[tool.poetry.dev-dependencies]
mypy = "^0.971"
Expand Down
5 changes: 4 additions & 1 deletion src/feecc_workbench/ipfs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
import pathlib

import asyncio
import httpx
from loguru import logger

from .config import CONFIG
from .Messenger import messenger
from .translation import translation
from .rabbit_queue import rabbit_queue
from .utils import async_time_execution, get_headers, service_is_up

IPFS_GATEWAY_ADDRESS: str = CONFIG.ipfs_gateway.ipfs_server_uri
Expand Down Expand Up @@ -36,6 +37,8 @@ async def publish_file(rfid_card_id: str, file_path: pathlib.Path) -> tuple[str,
response = await client.post(url="/by-path", headers=headers, json=json)

if response.is_error:
message=[file_path,headers,base_url,files]
await rabbit_queue(message)
messenger.error(translation('ErrorIPFS') +" "+ response.json().get('detail', ''))
raise httpx.RequestError(response.json().get("detail", ""))

Expand Down
76 changes: 76 additions & 0 deletions src/feecc_workbench/rabbit_queue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
import asyncio
import httpx
from aio_pika import Message, connect
from aio_pika import connect
from aio_pika.abc import AbstractIncomingMessage

n=0
file_path = ''
headers = ''
base_url = ''
files = ''

async def send_message_to_queue(message) -> None:
# Perform connection
connection = await connect("amqp://guest:guest@localhost/")

async with connection:
# Creating a channel
channel = await connection.channel()

# Declaring queue
queue = await channel.declare_queue("ipfs_message")

# Sending the message
await channel.default_exchange.publish(
Message(message.encode()),
routing_key=queue.name,
)

#print(f" [x] Sent {message}")

async def send_message_to_client(file_path,headers,base_url,files):
async with httpx.AsyncClient(base_url=base_url, timeout=None) as client:
if file_path.exists():
httpx.Response = await client.post(url="/upload-file", headers=headers, files=files)
else:
json = {"absolute_path": str(file_path)}
await client.post(url="/by-path", headers=headers, json=json)

async def callback(message: AbstractIncomingMessage):
global n, headers,files,base_url,file_path

message_from_queue = message.body.decode("utf-8")
match n:
case 0: file_path = message_from_queue
case 1: headers = message_from_queue
case 2: base_url = message_from_queue
case 3: files = message_from_queue
n += 1
if n > 3:
n = 0
send_message_to_client(file_path,headers,base_url,files)

async def message_receiv_from_queue() -> None:
# Perform connection
connection = await connect("amqp://guest:guest@localhost/")
async with connection:
# Creating a channel
channel = await connection.channel()

# Declaring queue
queue = await channel.declare_queue("ipfs_message")

# Start listening the queue
await queue.consume(callback, no_ack=True)

#print(" [*] Waiting for messages. To exit press CTRL+C")
asyncio.Future()

await channel.close()

async def rabbit_queue(message):
#message = [_file_path,_headers,_base_url,_files]
for mess in message:
await asyncio.create_task(send_message_to_queue(mess))
await asyncio.create_task(message_receiv_from_queue())