# Getting Started (/connectors/mqtt/tutorials/getting-started)
Get a message flowing through the KubeMQ MQTT connector in minutes. You point a standard
MQTT client at the broker, subscribe to an `events/` topic filter, publish a message to a
matching topic, and watch it arrive — all over the native MQTT wire, with no KubeMQ SDK.
This walkthrough takes you from a running server to a verified pub/sub round-trip.
## Prerequisites [#prerequisites]
* A running **kubemq-server** with the MQTT connector **enabled** and reachable on **port 1883** (plain
TCP). The connector is **opt-in (disabled by default)** — see the enable step below.
* One of the MQTT clients below for your language (the examples pin a native client per
language — there is no KubeMQ SDK). For a quick smoke test, the `mosquitto_pub` /
`mosquitto_sub` command-line tools work too.
## Enable the connector [#enable-the-connector]
The MQTT connector is **disabled by default** — a stock kubemq-server does **not** bind the
MQTT listeners until you turn it on. Enable it with its enable variable:
**The enable variable is `CONNECTORSMQTT_ENABLE` — there is no underscore between
`CONNECTORS` and `MQTT`, and no `KUBEMQ_` prefix.** This is irregular: most other env vars
carry a separator. Variants like `CONNECTORS_MQTT_ENABLE` do **not** bind to the
`Connectors.MQTT.Enable` field and are silently ignored. For Kubernetes, set
`spec.mqtt.enabled: true` in the `KubemqCluster` CR.
Bring up a throwaway local broker with MQTT enabled:
Every example reads a single environment variable for the broker endpoint. The scheme
selects the transport — `tcp://` (1883), `tls://` (8883), or `ws://` (8083, path `/`):
```bash
# default: tcp://localhost:1883
export KUBEMQ_MQTT_URL="tcp://localhost:1883"
```
To **disable** MQTT again after enabling it, set its enable variable to `false`:
When `Enable` is `false`, no MQTT listener binds and the rest of the MQTT config is skipped. See
[Configuration](/connectors/mqtt/concepts/configuration) for the full settings list.
## How it works [#how-it-works]
A subscriber registers a topic filter; a publisher sends to a topic with the same prefix.
The connector resolves the topic prefix (`events/`) to a KubeMQ pattern, translates `/` to
`.` for the channel, and delivers every published message to matching subscribers.
*A publish to `events/demo/x` maps to the Events channel `demo.x`; the `+` wildcard filter `events/demo/+` matches it and the connector delivers the message to the subscriber.*
## Steps [#steps]
### Connect to the broker [#connect-to-the-broker]
Open an MQTT 5.0 connection to the endpoint in `KUBEMQ_MQTT_URL`. The connector accepts
MQTT 3.1.1 and 5.0 on the same listener; the examples use 5.0 so User-Properties carry
across as KubeMQ Tags.
The language tabs across all three steps run the **complete** round-trip from a single
program: connect a subscriber and a publisher, subscribe to `events/demo/+`, publish one
message to `events/demo/x`, and confirm the subscriber receives it.
```go
package main
import (
"context"
"fmt"
"log"
"net"
"os"
"strings"
"time"
"github.com/eclipse/paho.golang/paho"
)
func brokerURL() string {
if u := os.Getenv("KUBEMQ_MQTT_URL"); u != "" {
return u
}
return "tcp://localhost:1883"
}
func tcpAddr(rawURL string) string {
for _, pfx := range []string{"tcp://", "ws://", "tls://"} {
if strings.HasPrefix(rawURL, pfx) {
return rawURL[len(pfx):]
}
}
return rawURL
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
// Subscribe with the single-level '+' wildcard (KubeMQ '*'); publish to a
// concrete leaf it matches. "events/demo/x" maps to channel "demo.x".
const subFilter = "events/demo/+"
const pubTopic = "events/demo/x"
addr := tcpAddr(brokerURL())
recvCh := make(chan string, 1)
// SUBSCRIBER — connect and subscribe before publishing.
subConn, err := net.Dial("tcp", addr)
if err != nil {
log.Fatalf("sub dial: %v", err)
}
subClient := paho.NewClient(paho.ClientConfig{
Conn: subConn,
OnPublishReceived: []func(paho.PublishReceived) (bool, error){
func(pr paho.PublishReceived) (bool, error) {
recvCh <- string(pr.Packet.Payload)
return true, nil
},
},
})
if _, err := subClient.Connect(ctx, &paho.Connect{ClientID: "go-sub", KeepAlive: 30, CleanStart: true}); err != nil {
log.Fatalf("sub connect: %v", err)
}
subAck, err := subClient.Subscribe(ctx, &paho.Subscribe{
Subscriptions: []paho.SubscribeOptions{{Topic: subFilter, QoS: 1}},
})
if err != nil {
log.Fatalf("subscribe: %v", err)
}
// Reason code > 2 is a rejection (0xA2 wildcard-not-supported, 0x83 impl-specific).
if subAck.Reasons[0] > 2 {
log.Fatalf("SUBACK reason=0x%02X", subAck.Reasons[0])
}
fmt.Printf("[sub] subscribed to %q\n", subFilter)
time.Sleep(300 * time.Millisecond) // let the subscription register
// PUBLISHER — connect and publish one QoS-1 message.
pubConn, err := net.Dial("tcp", addr)
if err != nil {
log.Fatalf("pub dial: %v", err)
}
pubClient := paho.NewClient(paho.ClientConfig{Conn: pubConn})
if _, err := pubClient.Connect(ctx, &paho.Connect{ClientID: "go-pub", KeepAlive: 30, CleanStart: true}); err != nil {
log.Fatalf("pub connect: %v", err)
}
if _, err := pubClient.Publish(ctx, &paho.Publish{Topic: pubTopic, QoS: 1, Payload: []byte("hello")}); err != nil {
log.Fatalf("publish: %v", err)
}
fmt.Printf("[pub] published to %q\n", pubTopic)
// RECEIVE — wait for the subscriber to get the message.
select {
case payload := <-recvCh:
fmt.Printf("[sub] received: %s (channel demo.x)\n", payload)
case <-ctx.Done():
log.Fatal("timed out waiting for the event")
}
_ = subClient.Disconnect(&paho.Disconnect{ReasonCode: 0})
_ = pubClient.Disconnect(&paho.Disconnect{ReasonCode: 0})
}
```
```python
import os
import threading
import time
import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion
def parse_url(url: str) -> tuple[str, int]:
scheme, rest = url.split("://", 1)
host_port = rest.rstrip("/")
if ":" in host_port:
host, port = host_port.rsplit(":", 1)
return host, int(port)
return host_port, {"tcp": 1883, "tls": 8883, "ws": 8083}.get(scheme, 1883)
def main() -> None:
host, port = parse_url(os.environ.get("KUBEMQ_MQTT_URL", "tcp://localhost:1883"))
# Subscribe with the single-level "+" wildcard; publish to a matching leaf.
sub_topic = "events/demo/+" # KubeMQ channel demo.*
pub_topic = "events/demo/x" # KubeMQ channel demo.x
received = threading.Event()
# SUBSCRIBER
sub = mqtt.Client(callback_api_version=CallbackAPIVersion.VERSION2,
client_id="python-sub", protocol=mqtt.MQTTv5)
def on_connect(client, userdata, flags, rc, props):
client.subscribe(sub_topic, qos=1)
def on_message(client, userdata, msg):
print(f"[sub] received: {msg.payload.decode()!r} (channel demo.x)")
received.set()
sub.on_connect = on_connect
sub.on_message = on_message
sub.connect(host, port, keepalive=30, clean_start=True)
sub.loop_start()
time.sleep(0.5) # let the subscription register
# PUBLISHER
pub = mqtt.Client(callback_api_version=CallbackAPIVersion.VERSION2,
client_id="python-pub", protocol=mqtt.MQTTv5)
pub.connect(host, port, keepalive=30, clean_start=True)
pub.loop_start()
pub.publish(pub_topic, payload=b"hello", qos=1).wait_for_publish(timeout=10)
print(f"[pub] published to {pub_topic!r}")
if not received.wait(timeout=10):
raise TimeoutError("timed out waiting for the event")
pub.loop_stop(); pub.disconnect()
sub.loop_stop(); sub.disconnect()
if __name__ == "__main__":
main()
```
```java
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
public final class Main {
public static void main(String[] args) throws Exception {
String broker = System.getenv().getOrDefault("KUBEMQ_MQTT_URL", "tcp://localhost:1883");
String subTopic = "events/demo/+"; // single-level '+' wildcard
String pubTopic = "events/demo/x"; // KubeMQ channel demo.x
MqttConnectionOptions opts = new MqttConnectionOptions();
opts.setCleanStart(true);
opts.setKeepAliveInterval(30);
CountDownLatch received = new CountDownLatch(1);
// SUBSCRIBER
MqttAsyncClient sub = new MqttAsyncClient(broker,
"java-sub-" + UUID.randomUUID().toString().substring(0, 8), new MemoryPersistence());
sub.setCallback(new MqttCallback() {
@Override public void messageArrived(String topic, MqttMessage msg) {
System.out.printf("[sub] received: %s (channel demo.x)%n",
new String(msg.getPayload(), StandardCharsets.UTF_8));
received.countDown();
}
@Override public void disconnected(MqttDisconnectResponse r) { }
@Override public void mqttErrorOccurred(MqttException e) { }
@Override public void deliveryComplete(IMqttToken t) { }
@Override public void connectComplete(boolean reconnect, String uri) { }
@Override public void authPacketArrived(int code, MqttProperties p) { }
});
sub.connect(opts).waitForCompletion(10_000);
sub.subscribe(subTopic, 1).waitForCompletion(10_000);
System.out.printf("[sub] subscribed to '%s'%n", subTopic);
// PUBLISHER
MqttAsyncClient pub = new MqttAsyncClient(broker,
"java-pub-" + UUID.randomUUID().toString().substring(0, 8), new MemoryPersistence());
pub.connect(opts).waitForCompletion(10_000);
MqttMessage msg = new MqttMessage("hello".getBytes(StandardCharsets.UTF_8));
msg.setQos(1);
pub.publish(pubTopic, msg).waitForCompletion(10_000);
System.out.printf("[pub] published to '%s'%n", pubTopic);
if (!received.await(10, TimeUnit.SECONDS)) {
throw new IllegalStateException("timed out waiting for the event");
}
pub.disconnect().waitForCompletion(5_000); pub.close();
sub.disconnect().waitForCompletion(5_000); sub.close();
}
}
```
```typescript
import mqtt, { type MqttClient } from "mqtt";
function brokerUrl(): string {
return process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883";
}
async function main(): Promise {
const url = brokerUrl();
const subFilter = "events/demo/+"; // single-level '+' wildcard
const pubTopic = "events/demo/x"; // KubeMQ channel demo.x
// SUBSCRIBER
const subscriber: MqttClient = mqtt.connect(url, {
clientId: "js-sub",
protocolVersion: 5,
clean: true,
});
await new Promise((resolve, reject) => {
subscriber.once("connect", () => resolve());
subscriber.once("error", reject);
});
const received = new Promise((resolve) => {
subscriber.on("message", (topic, payload) => {
console.log(`[sub] received: ${payload.toString()} (channel demo.x)`);
resolve();
});
});
await new Promise((resolve, reject) => {
subscriber.subscribe(subFilter, { qos: 1 }, (err) => (err ? reject(err) : resolve()));
});
console.log(`[sub] subscribed to ${subFilter}`);
await new Promise((r) => setTimeout(r, 300));
// PUBLISHER
const publisher: MqttClient = mqtt.connect(url, {
clientId: "js-pub",
protocolVersion: 5,
clean: true,
});
await new Promise((resolve, reject) => {
publisher.once("connect", () => resolve());
publisher.once("error", reject);
});
await new Promise((resolve, reject) => {
publisher.publish(pubTopic, "hello", { qos: 1, retain: false }, (err) =>
err ? reject(err) : resolve());
});
console.log(`[pub] published to ${pubTopic}`);
await received;
await publisher.endAsync();
await subscriber.endAsync();
}
main().catch((err) => {
console.error(err);
process.exit(1);
});
```
```csharp
using System.Text;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
static (string host, int port) ParseEndpoint(string url)
{
foreach (var prefix in new[] { "tcp://", "tls://", "ws://" })
if (url.StartsWith(prefix, StringComparison.OrdinalIgnoreCase))
url = url[prefix.Length..];
var parts = url.TrimEnd('/').Split(':');
return (parts[0], parts.Length > 1 ? int.Parse(parts[1]) : 1883);
}
var url = Environment.GetEnvironmentVariable("KUBEMQ_MQTT_URL") ?? "tcp://localhost:1883";
var (host, port) = ParseEndpoint(url);
const string subTopic = "events/demo/+"; // single-level '+' wildcard
const string pubTopic = "events/demo/x"; // KubeMQ channel demo.x
var factory = new MqttFactory();
var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
// SUBSCRIBER
using var subClient = factory.CreateMqttClient();
subClient.ApplicationMessageReceivedAsync += e =>
{
received.TrySetResult(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment));
return Task.CompletedTask;
};
var subOptions = new MqttClientOptionsBuilder()
.WithTcpServer(host, port)
.WithClientId("csharp-sub")
.WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
.WithCleanSession(true)
.Build();
await subClient.ConnectAsync(subOptions);
await subClient.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter(f => f.WithTopic(subTopic).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
.Build());
Console.WriteLine($"[sub] subscribed to '{subTopic}'");
await Task.Delay(300);
// PUBLISHER
using var pubClient = factory.CreateMqttClient();
var pubOptions = new MqttClientOptionsBuilder()
.WithTcpServer(host, port)
.WithClientId("csharp-pub")
.WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
.WithCleanSession(true)
.Build();
await pubClient.ConnectAsync(pubOptions);
await pubClient.PublishAsync(new MqttApplicationMessageBuilder()
.WithTopic(pubTopic)
.WithPayload(Encoding.UTF8.GetBytes("hello"))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
.Build());
Console.WriteLine($"[pub] published to '{pubTopic}'");
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var body = await received.Task.WaitAsync(cts.Token);
Console.WriteLine($"[sub] received: {body} (channel demo.x)");
await subClient.DisconnectAsync();
await pubClient.DisconnectAsync();
```
```ruby
# The Ruby `mqtt` gem speaks MQTT 3.1.1 only.
require "mqtt"
require "uri"
require "timeout"
uri = URI.parse(ENV.fetch("KUBEMQ_MQTT_URL", "tcp://localhost:1883"))
conn = { host: uri.host, port: uri.port, ssl: uri.scheme == "tls" }
sub_topic = "events/demo/+" # single-level '+' wildcard
pub_topic = "events/demo/x" # KubeMQ channel demo.x
received = Queue.new
sub_ready = Queue.new
# SUBSCRIBER thread
subscriber = Thread.new do
MQTT::Client.connect(**conn, client_id: "ruby-sub", clean_session: true) do |client|
client.subscribe([sub_topic, 1])
sub_ready.push(:ready)
topic, payload = client.get
received.push(payload)
end
end
sub_ready.pop
puts "[sub] subscribed to '#{sub_topic}'"
# PUBLISHER
MQTT::Client.connect(**conn, client_id: "ruby-pub", clean_session: true) do |client|
# retain=false is mandatory — the broker silently drops retained publishes.
client.publish(pub_topic, "hello", false, 1)
puts "[pub] published to '#{pub_topic}'"
end
payload = Timeout.timeout(10) { received.pop }
puts "[sub] received: #{payload.inspect} (channel demo.x)"
subscriber.kill
```
```rust
use rumqttc::v5::mqttbytes::v5::Packet;
use rumqttc::v5::mqttbytes::QoS;
use rumqttc::v5::{AsyncClient, Event, MqttOptions};
use std::env;
use std::time::Duration;
use tokio::sync::oneshot;
use tokio::time::timeout;
fn parse_host_port(url: &str) -> (String, u16) {
let stripped = url
.trim_start_matches("tcp://")
.trim_start_matches("tls://")
.trim_start_matches("ws://");
let host_port = stripped.split('/').next().unwrap_or(stripped);
let mut parts = host_port.splitn(2, ':');
let host = parts.next().unwrap_or("localhost").to_string();
let port: u16 = parts.next().and_then(|p| p.parse().ok()).unwrap_or(1883);
(host, port)
}
#[tokio::main]
async fn main() -> Result<(), Box> {
let url = env::var("KUBEMQ_MQTT_URL").unwrap_or_else(|_| "tcp://localhost:1883".to_string());
let (host, port) = parse_host_port(&url);
let sub_topic = "events/demo/+"; // single-level '+' wildcard
let pub_topic = "events/demo/x"; // KubeMQ channel demo.x
// SUBSCRIBER
let mut sub_opts = MqttOptions::new("rust-sub", &host, port);
sub_opts.set_keep_alive(Duration::from_secs(30));
let (sub_client, mut sub_loop) = AsyncClient::new(sub_opts, 10);
let (ready_tx, ready_rx) = oneshot::channel::<()>();
let (msg_tx, msg_rx) = oneshot::channel::();
tokio::spawn(async move {
let mut ready = Some(ready_tx);
let mut msg = Some(msg_tx);
loop {
match sub_loop.poll().await {
Ok(Event::Incoming(Packet::SubAck(_))) => {
if let Some(tx) = ready.take() { let _ = tx.send(()); }
}
Ok(Event::Incoming(Packet::Publish(p))) => {
if let Some(tx) = msg.take() {
let _ = tx.send(String::from_utf8_lossy(&p.payload).to_string());
}
return;
}
Ok(_) => {}
Err(_) => return,
}
}
});
sub_client.subscribe(sub_topic, QoS::AtLeastOnce).await?;
timeout(Duration::from_secs(5), ready_rx).await??;
println!("[sub] subscribed to '{sub_topic}'");
// PUBLISHER
let mut pub_opts = MqttOptions::new("rust-pub", &host, port);
pub_opts.set_keep_alive(Duration::from_secs(30));
let (pub_client, mut pub_loop) = AsyncClient::new(pub_opts, 10);
tokio::spawn(async move { while pub_loop.poll().await.is_ok() {} });
// retain = false — a retained publish is silently dropped.
pub_client.publish(pub_topic, QoS::AtLeastOnce, false, b"hello".as_ref()).await?;
println!("[pub] published to '{pub_topic}'");
let payload = timeout(Duration::from_secs(10), msg_rx).await??;
println!("[sub] received: {payload} (channel demo.x)");
Ok(())
}
```
### Publish a message [#publish-a-message]
The publisher in the program above sends one QoS-1 message to `events/demo/x`. The prefix
`events/` selects the Events pattern, and the remaining segments become the KubeMQ channel
with `/` translated to `.` — so `events/demo/x` lands on channel `demo.x`. QoS 1 returns a
PUBACK so you know the broker accepted the publish. **Never set the retain flag** — a
retained publish returns PUBACK `0x00` but the message is silently dropped.
### Subscribe and verify [#subscribe-and-verify]
The subscriber registers the filter `events/demo/+`. The single-level `+` wildcard maps to
KubeMQ `*`, matching any one trailing segment — so it receives the publish to
`events/demo/x`. When the message arrives the program prints it and exits:
```text
[sub] subscribed to 'events/demo/+'
[pub] published to 'events/demo/x'
[sub] received: hello (channel demo.x)
```
Events is fire-and-forget pub/sub: subscribe **before** you publish, or the message is
gone. For persistence and replay-on-reconnect, use the Events-Store pattern (`store/`)
instead.
Wildcards (`+` → `*`, `#` → `>`) are accepted on the **Events** pattern only — a wildcard
subscribe on any other prefix returns SUBACK `0xA2`. Avoid literal `.` in topic segments:
`events/a.b/c` and `events/a/b/c` both map to channel `a.b.c`. See
[Topic mapping](/connectors/mqtt/concepts/topic-mapping).
## Next steps [#next-steps]