Azure Service Bus Topics

28/06/2016 Manuel Marín Martín 0 Comments

Otro servicio que ofrece el Service Bus de Azure, junto con las ‘Queues’, ‘Relays’ e ‘Event Hub’ son los ‘Topics’. A diferencia de las queues, donde sólo hay un receptor, en los topics puede haber múltiples. Esto es así debido al sistema que emplea basado en suscripciones, en el cual un topic puede tener registradas múltiples suscripciones, estando cada una asociada a un receptor. Dicho receptor solo recibirá los mensajes que hayan sido previamente filtrados por la propia suscripción, lo que permite controlar la información que llega a cada receptor atendiendo, por ejemplo, al perfil del mismo.

Instalación

El proceso de instalación o creación es el mismo que el de una Azure Service Bus Queue. Unicamente tener en cuenta que ‘Messaging Tier’ debe estar establecido como mínimo en ‘Standard’, a diferencia de las queues.
La cadena de conexión se recupera de manera análoga.

Funcionamiento

El funcionamiento es muy similar al visto para las ‘queues’, con lo que todo lo comentado para estas se aplican aquí. De esta manera, este servicio es susceptible de ser empleado con diferentes lenguajes (Java, Node.js, .NET, PHP, Ruby o Python). Para los ejemplos se empleará Python.
La gestión de los topics, así como de las suscripciones pueden hacerse asimismo desde el propio portal.

Un código de ejemplo con las operaciones básicas sería el siguiente:

# Se incluyen las librerias necesarias
from azure.servicebus import ServiceBusService, Message, Topic, Rule, DEFAULT_RULE_NAME

# Conexión con el servicio. Los datos se extraen de la cadena de conexión 
bus_service = ServiceBusService(
    service_namespace='mmariniot-asm-ns',
    shared_access_key_name='RootManageSharedAccessKey',
    shared_access_key_value='XXXXXXXXXXXXXXXXXXXXXXXXXXXX')

# Se crea un topic llamado 'topictest'. Se completan con opciones
bus_service.create_topic('topictest')
topic_options = Topic()
topic_options.max_size_in_megabytes = '5120'
topic_options.default_message_time_to_live = 'PT1M'
bus_service.create_topic('topictest', topic_options)

# Se crea una suscripción llamada 'AllMessages' sin filtros (por defecto),
# lo que significa que recibirá todos los mensajes
bus_service.create_subscription('topictest', 'AllMessages')

# Se crea otra suscripción con un filtro de modo que solo selecciona los mensajes
# cuyo campo predefinido 'messagenumber' sea <=2
bus_service.create_subscription('mytopic', 'LowMessages')
rule = Rule()
rule.filter_type = 'SqlFilter'
rule.filter_expression = 'messagenumber <= 2'
bus_service.create_rule('mytopic', 'LowMessages', 'LowMessageFilter', rule)
bus_service.delete_rule('mytopic', 'LowMessages', DEFAULT_RULE_NAME)

# Envia 3 mensajes al topic
for i in range(3):
    msg_out = Message('Msg {0}'.format(i).encode('utf-8'), custom_properties={'messagenumber':i})
    bus_service.send_topic_message('topictest', msg_out)

# Se recuperan los 3 mensajes para la suscripción 'LowMessages' y se imprimen
for i in range(3):
    msg_in = bus_service.receive_subscription_message('topictest', 'LowMessages', peek_lock=False)
    print(msg_in.body)

# Elimina el topic
bus_service.delete_topic('topictest')

# Elimina las suscripciones
bus_service.delete_subscription('topictest', 'AllMessages')

# Se incluyen las librerias necesarias

from azure.servicebus import ServiceBusService, Message, Topic, Rule, DEFAULT_RULE_NAME

# Conexión con el servicio. Los datos se extraen de la cadena de conexión

bus_service = ServiceBusService(

service_namespace='mmariniot-asm-ns',

shared_access_key_name='RootManageSharedAccessKey',

shared_access_key_value='XXXXXXXXXXXXXXXXXXXXXXXXXXXX')

# Se crea un topic llamado 'topictest'. Se completan con opciones

bus_service.create_topic('topictest')

topic_options = Topic()

topic_options.max_size_in_megabytes = '5120'

topic_options.default_message_time_to_live = 'PT1M'

bus_service.create_topic('topictest', topic_options)

# Se crea una suscripción llamada 'AllMessages' sin filtros (por defecto),

# lo que significa que recibirá todos los mensajes

bus_service.create_subscription('topictest', 'AllMessages')

# Se crea otra suscripción con un filtro de modo que solo selecciona los mensajes

# cuyo campo predefinido 'messagenumber' sea <=2

bus_service.create_subscription('mytopic', 'LowMessages')

rule = Rule()

rule.filter_type = 'SqlFilter'

rule.filter_expression = 'messagenumber <= 2'

bus_service.create_rule('mytopic', 'LowMessages', 'LowMessageFilter', rule)

bus_service.delete_rule('mytopic', 'LowMessages', DEFAULT_RULE_NAME)

# Envia 3 mensajes al topic

for i in range(3):

msg_out = Message('Msg {0}'.format(i).encode('utf-8'), custom_properties={'messagenumber':i})

bus_service.send_topic_message('topictest', msg_out)

# Se recuperan los 3 mensajes para la suscripción 'LowMessages' y se imprimen

for i in range(3):

msg_in = bus_service.receive_subscription_message('topictest', 'LowMessages', peek_lock=False)

print(msg_in.body)

# Elimina el topic

bus_service.delete_topic('topictest')

# Elimina las suscripciones

bus_service.delete_subscription('topictest', 'AllMessages')

Conclusiones

Como se ha visto, el funcionamiento es muy similar a las colas de Services Bus, con la diferencia fundamental del número de receptores del mensaje. Esta característica, unida a la posibilidad de establecer filtros en cada suscripción, permite un grado de versatilidad con que la anterior estructura de datos no contaba.

Cloud Solutions