---
# System prepended metadata

title: T.1.3.5 Protocolo AMQP - Topic Exchange

---

# T.1.3.5 Protocolo AMQP - Topic Exchange

![](https://i.imgur.com/dszF8Ih.png)

En este taller se describe el escenario número uno explicado en clase sobre el uso de AMQP-Topic.

## Configuración del consumidor

Copie el siguiente código en un nuevo archivo denominado `amqp-topic-consumer.py` y guarde en la carpeta AMQP . 

```Python=
"""
Created on Sat Jun  5 14:29:11 2021

@author: ADAPTAD By Willian Zamora
         TOPIC STAGE - CONSUMER
"""
import pika
import json

mExchange="puce"
mQueue="portoviejo_puce"
mRouting_Key= "puce.manabi.portoviejo.#" #COMILLAS DOBLES Y SIMPLE PROBAR MAÑANA

credentials = pika.PlainCredentials(username='guest', password='guest')
connection = pika.BlockingConnection(pika.ConnectionParameters
    (host='localhost', port=5672, credentials=credentials))

channel = connection.channel()
channel.exchange_declare(exchange=mExchange, exchange_type='topic')

channel.queue_declare(queue=mQueue)

 #PUEDO RECIBIR UN LISTADO DE KEYS
channel.queue_bind(exchange=mExchange, queue=mQueue, routing_key=mRouting_Key)
print(' [*] Waiting for logs. To exit press CTRL+C')
 
def callback(ch, method, properties, body):
    m_data=json.loads(body)#Decode the JSON string
    print(" [x] %r:%r" % (method.routing_key, m_data))

channel.basic_consume(queue=mQueue, on_message_callback=callback, auto_ack=True)

try:
    channel.start_consuming()
except KeyboardInterrupt:
    print('Interrupted')

print("Fin de ejecución")
```
En este código presenta lo siguiente: i) línea 19 se cambia el tipo de intercambio por "topic" y en la línea 24, se asigna los parámetros de enrutamiento, cola, e intercambio.  Cabe indicar que éste código ya inicia sesión con  usuario y contraseña respectivo.

## Configuración del productor-publicador

Copie el siguiente código en un nuevo archivo denominado `amqp-topic-publisher.py` y guarde en la carpeta AMQP . 

```Python=
"""
Created on Sat Jun  5 14:29:11 2021

@author: ADAPTAD By Willian Zamora
         TOPIC STAGE - PUBLISHED
"""
import pika
import calendar
import time
import json

import random
import geocoder #pip install geocoder

from datetime import datetime

mExchange='puce'
mQueue='portoviejo_puce' 
mRouting_Key = 'puce.manabi.portoviejo.air'

def querymessage(sensor):
    now = datetime.now()
    date_time = now.strftime("%d/%m/%Y %H:%M:%S")
    myloc = geocoder.ip('me')
    latitud = myloc.latlng[0]
    longitud = myloc.latlng[1]
    
    message = {
    "temperatura":random.randint(0,50),
    "humedad":random.randint(50,100),
    "Sensor":sensor,
    "latitud":latitud,
    "longitud":longitud,
    "timestamp":calendar.timegm(time.gmtime()),
    "dateTime": date_time}        
    return message

credentials = pika.PlainCredentials(username='guest', password='guest')
connection = pika.BlockingConnection(pika.ConnectionParameters
    (host='localhost', port=5672, credentials=credentials))

channel = connection.channel()
message = querymessage("A05")

data_out=json.dumps(message)# encode oject to JSON

channel.queue_declare(queue=mQueue)
channel.exchange_declare(exchange=mExchange, exchange_type='topic')
channel.basic_publish(exchange=mExchange, routing_key=mRouting_Key, body=data_out)
print(" [x] Sent %r:%r" % (mRouting_Key, data_out))
connection.close()
```
El código define el intercambiador, la cola y el enrutamiento que debe tener el mensaje. La conexión usa credenciales. Además, se observa una función que nos permite obtener valores de latitud, longitud, temperatura y humedad. Estos valores de temperatura y humedad son valores randomicos.

## Ejecución del programa

Nos encontramos en qué, el consumidor sólo puede recibir valores que se encuentren dentro del dominio "puce.manabi.portoviejo.#".  En este sentido para validar el taller se debe enviar valores que se encuentren en dicho dominio. Es decir: el publicador tiene que ser cualquier sensor ubicado dentro de portoviejo. Ejemplo:
1. puce.manabi.portoviejo.air
2. puce.manabi.portoviejo.water
3. puce.manabi.portoviejo.others

En este contexto procedemos a realizar los siguientes pasos:

1. Ejecute el programa ``amqp-topic-consumer.py`` en la consola número 1.

2. Ejecute el programa ``amqp-topic-publisher.py`` en la consola número 2.

3. Verifique los resultados obtenidos en ambas consolas. En la figura siguiente se muestra dichos resultados. Se evidencia que servidor AMQP recibió los datos del sensor puce.manabi.portoviejo.air.

    ![](https://i.imgur.com/3kGq2xE.png)

4. Ahora valide enviando datos de un sensor de agua. Para dicho propósito modifique el código del publisher y cambie la variable de enrutamiento por: "puce.manabi.portoviejo.water"

    ![](https://i.imgur.com/sXFJydR.png)

5. Verificamos si el mensaje fue recibido por el consumidor
    ![](https://i.imgur.com/z6xfDA2.png)

:::info
:bulb: **Tarea**: Se le solicita que realice lo siguiente:
1. Compruebe lo anterior enviando mensaje de las siguientes rutas:
    1. puce.manabi.manta.air
    2. puce.manabi.portoviejo.otros
    3. puce.guayas.guayaquil.air 
3. Evalue  los escenarios 2 y 3 realizando los cambios respectivos.
:::

