Hola a todos!

Durante un proyecto necesitábamos poder definir temas sobre los que publicar mensajes, y crear diferentes subscriptores que recibiesen esos mensajes de forma asíncrona. Necesitábamos además que fuese lo más sencillo posible y que se pudiese utilizar directamente en InterSystems IRIS.

A modo de experimento os paso este iris-pubsub.

Infraestructura

Está construido sobre las funcionalidades de interoperabilidad de InterSystems IRIS, necesita tener una producción en marcha.

La primera vez que lo arranquemos definimos cuántas particiones queremos dedicar para atender mensajes. A mayor número de particiones, mayor capacidad. Por cada partición creará un Business Service y un Business Operation.

do ##class(dc.PubSub.API).AddPartitions(3)

Temas (topics)

A continuación podemos crear temas o topics sobre los que queramos publicar mensajes. Opcionalmente se puede definir un PartitionKey que servirá para indicar qué campo del mensaje que publicamos se utilizará para decidir la partición lo gestionará. Los mensajes que se publican en una misma partición se procesan en orden.

Aquí creamos un tema llamado simple/topic e indicamos que la PartitionKey será el campo patientId de los mensajes que se publiquen.

set topic = ##class(dc.PubSub.API).CreateTopic("simple/topic", { "PartitionKey": "patientId" })

Subscriptores

Ahora podemos crear subscriptores para el tema que hemos creado. Cuando se publique un mensaje en el tema o topic los subscriptores serán notificados.

Crearemos dos subscriptores que son dos classmethod de clases en ObjectScript. Una posible mejora sería poder definir subscriptores que fuesen endpoints HTTP.

do ##class(dc.PubSub.API).CreateSubscription("simple/topic", { "Protocol": "ClassMethod", "Endpoint": "USER:dc.PubSub.Test.Simple:Subscriber"})
do ##class(dc.PubSub.API).CreateSubscription("simple/topic", { "Protocol": "ClassMethod", "Endpoint": "USER:dc.PubSub.Test.Simple:Sub2"})

En este caso, los subscriptores son muy sencillos y sólo van a registrar información en la global ^zlog:

ClassMethod Subscriber(payload As %String)
{
    set obj = {}.%FromJSON(payload)
    set ^zlog($i(^zlog)) = "["_$classname()_":Subscriber] Received: "_obj.%ToJSON()
}

ClassMethod Sub2(payload As %String)
{
    set obj = {}.%FromJSON(payload)
    set ^zlog($i(^zlog)) = "["_$classname()_":Sub2] Received: "_obj.%ToJSON()
}

Publicar mensajes

Hemos creado un tema llamado simple/topic y hemos definido dos subscriptores. Vamos a publicar un par de mensajes:

do ##class(dc.PubSub.API).Publish("simple/topic", {"patientId": "HA98744455", "data": "dummy" } )
do ##class(dc.PubSub.API).Publish("simple/topic", {"patientId": "12TFFFHM88", "data": "dummy999" } )

Los mensajes podemos verlos en las trazas de interoperabilidad de IRIS:

Y si comprobamos la global ^zlog donde los subscriptores guardan información, veremos lo siguiente:

USER>zw ^zlog
^zlog=4
^zlog(1)="[dc.PubSub.Test.Simple:Subscriber] Received: {""patientId"":""HA98744455"",""data"":""dummy""}"
^zlog(2)="[dc.PubSub.Test.Simple:Sub2] Received: {""patientId"":""HA98744455"",""data"":""dummy""}"
^zlog(3)="[dc.PubSub.Test.Simple:Subscriber] Received: {""patientId"":""12TFFFHM88"",""data"":""dummy999""}"
^zlog(4)="[dc.PubSub.Test.Simple:Sub2] Received: {""patientId"":""12TFFFHM88"",""data"":""dummy999""}"

¡Y esto es todo! Cualquier sugerencia o pull-request es bienvenida :)