# Topic Grammar (/connectors/mqtt/reference/topic-grammar)



The KubeMQ MQTT connector maps every MQTT topic (on publish) and filter (on subscribe) to
one of five KubeMQ messaging patterns. The first segment of the topic string is the
**pattern prefix**; the remaining segments become the **KubeMQ channel**.

<Callout type="info">
  **The Ruby client ships the MQTT v3.1.1 subset.** The `mqtt` gem speaks MQTT 3.1.1 only, so
  it cannot use RPC (`commands/`, `queries/`) or `$share` queue consumption — both require
  MQTT 5.0. Use Ruby for Events, Events-Store, and queue **produce** only.
</Callout>

## Pattern prefix table [#pattern-prefix-table]

| MQTT topic / filter          | KubeMQ pattern        | Direction                            | Example topic                 | KubeMQ channel |
| ---------------------------- | --------------------- | ------------------------------------ | ----------------------------- | -------------- |
| `events/<ch>`                | Events                | publish + subscribe                  | `events/site1/temp`           | `site1.temp`   |
| `store/<ch>`                 | Events-Store          | publish + subscribe (start-new-only) | `store/site1/temp`            | `site1.temp`   |
| `queues/<ch>`                | Queues — produce only | publish                              | `queues/jobs/email`           | `jobs.email`   |
| `$share/<group>/queues/<ch>` | Queues — consume      | subscribe (QoS ≥ 1)                  | `$share/g1/queues/jobs/email` | `jobs.email`   |
| `commands/<ch>`              | Commands (RPC send)   | publish — MQTT 5.0 only              | `commands/svc/reboot`         | `svc.reboot`   |
| `queries/<ch>`               | Queries (RPC send)    | publish — MQTT 5.0 only              | `queries/svc/status`          | `svc.status`   |
| `$reply/<clientID>/<suffix>` | local (not routed)    | subscribe + RPC ResponseTopic        | `$reply/c1/inbox`             | not routed     |
| *(no prefix)*                | DefaultPattern        | publish + subscribe                  | `site1/temp`                  | `site1.temp`   |

A prefixless topic is routed to `DefaultPattern` (`events` by default). When
`DefaultPattern=none` the connector rejects the message: PUBACK `0x90` on publish, SUBACK
`0x8F` on subscribe.

## Path-separator mapping: `/` becomes `.` [#path-separator-mapping--becomes-]

Topic segments are joined with `.` to form the KubeMQ channel:

```text
events/site1/factory/temp  →  channel: site1.factory.temp
store/orders/europe        →  channel: orders.europe
queues/jobs/email          →  channel: jobs.email
```

The connector splits the non-prefix segments on `/` and joins them with `.`.

### Gotcha — literal `.` in a segment conflates with `/` [#gotcha--literal--in-a-segment-conflates-with-]

A literal `.` inside a segment passes through unchanged, making the channel
indistinguishable from one produced by an extra `/`:

```text
events/a.b/c   →  channel: a.b.c
events/a/b/c   →  channel: a.b.c   ← same result
```

These two topics map to **identical** KubeMQ channels. This is a **lossy round-trip**: the
reverse mapping converts `.` back to `/`, so the injected delivery topic is `events/a/b/c`
regardless of which form was published. Avoid dots inside individual path segments to
prevent ambiguity.

## Wildcard subscriptions [#wildcard-subscriptions]

MQTT wildcards are translated to KubeMQ wildcard syntax **only on Events subscriptions**.

| MQTT wildcard | KubeMQ equivalent  | Position rule      |
| ------------- | ------------------ | ------------------ |
| `+`           | `*` (single-level) | any segment        |
| `#`           | `>` (multi-level)  | final segment only |

Examples:

| MQTT subscribe filter | KubeMQ channel filter | Matches                                  |
| --------------------- | --------------------- | ---------------------------------------- |
| `events/+/temp`       | `*.temp`              | `events/site1/temp`, `events/site2/temp` |
| `events/site1/#`      | `site1.>`             | `events/site1/temp`, `events/site1/a/b`  |
| `events/#`            | `>`                   | all events channels                      |

**Non-Events wildcards are rejected**: subscribing to `queues/+/foo` or `store/#` returns
SUBACK `0xA2` (wildcard-subscriptions-not-supported).

### Gotcha — overlapping wildcard filters deliver multiple copies [#gotcha--overlapping-wildcard-filters-deliver-multiple-copies]

Each distinct subscribe filter creates an independent bridge registry entry (one
subscription per `(pattern, filter)` pair). A publish that matches **N** overlapping filters
fires **N** deliveries — one per matching entry. There is **no cross-entry deduplication**
in the connector.

Example: a client subscribes to both `events/#` and `events/site1/#`; a publish to
`events/site1/temp` matches both entries and the client receives **two** copies.

## Shared subscriptions — queue consume [#shared-subscriptions--queue-consume]

Queue consumption uses MQTT 5.0 shared subscriptions:

```text
$share/<group>/queues/<channel>
```

Rules:

| Condition                                   | Result                              |
| ------------------------------------------- | ----------------------------------- |
| `$share/<group>/queues/<ch>` at QoS ≥ 1     | SUBACK `0x00` / `0x01` (granted)    |
| `$share/<group>/queues/<ch>` at QoS 0       | SUBACK `0x83` (impl-specific error) |
| Plain `queues/<ch>` subscribe (no `$share`) | SUBACK `0x83`                       |
| `$share/<group>` on any non-queues prefix   | SUBACK `0x83`                       |
| Empty group: `$share//queues/<ch>`          | SUBACK `0x8F` (invalid)             |

**QoS 2 is downgraded to QoS 1**: the queue bridge operates at QoS 1 (PUBACK = acknowledge;
no QoS 2 two-phase commit in the bridge).

### Gotcha — the `$share` group name is audit-only; all groups compete in one pool [#gotcha--the-share-group-name-is-audit-only-all-groups-compete-in-one-pool]

The KubeMQ queue is a **single shared pool**. Unlike standard MQTT brokers — which deliver a
separate copy per `$share` group — the KubeMQ MQTT connector routes **all** `$share` group
subscribers into one competing-consumer pool. Every message is consumed exactly once across
all subscribers, regardless of group name. The group name is recorded in audit and metrics
only.

## RPC topics — MQTT 5.0 only [#rpc-topics--mqtt-50-only]

Commands and Queries are **publish-only** patterns over MQTT:

```text
commands/<channel>   # send a command RPC
queries/<channel>    # send a query RPC
```

### Gotcha — RPC requires MQTT 5.0; a v3.1.1 publish is silently dropped [#gotcha--rpc-requires-mqtt-50-a-v311-publish-is-silently-dropped]

A v3.1.1 (or earlier) client publishing to `commands/` or `queries/` receives a **success
PUBACK** (`0x00`), but the message is **silently dropped** and a `publish.error` audit event
is emitted. No RPC is issued.

MQTT 5.0 is required because:

* The `ResponseTopic` property carries the reply address.
* The `CorrelationData` property associates the response.
* User Properties carry RPC metadata.

### RPC subscribe is not allowed for MQTT clients [#rpc-subscribe-is-not-allowed-for-mqtt-clients]

MQTT clients **cannot** subscribe to `commands/<ch>` or `queries/<ch>` as responders. These
subscriptions return SUBACK `0x83`. RPC responders must be implemented on the gRPC side
using the KubeMQ Go SDK or another KubeMQ client.

### Reply namespace [#reply-namespace]

```text
$reply/<own-clientID>/<suffix>
```

* A client must **subscribe** to its own `$reply/<clientID>/<suffix>` before issuing an RPC
  publish.
* The `ResponseTopic` property of the RPC PUBLISH **must** point to the client's own
  `$reply/<clientID>/...` namespace. Using another client's `$reply` namespace returns
  PUBACK `0x83`.
* `$reply` topics are **local** — they are never routed to the broker.

### Gotcha — PUBACK is sent immediately; the client must implement its own timeout [#gotcha--puback-is-sent-immediately-the-client-must-implement-its-own-timeout]

The PUBACK for an RPC PUBLISH is sent **immediately on receipt**, before the response
arrives. The response is later injected as a separate PUBLISH on the `$reply` topic. The
client must implement its own timeout to detect missing responses. `RpcTimeoutSeconds`
(default 30 s) is the server-side bound; after that the server audits `rpc.timeout` and
discards the pending entry, but the client receives no notification.

## Empty segments and structural errors [#empty-segments-and-structural-errors]

The connector rejects topics with empty segments (leading slash, trailing slash, or double
slash):

| Offending topic                     | Error                        | Wire code                     |
| ----------------------------------- | ---------------------------- | ----------------------------- |
| `/events/foo`                       | empty segment (leading `/`)  | PUBACK `0x90` / SUBACK `0x8F` |
| `events//foo`                       | empty segment (double `/`)   | PUBACK `0x90` / SUBACK `0x8F` |
| `events/foo/`                       | empty segment (trailing `/`) | PUBACK `0x90` / SUBACK `0x8F` |
| `events/` (prefix only, no channel) | empty channel                | PUBACK `0x90` / SUBACK `0x8F` |

## Reverse mapping: delivery topics [#reverse-mapping-delivery-topics]

When the connector injects a delivery back to subscribing MQTT clients it replaces every
`.` with `/` and prepends the pattern prefix:

```text
channel: site1.factory.temp  +  pattern: events  →  events/site1/factory/temp
channel: jobs.email          +  pattern: queues  →  queues/jobs/email
```

For wildcard subscriptions the injected topic is the **concrete per-message channel** (e.g.
`events/site1/temp`), not the wildcard filter (e.g. `events/+/temp`). For exact
subscriptions the injected topic is the original filter string.

## Summary [#summary]

```text
MQTT publish/subscribe topic
        │
        ├── starts with "events/"   → Events
        ├── starts with "store/"    → Events-Store
        ├── starts with "queues/"   → Queues (produce)
        ├── starts with "commands/" → Commands (MQTT 5.0 only)
        ├── starts with "queries/"  → Queries  (MQTT 5.0 only)
        ├── starts with "$share/<g>/queues/"
        │                           → Queues (consume, QoS ≥ 1)
        ├── starts with "$reply/<own-clientID>/"
        │                           → local (no broker routing)
        └── no prefix               → DefaultPattern (default: events)
                                      DefaultPattern=none → rejected
```

## Related [#related]

<Cards>
  <Card title="Capabilities" href="/connectors/mqtt/reference/capabilities" description="Forced capabilities, protocol-version support, and the connector's limitations." />

  <Card title="Reason Codes" href="/connectors/mqtt/reference/reason-codes" description="The wire codes (0x8F, 0x90, 0xA2, 0x83) every grammar rule above returns." />

  <Card title="Connections Endpoint" href="/connectors/mqtt/reference/connections-endpoint" description="See the resolved pattern and channel for every live subscription." />
</Cards>