IoT (Internet of Things) is a network of interconnected things, including vehicles, machines, buildings, domestic devices or any other thing with embedded TCP/IP remote connection available, allowing it to receive and send execution instructions and data. Each thing provides one or more services to the IoT network. For instance, smart light bulbs provide services of turning off and turning on the lights; smart air conditioners maintain the environment temperature; smart cameras send notifications when capturing movement. Yet, all those devices can operate together to provide building security, for example.When the camera detects movement, the lights turn on. The orchestration of those processes that can deliver business and integrated services, such as building security, is done by a server software controller. It can receive and send data and instructions from a dozen smart devices to a TCP/IP network (like the Internet) to operate these things as major services.
There are two options of protocols used by things and server software to interoperate data and instructions: MQTT (Message Queuing Telemetry Transport) and REST (Representational State Transfer) API. The first one is more common because it has a simpler structure. Some devices have limited memory and processor features and don’t have enough resources to use REST. Thus, this article will detail how to implement the server-side software to interoperate the IoT using MQTT.
About MQTT
MQTT is a lightweight, publish-subscribe, machine to machine network protocol for message queue/message queuing service. It is designed for connections with remote locations that have devices with resource constraints or limited network bandwidth. It must run over a transport protocol that provides ordered, lossless, bi-directional connections—typically, TCP/IP. It is an open OASIS standard and an ISO recommendation (ISO/IEC 20922). See more on https://en.wikipedia.org/wiki/MQTT.
The MQTT protocol has two actors to interoperate data between things: the message broker and the clients. The first one is a server software used to receive and publish messages to be produced or consumed by one or more clients. The last ones are the smart devices and server softwares used to provide major services using the thing’s services, such as a server software to provide a building automation service, or a software to control the traffic in the city using smart semaphores and public cameras services.
MQTT Broker
There are various MQTT message brokers on the market, but the most popular one is the Eclipse Mosquitto (https://mosquitto.org/), so we will use it in our example.The MQTT Broker is similar to JMS or MQ brokers. It is an intermediate employed to receive messages and post them into topics. Thus, the connected clients can subscribe to one or more topics to produce or consume messages from those topics. The topics allow the clients (software and devices) to do their job in an uncoupled and asynchronous way. It means that each client observes a message and takes action if necessary. They can also post new messages themselves. For instance, if a thermostat sends a message indicating high temperature in the topic “building A”, the air conditioning installed in that building and listed in the topic “building A” can take this message from there and turn itself on to reduce the temperature. However, it is common to see server-side software with business rules automated to turn on the air conditioning when the temperature reaches a specific value. The MQTT broker can also guarantee message delivery if it is configured to retain the messages when down and restore them when on again.
Other advantages of the MQTT Broker include (from Wikipedia):
Eliminating vulnerable and insecure client connections, if configured to do so.
Being able to easily scale from a single device to thousands of gadgets.
Managing and tracking all client connection states, including security credentials and certificates, if configured to do so.
Reducing network strain without compromising the security, if configured to do so (when working with a cellular or satellite network).
MQTT messages
There are 3 types of messages:
Connect: establish a connection between the broker and clients in a TCP/IP session.
Disconnect: disconnect the server and the client from a TCP/IP session.
Publish: put a message into a topic and send a copy of that message to each client that subscribed to the topic.
Check out an example of messages in action:
InterSystems IRIS support to IoT and MQTT protocol
According to the
InterSystems IRIS documentation, the currently supported version is MQTT 3.1. This MQTT specification is defined as an OASIS standard, MQTT Version 3.1.1. The Interoperability module from IRIS defines an Inbound (to consume MQTT messages) and an Outbound (to produce MQTT messages) MQTT Adapter. It is possible to develop custom business services and operations using these adapters, or you can use the built-in business service and operation, EnsLib.MQTT.Service.Passthrough and EnsLib.MQTT.Operation.Passthrough, respectively.
If you want to use the MQTT protocol outside of an interoperability production, you can use the lower-level %Net.MQTT classes. The MQTT classes use the Eclipse Paho MQTT C Client Library.
To create a Business Service to consume MQTT messages
Class packagename.NewService1 Extends Ens.BusinessService
{
Parameter ADAPTER = "EnsLib.MQTT.Adapter.Inbound";
Method OnProcessInput(pInput As EnsLib.MQTT.Message, pOutput As %RegisteredObject) As %Status
{
set tsc=$$$OK
set messageContent = pInput.StringValue
….
Quit tsc
}
}
To create a Business Operation to produce/send MQTT messages
Class packagename.NewOperation1 Extends Ens.BusinessOperation
{
Parameter ADAPTER = "EnsLib.MQTT.Adapter.Outbound";
Parameter SETTINGS = "-SendSuperSession";
Method OnMessage(pRequest As packagename.Request, Output pResponse As packagename.Response) As %Status
{
#dim tSC As %Status = $$$OK
#dim e As %Exception.AbstractException
Try {
Set message = ##class(EnsLib.MQTT.Message).%New()
Set message.Topic = ..Adapter.Topic
Set jsonValue = {}
Set jsonValue.message = "Response message”
Set message.StringValue = jsonValue.%ToJSON()
Set tSC=..Adapter.Send(message.Topic,message.StringValue)
Set pResponse = ##class(packagename.Response).%New()
Set pResponse.message = “Message response”
} Catch e {
Set tSC=e.AsStatus()
}
Quit tSC
}
Settings for the Inbound and Outbound MQTT Adapter
Both Business Service and Business Operation can be configured with the following parameters (source: https://docs.
intersystems.com/irisforhealthlatest/csp/docbook/DocBook.UI.Page.cls?KEY=EMQTT_reference):
Client ID: This is the string that identifies this client to the broker. It must be ASCII-encoded and contain between 1 and 23 characters.
Connect Timeout: This is the connection timeout. Connecting to a busy server may take some time, so this timeout can be used to avoid a premature connection failure. It specifies the number of seconds to wait before a connection attempt fails.
Credentials Name: This is the ID name of the set of credentials values used to access the MQTT broker. The username and password defined in your Credentials item must be ASCII-encoded. It is not needed if the broker does not require login credentials.
Keep Alive: Specifies the maximum number of seconds to pass between messages sent by the client to the broker.
QOS: This determines the quality of service required. It can have either of these two values: 0 - QOSFireAndForget: Do not wait for a response from the broker and 1 - QOSWaitForDelivery: Wait for a response from the broker and issue an error if the broker does not respond.
Retained: This is the flag that indicates to the broker whether the message should be retained.
Topic: This is the name of the topic to which you wish to publish or subscribe. The topic must be ASCII-encoded. The topic is typically a hierarchical string with levels of subtopics separated by a / (forward slash). In a subscription, a topic can have wildcards as a topic level.
URL: This is the URL of the broker with which you wish to communicate. The scheme is either “tcp” or “ssl” followed by the domain name and the port delimited by a “:”, for example, “tcp://BIGBADAPPLE.local:1883”. Typically TLS-enabled endpoints are configured with a port of 8883, but this is not mandatory.
Download and Install the Sample application
The IRIS IoT Sample is a simple application to show you how to consume (receive) and produce (send) MQTT messages using an Interoperability Production. To get it, go to https://openexchange.
intersystems.com/package/IoT-Sample. Now, execute the next steps.
1. Clone/git pull the repo into any local directory
$ git clone https://github.com/yurimarx/iris-iot-sample.git
2. Open the terminal in this directory and run:
$ docker-compose build
3. Run the IRIS container with your project:
$ docker-compose up -d
If you want to install it using ZPM, follow the next steps:
1. Open IRIS Namespace with Interoperability Enabled.2. Open Terminal and call:
USER>zpm "install iris-iot-sample"
Run the Sample Application.
1. Open the production and start it. It will begin observing the MQTT topic /DeviceStatusInputTopic and produce responses to the MQTT topic /DeviceStatusOutputTopic. Check out:
2. Use an MQTT client to send a message and test the production. To do it, go to https://chrome.google.com/webstore/detail/mqttbox/kaajoficamnjijhkeomgfljpicifbkaf on your Google Chrome browser. It is the Chrome Plugin MQTTBox. Just click Add to Chrome and then Add App.
3. In your Google Chrome browser go to chrome://apps/ and select MQTTBox (if required, click Open Anyway).
4. Click Create MQTT Client button.
5. Configure the MQTT connection with these settings:
Client name: Local
Protocol: mqtt / tcp
Host: localhost:1883
Username: admin
Password: admin
All other settings stay the default values.
6. Configure the MQTT topics to send and receive MQTT messages:
Topic to publish: /DeviceStatusInputTopic
Topic to subscribe: /DeviceStatusOutputTopic
Payload:
{
"deviceId":"Air Conditioner Level 1",
"statusDate":"2023-01-07 14:03:00",
"status": 0
}
7. Click the button Subscribe to check the messages on the topic /DeviceStatusOutputTopic.
8. Click the button Publish to send a message to /DeviceStatusInputTopic and see the results produced by IRIS production on /DeviceStatusOutputTopic.
9. Check out the message processing session on IRIS Management Portal Visual Trace
The Sample Application source code
The Dockerfile
ARG IMAGE=intersystemsdc/irishealth-community
ARG IMAGE=intersystemsdc/iris-community
FROM $IMAGE
WORKDIR /home/irisowner/irisbuild
ARG TESTS=0
ARG MODULE="iris-iot-sample"
ARG NAMESPACE="USER"
RUN --mount=type=bind,src=.,dst=. \
iris start IRIS && \
iris session IRIS < iris.script && \
([ $TESTS -eq 0 ] || iris session iris -U $NAMESPACE "##class(%ZPM.PackageManager).Shell(\"test $MODULE -v -only\",1,1)") && \
iris stop IRIS quietly
The latest version of
InterSystems IRIS Community is used to create a docker instance of
InterSystems IRIS with a namespace called USER.
The docker-compose file
version: '3.6'
services:
mosquitto:
image: eclipse-mosquitto:2
container_name: mosquitto
user: root
volumes:
- ./mosquitto/config/:/mosquitto/config/
- ./mosquitto/log/:/mosquitto/log/
- ./mosquitto/data/:/mosquitto/data/
ports:
- 1883:1883
- 9001:9001
iris:
build:
context: .
dockerfile: Dockerfile
restart: always
command: --check-caps false
ports:
- 1972
- 52795:52773
- 53773
volumes:
- ./:/irisdev/app
Two docker container instances are now created, and up and running. The first one is for the MQTT broker (mosquitto service), based on the Eclipse Mosquitto product, and the second one is for the MQTT server-side software, responsible for consuming and producing MQTT messages in order to provide Device Monitoring services.The mosquitto docker instance is configured in the file /mosquitto/config/mosquitto.conf. The port, message persistence, security and log questions are defined there. The file /mosquitto/config/password.txt determines the user as admin with the password admin too. However, it is encrypted by the command mosquitto_passwd -U password.txt (you can read more about this at https://mosquitto.org/man/mosquitto_passwd-1.html).
The iris.script file
zn "%SYS"
Do ##class(Security.Users).UnExpireUserPasswords("*")
zn "USER"
Do ##class(EnsPortal.Credentials).SaveItem(0, "mosquitto_cred","mosquitto_cred","admin","admin","")
zpm "load /home/irisowner/irisbuild/ -v":1:1
halt
This file creates the credentials for the business service and business operation. Login to the MQTT Broker and run the ZPM file module.xml.
The module.xml
This file is used to compile the source code on the server and to install the sample application when using ZPM.
<?xml version="1.0" encoding="UTF-8"?>
<Export generator="Cache" version="25">
<Document name="iris-iot-sample.ZPM">
<Module>
<Name>iris-iot-sample</Name>
<Description>A simple IRIS interoperability application - for IoT using MQTT.</Description>
<Version>1.0.8</Version>
<Packaging>module</Packaging>
<Dependencies>
<ModuleReference>
<Name>sslclient</Name>
<Version>1.0.1</Version>
</ModuleReference>
</Dependencies>
<SourcesRoot>src</SourcesRoot>
<Resource Name="dc.irisiotsample.PKG"/>
<SystemRequirements Version=">=2020.1" Interoperability="enabled" />
</Module>
</Document>
</Export>
On the SourceRoot tag there are source packages to be compiled.
The DeviceStatus persistent class
Class dc.irisiotsample.DeviceStatus Extends %Persistent
{
Property deviceId As %String;
Property statusDate As %TimeStamp;
Property status As %Boolean;
Storage Default
{
<Data name="DeviceStatusDefaultData">
<Value name="1">
<Value>%%CLASSNAME</Value>
</Value>
<Value name="2">
<Value>deviceId</Value>
</Value>
<Value name="3">
<Value>statusDate</Value>
</Value>
<Value name="4">
<Value>status</Value>
</Value>
</Data>
<DataLocation>^dc.irisiotsample.DeviceStatusD</DataLocation>
<DefaultData>DeviceStatusDefaultData</DefaultData>
<IdLocation>^dc.irisiotsample.DeviceStatusD</IdLocation>
<IndexLocation>^dc.irisiotsample.DeviceStatusI</IndexLocation>
<StreamLocation>^dc.irisiotsample.DeviceStatusS</StreamLocation>
<Type>%Storage.Persistent</Type>
}
}
This class is used to persist the DeviceStatus in an SQL table.
The DeviceStatusRequest class
Class dc.irisiotsample.DeviceStatusRequest Extends Ens.Request
{
Property deviceId As %String;
Property statusDate As %TimeStamp;
Property status As %Boolean;
Storage Default
{
<Data name="DeviceStatusRequestDefaultData">
<Subscript>"DeviceStatusRequest"</Subscript>
<Value name="1">
<Value>deviceId</Value>
</Value>
<Value name="2">
<Value>statusDate</Value>
</Value>
<Value name="3">
<Value>status</Value>
</Value>
</Data>
<DefaultData>DeviceStatusRequestDefaultData</DefaultData>
<Type>%Storage.Persistent</Type>
}
}
This class is employed to exchange data between interoperability components.
The DeviceStatusService
Class dc.irisiotsample.DeviceStatusService Extends Ens.BusinessService
{
Parameter ADAPTER = "EnsLib.MQTT.Adapter.Inbound";
Method OnProcessInput(pInput As EnsLib.MQTT.Message, pOutput As Ens.StringResponse) As %Status
{
set tsc=$$$OK
set DeviceStatusValue = ##class(%DynamicAbstractObject).%FromJSON(pInput.StringValue)
set DeviceStatus = ##class(dc.irisiotsample.DeviceStatusRequest).%New()
set DeviceStatus.deviceId = DeviceStatusValue.deviceId
set DeviceStatus.statusDate = DeviceStatusValue.statusDate
set DeviceStatus.status = DeviceStatusValue.status
set tsc =..SendRequestSync("DeviceStatusProcess", DeviceStatus, .Response, -1, "Device Status Process")
set pOutput = Response
quit tsc
}
}
The parameter adapter is indicating the use of the MQTT adapter when it comes to receiving MQTT messages. The method OnProcessInput receives the MQTT message on pInput and sends it to the Business Process, calling the method SendRequestSync as a synchronous message.
The DeviceStatusProcess
Class dc.irisiotsample.DeviceStatusProcess Extends Ens.BusinessProcess
{
Method OnRequest(request As dc.irisiotsample.DeviceStatusRequest, Output response As Ens.StringResponse) As %Status
{
Set tsc = 1
Set response = ##class(Ens.StringResponse).%New()
Set DeviceStatus = ##class(dc.irisiotsample.DeviceStatus).%New()
Set DeviceStatus.deviceId = request.deviceId
Set DeviceStatus.status = request.status
Set DeviceStatus.statusDate = request.statusDate
Set tsc = DeviceStatus.%Save()
If $$$ISOK(tsc) {
Set tsc =..SendRequestSync("DeviceStatusOperation", request, .pResponse, -1, "Device Status Operation")
Set response.StringValue = "Device id "_pResponse.deviceId_" has the status "_pResponse.status
} Else {
Set response.StringValue = "Error on save the device status"
Set SuspendMessage = 1
}
quit tsc
}
Storage Default
{
<Type>%Storage.Persistent</Type>
}
}
This class receives the message from the business service, saves it to the database, and calls the DeviceStatusOperation to send (produce) a message with the results.
The DeviceStatusOperation
Class dc.irisiotsample.DeviceStatusOperation Extends Ens.BusinessOperation
{
Parameter ADAPTER = "EnsLib.MQTT.Adapter.Outbound";
Parameter SETTINGS = "-SendSuperSession";
Method NotifyDeviceStatus(pRequest As dc.irisiotsample.DeviceStatusRequest, Output pResponse As dc.irisiotsample.DeviceStatusResponse) As %Status
{
#dim tSC As %Status = $$$OK
#dim e As %Exception.AbstractException
Try {
Set message = ##class(EnsLib.MQTT.Message).%New()
Set message.Topic = ..Adapter.Topic
Set jsonValue = {}
Set jsonValue.message = "Device "_pRequest.deviceId_" has status "_pRequest.status
Set message.StringValue = jsonValue.%ToJSON()
Set tSC=..Adapter.Send(message.Topic,message.StringValue)
Set pResponse = ##class(dc.irisiotsample.DeviceStatusResponse).%New()
Set pResponse.deviceId = pRequest.deviceId
Set pResponse.status = pRequest.status
} Catch e {
Set tSC=e.AsStatus()
}
Quit tSC
}
XData MessageMap
{
<MapItems>
<MapItem MessageType="dc.irisiotsample.DeviceStatusRequest">
<Method>NotifyDeviceStatus</Method>
</MapItem>
</MapItems>
}
}
This class receives a message from the Business Process and produces an MQTT message to a topic with the process response.
The DeviceStatusProduction
Class dc.irisiotsample.DeviceStatusProduction Extends Ens.Production
{
XData ProductionDefinition
{
<Production Name="dc.irisiotsample.DeviceStatusProduction" LogGeneralTraceEvents="false">
<Description></Description>
<ActorPoolSize>2</ActorPoolSize>
<Item Name="DeviceStatusService" Category="" ClassName="dc.irisiotsample.DeviceStatusService" PoolSize="1" Enabled="true" Foreground="false" Comment="" LogTraceEvents="false" Schedule="">
<Setting Target="Adapter" Name="ClientID">InterSystemsIRIS</Setting>
<Setting Target="Adapter" Name="Topic">/DeviceStatusInputTopic</Setting>
<Setting Target="Adapter" Name="Url">tcp://mosquitto:1883</Setting>
<Setting Target="Adapter" Name="CredentialsName">mosquitto_cred</Setting>
</Item>
<Item Name="DeviceStatusProcess" Category="" ClassName="dc.irisiotsample.DeviceStatusProcess" PoolSize="1" Enabled="true" Foreground="false" Comment="" LogTraceEvents="false" Schedule="">
</Item>
<Item Name="DeviceStatusOperation" Category="" ClassName="dc.irisiotsample.DeviceStatusOperation" PoolSize="1" Enabled="true" Foreground="false" Comment="" LogTraceEvents="false" Schedule="">
<Setting Target="Adapter" Name="ClientID">InterSystemsIRIS</Setting>
<Setting Target="Adapter" Name="Topic">/DeviceStatusOutputTopic</Setting>
<Setting Target="Adapter" Name="Url">tcp://mosquitto:1883</Setting>
<Setting Target="Adapter" Name="CredentialsName">mosquitto_cred</Setting>
</Item>
</Production>
}
}
This class configures the Business Service, the Business Process, and the Business Operation with the required parameters, and puts these components to work together.
The main configured parameters are:
URL: to set the MQTT Broker address.
Topic: the queue to receive and send MQTT messages. DeviceStatusInputTopic for the business service and DeviceStatusOutputTopic for the Business Operation.
CredentialsName: set the credentials with the username and password to connect with the MQTT broker.
ClientID: it is a logical name assigned for the
InterSystems IRIS and required by the broker to identify MQTT clients.
As you can see, the IRIS interoperability adapter for MQTT is a very easy and powerful tool for consuming and producing MQTT messages and automating the business flow when it comes to IoT devices. So, enjoy it. Very good article, Yuri.
Wonder when
Intersystems will support AMQP protocol, to integrate seamlessly with ASB (Azure Service Bus) Thanks! Yaron, I'd appreciate if you submit it as an idea This is cool. Thank you very much. This gives you the opportunity to do a "Smart Home"" server in COS :-) - funny. Many thanks Thanks Yuri for this article and app.
As MQTTBox doesn't work for me (no longer compatible), I'm using MQTT Explorer app to send messages :https://apps.apple.com/app/apple-store/id1455214828 Thanks! Of course MQTT Explorer also works on Windows, Ubuntu and any Linux platforms.
http://mqtt-explorer.com/
https://github.com/thomasnordquist/MQTT-Explorer/releases
