This keeps coming up so here is a quick tutorial on how to set up an event bus configuration using the MQTT 2.5 M1 binding. I don’t know if all of these features exist for the 2.4 Release version.
Setting up the binding and MQTT Broker Thing is beyond the scope of this tutorial.
What’s an Event Bus?
There are times when one may have more than one OH instance that needs to have some or all of it’s Items synchronized across all the instances. In the MQTT 1.x binding there was a special configuration that enabled this that no longer exists in MQTT 2.x.
The event bus will have three parts, the subscription, the publishing, and online status reporting.
We will use the following conventions:
| topic | purpose |
|---|---|
<openHAB name>/out/<Item>/command |
Topic where all commands that occur on <openHAB name> are published |
<openHAB name>/out/<Item>/state |
Topic where all state updates that occur on <openHAB name> are published |
<openHAB name>/in/<Item>/command |
Topic where commands that occur in another openHAB instance are published that should be reflected in <openHAB name>. You can have more than one of these groups of topics that you subscribe to if you have three or more openHAB instances to synchronize. |
<openHAB name>/in/<Item>/state |
Topic where state updates that occur in another openHAB instance are published that should be reflected in <openHAB name>You can have more than one of these groups of topics that you subscribe to if you have three or more openHAB instances to synchronize. |
<openHAB name> is a name you chose to uniquely identify each openHAB instance. <Item> will be replaced with the Item name.
Note that you will want to use the same name for all Items that need to be synchronized across the OH instances.
One important thing to note is that the synchronization can be one way. For example, one OH instance can just subscribe to another OH instance without publishing their own commands and updates. This raises a little bit of confusion as to which OH instance needs to publish to which set of topics and which to subscribe. Unfortunately the answer depends on how you are doing the synchronization.
Let’s say we have two OH instances that we want to synchronize: openHAB-main and openHAB-remote.
Then for commands openHAB-main would publish to openHAB-main/out/<Item>/command and subscribe to openHAB-remote/out/<Item>/command. openHAB-remote would publish to openHAB-remote/out/<Item>/command and subscribe to openHAB-main/out/<Item>/command.
Now let’s say that we have a bunch of OH instances and they are all just publishing sensor reading to openHAB-main. In that case openHAB-main would not publish at all and openHAB-main would subscribe to openHAB-main/in/<Item>/command. All the other instances would publish to that same topic.
For the code below, we will be using the following:
| Variable | Value for examples | Purpose |
|---|---|---|
eb_out_gr |
PubItems |
Group containing all the Items whose commands and updates are published to the bus. |
eb_name |
openhab-remote |
Name for the publishing OH instance, will be the root of the topics. |
eb_br |
mqtt:broker:mosquitto |
Thing ID for the MQTT Broker Thing |
eb_in_chan |
mqtt:broker:mosquitto:eventbus |
Channel ID for the Trigger Channel defined on the MQTT Broker Thing for the OH instance that subscribes to the event bus |
Publishing
The first step to publish commands and updates to the out topics is to create a Group. Be sure to give the Group a Type. It doesn’t matter what type we choose because we will not be aggregating the states of it’s members, but without a Type the Group will not receive events that are necessary to drive Rules. For this tutorial I’ll use:
Group:String PubItems
All Items that need to be published to the “event bus” need to be a member of this Group.
Next we will need Rules.
Python:
from core.rules import rule
from core.triggers import when
# !!!! Fill in these variables !!!!
eb_out_gr = "PubItems" # name of the Group containing the Items to publish on the bus
eb_name = "openhab-remote" # name of the openHAB instance, <openHAB name>
eb_br = "mqtt:broker:mosquitto" # Thing ID for the MQTT Broker Thing
@rule("Publish Item events",
description=("Publishes commands and update events for all members of {} "
"to the configured topic".format(eb_out_gr)),
tags=["eventbus"])
@when("Member of {} received command".format(eb_out_gr))
@when("Member of {} received update".format(eb_out_gr))
def pub(event):
is_cmd = hasattr(event, 'itemCommand')
msg = str(event.itemCommand if is_cmd else event.itemState)
topic = "{}/out/{}/{}".format(eb_name, event.itemName,
"command" if is_cmd else "state")
retained = False if is_cmd else True
pub.log.info("Publishing {} to {} on {}".format(msg, topic, eb_br))
action = actions.get("mqtt", eb_br)
if action:
action.publishMQTT(topic, msg, retained)
else:
pub.log.error("There is no broker thing {}!".format(eb_br))
Rules DSL:
rule "Publish commands to the event bus"
when
Member of PubItems received command
then
val mqttActions = getActions("mqtt","mqtt:broker:mosquitto") // use your broker Thing ID
mqttActions.publishMQTT("openhab-remote/out/"+triggeringItem.name+"/command",receivedCommand.toString)
end
rule "Publish updates to the event bus"
when
Member of PubItems received update
then
val mqttActions = getActions("mqtt","mqtt:broker:mosquitto") // use your broker Thing ID
mqttActions.publishMQTT("openHAB-remote/out/"+triggeringItem.name+"/state",triggeringItem.state.toString)
end
That’s all there is to it. All commands to all members of PubItems get published to the command topic and all updates get published to the state topic.
Subscribing
For subscription we need to create a publish trigger Channel on the MQTT Broker Thing on the subscribing instance of OH and configure it to subscribe to the output topics from the other openHAB, or the input topics for this openHAB instance. You can use the wild card subscription to use this one Channel for all the Items. Be sure to define a separator character so we can get the MQTT topic from the event in our Rule.
The screenshot above only subscribes to the command topic. If you want both commands and updates you can use openHAB-remote/out/# as the topic. # is used as the separator character. In the Rule we will split on # to get the topic and message. Take note of the Thing Channel ID, we will use that to trigger the Rule.
With the Channel in place we only need a single Rule.
Python:
from core.rules import rule
from core.triggers import when
# !!!! Fill in this variable !!!!
eb_in_chan = "mqtt:broker:mosquitto:eventbus" # Thing Event Channel ID
@rule("Eventbus subscribe",
description="Subscribe to eventbus events and synchronize the Items.",
tags=["eventbus"])
@when("Channel {} triggered".format(eb_in_chan))
def eb_sub(event):
topic = event.event.split("#")[0]
state = event.event.split("#")[1]
item_name = topic.split("/")[2]
event_type = topic.split("/")[3]
if item_name not in items:
eb_sub.log.debug("Local openHAB does not have Item {}, ignoring."
.format(item_name))
elif event_type == "command":
events.sendCommand(item_name, state)
else:
events.postUpdate(item_name, state)
Rules DSL:
rule "Subscribe for commands and updates from the event bus"
when
Channel 'mqtt:broker:broker:main-openHAB' triggered
then
var topic = receivedEvent.toString.split("#").get(0)
var state = receivedEvent.toString.split("#").get(1)
val itemName = topic.split("/").get(2)
val type = topic.split("/").get(3)
if(type == "command") sendCommand(itemName, state)
else postUpdate(itemName, state)
end
Note that the above handles both commands and updates.
Online Status
It is also desirable to know when the publishing OH instance goes offline. To handle this, we configure the LWT on the MQTT Broker Thing.
- Topic:
<openHAB-Name>/status - Retained: True
- Message: “OFFLINE”
When the OH instance loses it’s connection to the MQTT Broker, the Broker will publish “OFFLINE” to <openHAB-Name>/status. Since the message is retained, even if the subscribers are not online, they will get the OFFLINE message when they reconnect.
Next we need a Rule triggered at Systems started to publish ONLINE as a retained message when the OH instance comes back online.
Rules DSL:
from core.rules import rule
from core.triggers import when
# !!!! Fill in these variables !!!!
eb_name = "openhab-instance" # openHAB's instance name, <openHAB-name>
eb_br = "mqtt:broker:mosquitto" # MQTT Broker Thing ID
@rule("Eventbus Online",
description="Publish that this instance of OH is now online",
tags=["eventbus"])
@when("System started")
@when("Thing {} changed to ONLINE".format(eb_br))
def online(event):
online.log.info("Reporting eventbus as online")
actions.get("mqtt", eb_br).publishMQTT("{}/status".format(eb_name),
"ONLINE", True)
Rules DSL:
rule "Eventbus Online"
when
System started or
Thing mqtt:broker:mosquitto changed to ONLINE
then
logInfo("Reporting eventbus as online")
getActions("mqtt", "mqtt:broker:mosquitto").publishMQTT("openhab-remote/status", "ONLINE", True)
end