From 612eb3db7cdb792d82454c0c8b3c30f655a2bae0 Mon Sep 17 00:00:00 2001 From: gax Date: Wed, 1 Mar 2023 18:30:51 +0000 Subject: [PATCH 01/11] cliente mqtt con docker --- clienteMqtt/Dockerfile | 11 +++++++++++ clienteMqtt/clienteMqtt.py | 30 ++++++++++++++++++++++++++++++ clienteMqtt/requirements.txt | 3 +++ 3 files changed, 44 insertions(+) create mode 100755 clienteMqtt/Dockerfile create mode 100755 clienteMqtt/clienteMqtt.py create mode 100755 clienteMqtt/requirements.txt diff --git a/clienteMqtt/Dockerfile b/clienteMqtt/Dockerfile new file mode 100755 index 00000000..52d78340 --- /dev/null +++ b/clienteMqtt/Dockerfile @@ -0,0 +1,11 @@ +FROM python:3.11-slim + +WORKDIR /app + +COPY ./requirements.txt /app/requirements.txt + +RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt + +COPY . /app + +CMD ["python", "/app/clienteMqtt.py"] \ No newline at end of file diff --git a/clienteMqtt/clienteMqtt.py b/clienteMqtt/clienteMqtt.py new file mode 100755 index 00000000..96250d84 --- /dev/null +++ b/clienteMqtt/clienteMqtt.py @@ -0,0 +1,30 @@ +import asyncio, ssl, certifi, logging +from asyncio_mqtt import Client, ProtocolVersion +from environs import Env + +env = Env() +env.read_env() #lee el archivo con las variables. por defecto .env + +logging.basicConfig(format='%(asctime)s - cliente mqtt - %(levelname)s:%(message)s', level=logging.INFO, datefmt='%d/%m/%Y %H:%M:%S') + +async def main(): + tls_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + tls_context.minimum_version = ssl.TLSVersion.TLSv1_2 + tls_context.maximum_version = ssl.TLSVersion.TLSv1_3 + tls_context.verify_mode = ssl.CERT_REQUIRED + tls_context.check_hostname = True + tls_context.load_default_certs() + + async with Client( + env("SERVIDOR"), + protocol=ProtocolVersion.V31, + port=8883, + tls_context=tls_context, + ) as client: + async with client.messages() as messages: + await client.subscribe("#") + async for message in messages: + logging.info(str(message.topic) + ": " + message.payload.decode("utf-8")) + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/clienteMqtt/requirements.txt b/clienteMqtt/requirements.txt new file mode 100755 index 00000000..84ced5d6 --- /dev/null +++ b/clienteMqtt/requirements.txt @@ -0,0 +1,3 @@ +asyncio-mqtt==0.16.1 +certifi==2022.12.7 +environs==9.5.0 \ No newline at end of file From e1feec168b0d5a34dcdce069d992ac69aa4deae9 Mon Sep 17 00:00:00 2001 From: gax Date: Wed, 1 Mar 2023 17:29:12 -0300 Subject: [PATCH 02/11] uso de ENV --- clienteMqtt/Dockerfile | 2 ++ clienteMqtt/clienteMqtt.py | 9 +++------ 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/clienteMqtt/Dockerfile b/clienteMqtt/Dockerfile index 52d78340..e3586266 100755 --- a/clienteMqtt/Dockerfile +++ b/clienteMqtt/Dockerfile @@ -8,4 +8,6 @@ RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt COPY . /app +ENV TZ="America/Argentina/Buenos_Aires" + CMD ["python", "/app/clienteMqtt.py"] \ No newline at end of file diff --git a/clienteMqtt/clienteMqtt.py b/clienteMqtt/clienteMqtt.py index 96250d84..2e17b8e9 100755 --- a/clienteMqtt/clienteMqtt.py +++ b/clienteMqtt/clienteMqtt.py @@ -1,11 +1,8 @@ -import asyncio, ssl, certifi, logging +import asyncio, ssl, certifi, logging, os from asyncio_mqtt import Client, ProtocolVersion from environs import Env -env = Env() -env.read_env() #lee el archivo con las variables. por defecto .env - -logging.basicConfig(format='%(asctime)s - cliente mqtt - %(levelname)s:%(message)s', level=logging.INFO, datefmt='%d/%m/%Y %H:%M:%S') +logging.basicConfig(format='%(asctime)s - cliente mqtt - %(levelname)s:%(message)s', level=logging.INFO, datefmt='%d/%m/%Y %H:%M:%S %z') async def main(): tls_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) @@ -16,7 +13,7 @@ async def main(): tls_context.load_default_certs() async with Client( - env("SERVIDOR"), + os.environ['SERVIDOR'], protocol=ProtocolVersion.V31, port=8883, tls_context=tls_context, From d2fad211888ad88115c17850d7795165322f9a42 Mon Sep 17 00:00:00 2001 From: gax Date: Thu, 2 Mar 2023 07:53:47 -0300 Subject: [PATCH 03/11] orden ENV --- clienteMqtt/Dockerfile | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/clienteMqtt/Dockerfile b/clienteMqtt/Dockerfile index e3586266..8353d235 100755 --- a/clienteMqtt/Dockerfile +++ b/clienteMqtt/Dockerfile @@ -2,12 +2,12 @@ FROM python:3.11-slim WORKDIR /app +ENV TZ="America/Argentina/Buenos_Aires" + COPY ./requirements.txt /app/requirements.txt RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt COPY . /app -ENV TZ="America/Argentina/Buenos_Aires" - CMD ["python", "/app/clienteMqtt.py"] \ No newline at end of file From b1d71c60016beddabfe2a53ff472027e8481cb10 Mon Sep 17 00:00:00 2001 From: gax Date: Thu, 2 Mar 2023 08:44:01 -0300 Subject: [PATCH 04/11] env var TOPICO --- clienteMqtt/clienteMqtt.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/clienteMqtt/clienteMqtt.py b/clienteMqtt/clienteMqtt.py index 2e17b8e9..59303557 100755 --- a/clienteMqtt/clienteMqtt.py +++ b/clienteMqtt/clienteMqtt.py @@ -19,7 +19,7 @@ async def main(): tls_context=tls_context, ) as client: async with client.messages() as messages: - await client.subscribe("#") + await client.subscribe(os.environ['TOPICO']) async for message in messages: logging.info(str(message.topic) + ": " + message.payload.decode("utf-8")) From 83e690eaf209d1026f7cab0c30907a48cb0a5611 Mon Sep 17 00:00:00 2001 From: gax Date: Wed, 8 Mar 2023 16:46:10 +0000 Subject: [PATCH 05/11] Actualizar 'clienteMqtt/clienteMqtt.py' ya no se utiliza environs --- clienteMqtt/clienteMqtt.py | 1 - 1 file changed, 1 deletion(-) diff --git a/clienteMqtt/clienteMqtt.py b/clienteMqtt/clienteMqtt.py index 59303557..cc8f444f 100755 --- a/clienteMqtt/clienteMqtt.py +++ b/clienteMqtt/clienteMqtt.py @@ -1,6 +1,5 @@ import asyncio, ssl, certifi, logging, os from asyncio_mqtt import Client, ProtocolVersion -from environs import Env logging.basicConfig(format='%(asctime)s - cliente mqtt - %(levelname)s:%(message)s', level=logging.INFO, datefmt='%d/%m/%Y %H:%M:%S %z') From 6ded9fe60c620540b02f10dce58c85f8affa91e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Germ=C3=A1n=20Xander?= Date: Mon, 15 Apr 2024 10:38:39 -0300 Subject: [PATCH 06/11] upgrade to aiomqtt --- clienteMqtt/.gitignore | 1 + clienteMqtt/clienteMqtt.py | 14 +++++--------- 2 files changed, 6 insertions(+), 9 deletions(-) create mode 100644 clienteMqtt/.gitignore diff --git a/clienteMqtt/.gitignore b/clienteMqtt/.gitignore new file mode 100644 index 00000000..2eea525d --- /dev/null +++ b/clienteMqtt/.gitignore @@ -0,0 +1 @@ +.env \ No newline at end of file diff --git a/clienteMqtt/clienteMqtt.py b/clienteMqtt/clienteMqtt.py index cc8f444f..79fec415 100755 --- a/clienteMqtt/clienteMqtt.py +++ b/clienteMqtt/clienteMqtt.py @@ -1,26 +1,22 @@ import asyncio, ssl, certifi, logging, os -from asyncio_mqtt import Client, ProtocolVersion +import aiomqtt logging.basicConfig(format='%(asctime)s - cliente mqtt - %(levelname)s:%(message)s', level=logging.INFO, datefmt='%d/%m/%Y %H:%M:%S %z') async def main(): tls_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - tls_context.minimum_version = ssl.TLSVersion.TLSv1_2 - tls_context.maximum_version = ssl.TLSVersion.TLSv1_3 tls_context.verify_mode = ssl.CERT_REQUIRED tls_context.check_hostname = True tls_context.load_default_certs() - async with Client( + async with aiomqtt.Client( os.environ['SERVIDOR'], - protocol=ProtocolVersion.V31, port=8883, tls_context=tls_context, ) as client: - async with client.messages() as messages: - await client.subscribe(os.environ['TOPICO']) - async for message in messages: - logging.info(str(message.topic) + ": " + message.payload.decode("utf-8")) + await client.subscribe(os.environ['TOPICO']) + async for message in client.messages: + logging.info(str(message.topic) + ": " + message.payload.decode("utf-8")) if __name__ == "__main__": asyncio.run(main()) From 4c17792559431048ed533e2599b8c04816b8c1bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Germ=C3=A1n=20Xander?= Date: Mon, 15 Apr 2024 10:47:42 -0300 Subject: [PATCH 07/11] upgrade to aiomqtt --- clienteMqtt/requirements.txt | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/clienteMqtt/requirements.txt b/clienteMqtt/requirements.txt index 84ced5d6..412188d5 100755 --- a/clienteMqtt/requirements.txt +++ b/clienteMqtt/requirements.txt @@ -1,3 +1,3 @@ -asyncio-mqtt==0.16.1 -certifi==2022.12.7 -environs==9.5.0 \ No newline at end of file +aiomqtt==2.0.1 +certifi==2024.2.2 +environs==11.0.0 \ No newline at end of file From 48a10230cad61b69abc2a118ea56e7d96bb5b2e4 Mon Sep 17 00:00:00 2001 From: LAV419 Date: Mon, 14 Apr 2025 15:46:52 -0300 Subject: [PATCH 08/11] Ej --- .gitignore | 1 + compose.yaml | 10 ++++++++++ 2 files changed, 11 insertions(+) create mode 100644 .gitignore create mode 100644 compose.yaml diff --git a/.gitignore b/.gitignore new file mode 100644 index 00000000..2eea525d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +.env \ No newline at end of file diff --git a/compose.yaml b/compose.yaml new file mode 100644 index 00000000..e5ab8f2a --- /dev/null +++ b/compose.yaml @@ -0,0 +1,10 @@ +services: + clientemqtt: + image: clientemqtt + container_name: clientemqtt + environment: + - TZ=America/Argentina/Buenos_Aires + - SERVIDOR=${SERVIDOR} #Se reemplazará por la env var SERVIDOR defineda en .env + - TOPICO=${TOPICO} + restart: unless-stopped + \ No newline at end of file From aff4ba5252f638774c966876c34423c7e38325a2 Mon Sep 17 00:00:00 2001 From: LAV419 Date: Mon, 14 Apr 2025 15:53:51 -0300 Subject: [PATCH 09/11] final --- clienteMqtt/Dockerfile | 2 - clienteMqtt/clienteMqtt.py | 79 +++++++++++++++++++++++++++++++++----- compose.yaml | 5 ++- 3 files changed, 72 insertions(+), 14 deletions(-) diff --git a/clienteMqtt/Dockerfile b/clienteMqtt/Dockerfile index 8353d235..52d78340 100755 --- a/clienteMqtt/Dockerfile +++ b/clienteMqtt/Dockerfile @@ -2,8 +2,6 @@ FROM python:3.11-slim WORKDIR /app -ENV TZ="America/Argentina/Buenos_Aires" - COPY ./requirements.txt /app/requirements.txt RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt diff --git a/clienteMqtt/clienteMqtt.py b/clienteMqtt/clienteMqtt.py index 79fec415..304678f5 100755 --- a/clienteMqtt/clienteMqtt.py +++ b/clienteMqtt/clienteMqtt.py @@ -1,22 +1,81 @@ -import asyncio, ssl, certifi, logging, os +import asyncio +import ssl +import certifi +import logging +import os import aiomqtt -logging.basicConfig(format='%(asctime)s - cliente mqtt - %(levelname)s:%(message)s', level=logging.INFO, datefmt='%d/%m/%Y %H:%M:%S %z') +# Configuración del logger +logging.basicConfig( + format='%(asctime)s %(funcName)s - %(message)s', + level=logging.INFO, + datefmt='%d/%m/%Y %H:%M:%S ', +) + +class Contador: + def __init__(self): + self._valor = 0 + self._lock = asyncio.Lock() + + async def incrementar(self): + async with self._lock: + self._valor += 1 + + async def obtener(self): + async with self._lock: + return self._valor + +async def read_messages(nombre, queue): + logger = logging.getLogger(nombre) + while True: + mensaje = await queue.get() + logger.info(f"[{mensaje.topic}] : {mensaje.payload.decode('utf-8')}") + +async def distributor(client, topico1, topico2, queue1, queue2): + async for mensaje in client.messages: + if mensaje.topic.matches(topico1): + queue1.put_nowait(mensaje) + elif mensaje.topic.matches(topico2): + queue2.put_nowait(mensaje) + +async def incremento(contador): + while True: + await contador.incrementar() + await asyncio.sleep(3) + +async def publicacion(client, publish_topic, contador): + logger = logging.getLogger("publisher") + while True: + valor = await contador.obtener() + await client.publish(publish_topic, str(valor)) + logger.info(f"Publicado: {valor}") + await asyncio.sleep(5) async def main(): + servidor = os.environ["SERVIDOR"] + topico1 = os.environ["TOPICO1"] + topico2 = os.environ["TOPICO2"] + publish_topic = os.environ["TOPICO_PUB"] + tls_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) tls_context.verify_mode = ssl.CERT_REQUIRED tls_context.check_hostname = True tls_context.load_default_certs() - async with aiomqtt.Client( - os.environ['SERVIDOR'], - port=8883, - tls_context=tls_context, - ) as client: - await client.subscribe(os.environ['TOPICO']) - async for message in client.messages: - logging.info(str(message.topic) + ": " + message.payload.decode("utf-8")) + queue1 = asyncio.Queue() + queue2 = asyncio.Queue() + contador = Contador() + + async with aiomqtt.Client(servidor, port=8883, tls_context=tls_context) as client: + await client.subscribe(topico1) + await client.subscribe(topico2) + + async with asyncio.TaskGroup() as tg: + tg.create_task(distributor(client, topico1, topico2, queue1, queue2)) + tg.create_task(read_messages("topico1", queue1)) + tg.create_task(read_messages("topico2", queue2)) + tg.create_task(incremento(contador)) + tg.create_task(publicacion(client, publish_topic, contador)) if __name__ == "__main__": asyncio.run(main()) diff --git a/compose.yaml b/compose.yaml index e5ab8f2a..7f6f64ef 100644 --- a/compose.yaml +++ b/compose.yaml @@ -3,8 +3,9 @@ services: image: clientemqtt container_name: clientemqtt environment: - - TZ=America/Argentina/Buenos_Aires - SERVIDOR=${SERVIDOR} #Se reemplazará por la env var SERVIDOR defineda en .env - - TOPICO=${TOPICO} + - TOPICO1=${TOPICO1} + - TOPICO2=${TOPICO2} + - TOPICO_PUB=${TOPICO_PUB} restart: unless-stopped \ No newline at end of file From c2f0b06e17c1c16a594e04a9ea7799761692985b Mon Sep 17 00:00:00 2001 From: LAV419 Date: Mon, 14 Apr 2025 16:05:51 -0300 Subject: [PATCH 10/11] ahorasi --- clienteMqtt/clienteMqtt.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/clienteMqtt/clienteMqtt.py b/clienteMqtt/clienteMqtt.py index 304678f5..2fb82bfa 100755 --- a/clienteMqtt/clienteMqtt.py +++ b/clienteMqtt/clienteMqtt.py @@ -31,7 +31,7 @@ async def read_messages(nombre, queue): mensaje = await queue.get() logger.info(f"[{mensaje.topic}] : {mensaje.payload.decode('utf-8')}") -async def distributor(client, topico1, topico2, queue1, queue2): +async def repartidor(client, topico1, topico2, queue1, queue2): async for mensaje in client.messages: if mensaje.topic.matches(topico1): queue1.put_nowait(mensaje) @@ -71,9 +71,9 @@ async def main(): await client.subscribe(topico2) async with asyncio.TaskGroup() as tg: - tg.create_task(distributor(client, topico1, topico2, queue1, queue2)) tg.create_task(read_messages("topico1", queue1)) tg.create_task(read_messages("topico2", queue2)) + tg.create_task(repartidor(client, topico1, topico2, queue1, queue2)) tg.create_task(incremento(contador)) tg.create_task(publicacion(client, publish_topic, contador)) From 43f8ce46cff6d031276f6f1a460c9a0a6f34ba02 Mon Sep 17 00:00:00 2001 From: LAV419 <73648857+LAV419@users.noreply.github.com> Date: Mon, 14 Apr 2025 16:06:26 -0300 Subject: [PATCH 11/11] Update README.md --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index b5710abe..46e0b811 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,3 @@ -# docker +# Ejercicio docker +Vellbach, Lucas Alejandro