Hi all!

In a project we needed to define topics to publish messages and register different subscribers which will receive those messages asynchronously. We also needed it to be as simple as possible and that we could use it on InterSystems IRIS directly.

As an experiment, I'm sharing this iris-pubsub open exchange example.

Infrastructure

This is built on top of InterSystems IRIS interoperability features, it needs a running production.

The first time we need to define how many partitions we want to dedicate to handle messages. The more partitions you add, the more capacity to handle messages. For each partition you add it will create a Business Service and a Business Operation.

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

Topics

After that, we can create topics to publish messages. Optionally you can specify a PartitionKey to define which field, in a message published in your topic, will be used to determine the partition that will handle the message. Messages handled in the same partition will be processed in order.

Now we will create a topic called simple/topic and we also specify that the field patientId, in the messages published to the topic, will be used as PartitionKey.

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

Subscribers

We can now create subscribers to the topic we have just created. Each time a message is published to the topic, subscribers will be notified.

We will create two subscribers that are classmethods in ObjectScript. One nice feature would be that these subscribers could be HTTP endpoints as well.

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"})

In this case, subscribers are really simple and will only log information in ^zlog global:

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()
}

Publish messages

We have created a topic called simple/topic and defined two subscribers. Let's publish some messages:

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

You can have a look at the actual messages in the interoperability message viewer:

If we check the ^zlog global where subscribers logged information, we would get the following:

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""}"

An that's it! Any suggestion or pull-request is welcome :)