MQTT in LabVIEW: Libraries and Patterns for Test Benches

LabVIEW has no MQTT functions in the core palette, but adding them is easy: install a G library, call paho-mqtt through the Python Node, or run a small bridge next to your VI. The harder part is architecture. A test bench that publishes from inside its measurement loop will eventually stall a test when the network hiccups. This guide compares the options and shows a pattern that keeps measurements deterministic and results reliable.

Your options at a glance

The most widely used open-source option is the MQTT Client library from the LabVIEW Open Source Project, available through VIPM and developed on GitHub. Commercial toolkits also exist on the NI Tools Network. They all follow the same model: create a client, connect, subscribe, receive messages through an event or a queue, publish, disconnect. Feature sets change between releases, so read the package description for the protocol version and TLS options before you commit.

Design the topics first

Agree on a topic layout before wiring anything. For a test lab, one subtree per station works well:

The retained status topic, combined with a Last Will that publishes offline, gives every dashboard the true state of every station, even after a crash or a pulled network cable. See MQTT Last Will and retained messages for how both behave. For larger labs, the general rules in topic design best practices apply.

For payloads, build a cluster (serial number, test step results, limits, pass/fail, timestamps) and convert it with Flatten To JSON. On the receiving side, Unflatten From JSON with the same cluster type turns commands back into typed data.

The architecture that does not stall tests

Never call Publish from inside a timed measurement loop. A publish can block for seconds while the client reconnects, and your test timing will silently drift.

Use a producer/consumer or Queued Message Handler layout instead:

1. Measurement loop: runs the test sequence and enqueues results and telemetry onto a LabVIEW queue. It never touches the network.
2. MQTT loop: owns the client reference. It dequeues items and publishes them, receives commands (through the library's user event or a dequeue with timeout) and forwards them to the measurement loop through a second queue.
3. Connection handling: the MQTT loop reconnects with a back-off and resubscribes after every reconnect. Results that arrive while offline stay in the queue (or a local file) and are published once the connection is back.

This separation also makes the bench testable: you can run the measurement loop with the MQTT loop replaced by a stub that writes to a file.

Option A: a native G library

With a G library installed from VIPM, the block diagram of the MQTT loop typically looks like this:

1. Create/Open client: broker host broker.connectmq.com, port 1883 (or 8883 with TLS), a unique client ID such as labview-bench01-<random>, keep-alive 30 s.
2. Set Will: topic testmqtt/labview/bench01/status, payload offline, QoS 1, retain true.
3. Connect with username and password from a config file, never hard-coded in the VI.
4. Publish online retained to the status topic, then Subscribe to testmqtt/labview/bench01/cmd/# with QoS 1.
5. Event structure / while loop: handle "message received" events and dequeue outgoing items.
6. On shutdown: publish offline retained, then Disconnect cleanly (a clean disconnect does not trigger the Will).

If the library supports TLS, it usually relies on LabVIEW's built-in TLS functions and needs the broker's CA certificate. Test TLS from a PC first so that certificate problems do not look like LabVIEW bugs.

Option B: paho-mqtt through the Python Node

The Python Node (LabVIEW 2018+) calls functions from a Python module. Because a Python session keeps module state between calls, you can hold a paho client in a module and call it from LabVIEW like a small API. Install the library with pip install paho-mqtt in the same Python that LabVIEW uses.

import queue
import uuid
import paho.mqtt.client as mqtt

_client = None
_inbox = queue.Queue()
_station = ""

def _on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        return
    client.publish(f"testmqtt/labview/{_station}/status", "online", qos=1, retain=True)
    client.subscribe(f"testmqtt/labview/{_station}/cmd/#", qos=1)

def _on_message(client, userdata, msg):
    _inbox.put((msg.topic, msg.payload.decode("utf-8", "replace")))

def connect(host, port, username, password, station):
    global _client, _station
    _station = station
    _client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                          client_id=f"labview-{station}-{uuid.uuid4().hex[:6]}")
    if username:
        _client.username_pw_set(username, password)
    _client.will_set(f"testmqtt/labview/{station}/status", "offline", qos=1, retain=True)
    _client.on_connect = _on_connect
    _client.on_message = _on_message
    _client.reconnect_delay_set(min_delay=1, max_delay=30)
    _client.connect(host, int(port), keepalive=30)
    _client.loop_start()           # network thread: reconnects automatically
    return True

def publish(topic, payload, qos, retain):
    info = _client.publish(topic, payload, qos=int(qos), retain=bool(retain))
    return info.rc == mqtt.MQTT_ERR_SUCCESS

def poll(max_items):
    items = []
    while len(items) < int(max_items) and not _inbox.empty():
        items.append(_inbox.get_nowait())
    return items                   # list of (topic, payload) tuples

def disconnect():
    _client.publish(f"testmqtt/labview/{_station}/status", "offline", qos=1, retain=True).wait_for_publish(5)
    _client.disconnect()
    _client.loop_stop()
    return True

In LabVIEW:

• Open Python Session once at startup (version 3.x, matching LabVIEW's bitness) and keep the session reference in the MQTT loop's shift register.
• Call connect with the broker details, then call poll every 100 ms or so. Wire an array of clusters of two strings as the return type: each tuple maps to one cluster (topic, payload).
• Call publish with the JSON string from Flatten To JSON.
• On shutdown call disconnect, then Close Python Session.

paho runs its own network thread with automatic reconnect, so the LabVIEW side stays simple and never blocks on the network for long. The same module works for MQTT 5 if you pass protocol=mqtt.MQTTv5 to the client. More on the library itself in MQTT with Python.

Test the bench without hardware

Before connecting real instruments, check the MQTT side end to end. In one terminal (or the online client), watch everything the station publishes:

mosquitto_sub -h broker.connectmq.com -p 1883 -t 'testmqtt/labview/#' -v

In another, send a command as the MES would:

mosquitto_pub -h broker.connectmq.com -p 1883 -q 1 \
  -t 'testmqtt/labview/bench01/cmd/start' -m '{"serial":"SN-0001","recipe":"A"}'

You should see status = online, then a result message for SN-0001. Now kill the LabVIEW process (not a clean stop): within about 1.5 × the keep-alive interval the broker publishes the Will, and the status flips to offline. That one check proves your Will, retained status and reconnect logic work.

Watch your test stations in the browser

For automated regression tests, script the same steps: publish a command, wait for the result with a timeout and assert on its fields. The approach scales into full hardware-in-the-loop testing.

Common pitfalls

• Same client ID on two benches after copying a project: stations disconnect each other in a loop. Generate the ID from the station name plus a random suffix.
• Publishing doubles as a timing source: if results only go out when the network is up, the test log and the MQTT data disagree. Timestamp results when they are measured, not when they are published.
• Huge payloads: waveforms with millions of samples belong in files or a database; publish a summary and a link. Check message sizes against broker limits before you ship.
• Retained results: do not retain per-unit results, or every new subscriber receives a stale result for the last unit. Retain only state.

LabVIEW and MQTT fit well together once the network sits in its own loop: pick a library you can maintain, agree on topics and QoS up front, use a retained status with a Last Will, and test the whole path with a command line client before the first real unit goes on the bench. If MATLAB does the analysis downstream, see MQTT in MATLAB.

Frequently asked questions