# Topic wildcards

## Overview

In MQTT, wildcards provide a powerful mechanism for subscribing to multiple topics simultaneously. When a client subscribes to a topic, it can either subscribe to the exact topic of a published message or utilise wildcards to broaden its subscription.

### How are wildcards used?

* Clients can subscribe to a wildcard topic to receive messages from multiple matching topics. 
* Wildcards can reduce overhead by eliminating the need to subscribe to each topic individually.
* Wildcards are used when there is uncertainty about the topics that publishing clients will use.

## Single level wildcard

Taking advantage of MQTT wildcard capabilities is through the use of a custom event parser. This section will walk you through the process.

### Example

Events will be received from various rooms within a house using the follow example topic structure:

```bash
sensors/kitchen/temperaturehumidity
sensors/livingroom/temperaturehumidity
sensors/hallway/temperaturehumidity
sensors/bedroom1/temperaturehumidity
sensors/bedroom2/temperaturehumidity
```

And the subscription will use the single level wildcard method '+':

```
sensors/+/temperaturehumidity
```

{% tabs %}
{% tab title="DSL" %}
We shall use the '+' MQTT wildcard to recieve all sensor data from each connected room.

```yaml
mqttConsumer:
  broker: tcp://127.0.0.1:1883
  topic: sensors/groundfloor/+/temperaturehumidity

  clientId: tempHumidyManagementProcessor
  qos: 1

  deserializer:
    parser: com.fractalworks.mqtt.example.TemperatureHumiditySensorParser
    compressed: false
    batch: false
```

{% endtab %}

{% tab title="Parser" %}
This custom parser 

```java
public class TemperatureHumiditySensorParser implements StreamEventParser<List<StreamEvent>> {

    private String ROOM = "room";
    private ObjectMapper objectMapper;

    public TemperatureHumiditySensorParser() {
        // REQUIRED
    }

    @Override
    public void initialise() throws StreamsException {
        objectMapper = new ObjectMapper();
        objectMapper.configure(SerializationFeature.INDENT_OUTPUT, false);
        objectMapper.configure(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS, true);
        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_EMPTY);
    }

    @Override
    public List<StreamEvent> translate(Object payload) throws TranslationException {
        TemperatureHumidityEvent event = parseSingleEvent(payload);
        if(event == null){
            throw new TranslationException("Expected TemperatureHumidityEvent type to parse.");
        }
        return Collections.singletonList(event.toStreamEvent());
    }

    @Override
    public List<StreamEvent> translate(Object payload, String source) throws TranslationException {
        // First parse the event to the standard data type
        List<StreamEvent> events = translate(payload);

        // Now decorate the parsed events with room identifier
        String[] tokens = source.split("/");
        if(tokens.length != 4){
            // Now decorate the event with the source information
            events.forEach(event -> {
                event.addValue(ROOM, tokens[3]);
            });
        }
        return events;
    }

    /**
     * Parse object to a single TemperatureHumidityEvent
     *
     * @param obj
     * @return
     */
    private TemperatureHumidityEvent parseSingleEvent(final Object obj) {
        TemperatureHumidityEvent event = null;
        try {
            if (obj instanceof String str) {
                event = objectMapper.readValue(str, TemperatureHumidityEvent.class);
            } else if (obj instanceof byte[] byteArray) {
                event = objectMapper.readValue(byteArray, TemperatureHumidityEvent.class);
            }
        } catch (IOException e) {
            // Consume exception
        }
        return event;
    }

    /**
     * Parse object to multiple stream events
     *
     * @param obj
     * @return
     */
    private List<TemperatureHumidityEvent> parseMultipleEvents(final Object obj) {
        List<TemperatureHumidityEvent> events = null;
        try {
            if (obj instanceof String str) {
                events = Arrays.asList(objectMapper.readValue(str, TemperatureHumidityEvent[].class));
            } else if (obj instanceof byte[] byteArray) {
                events = Arrays.asList(objectMapper.readValue(byteArray, TemperatureHumidityEvent[].class));
            }
        } catch (IOException e) {
            // Consume exception
        }
        return events;
    }
}
```

{% endtab %}
{% endtabs %}

## Multi level wildcard

Nothing really changes much when using this wildcard except that you may want to change how you parse the topic string.

### Example

This example would now receive all sensor information from a house. The parser would need to coded such that events can be handled holistically within the target stream processor

```
// Some code
mqttConsumer:
  broker: tcp://127.0.0.1:1883
  topic: sensors/#

  clientId: tempHumidyManagementProcessor
  qos: 1

  deserializer:
    parser: com.fractalworks.mqtt.example.TemperatureHumiditySensorParser
    compressed: false
    batch: false
```

#### Notes

See the great documentation HiveMQ has produced that covers this [subject](https://www.hivemq.com/blog/mqtt-essentials-part-5-mqtt-topics-best-practices/).