HiveMQ MQTT Client

The HiveMQ MQTT Client is a high-performance library supporting MQTT 3.1.1 and 5.0 with the following key capabilities:

• Backpressure Support: Supports QoS 1 and 2, and QoS 0 (via dropping incoming messages). It integrates MQTT flow control with reactive pull backpressure.
• Transports: Supports TCP, SSL/TLS (up to TLS 1.3, including mutual authentication, SNI, and session resumption), WebSockets, and Secure WebSockets.
• Proxy Support: SOCKS4, SOCKS5, and HTTP CONNECT.
• Reliability: Automatic and configurable thread management, reconnect handling, message redelivery, and automatic resubscription if a session expires.
• Manual Acknowledgment: Allows selective manual acknowledgment for specific streams. The client ensures MQTT-compliant ordering of acknowledgments regardless of the order in which they are received.
• MQTT 5 Specifics: Pluggable Enhanced Authentication, automatic Topic Alias mapping, and interceptors for QoS flows.

The Async API provides non-blocking operations that return CompletableFuture. You can initialize it via .buildAsync() or switch using .toAsync().

Common Operations

• Connect/Publish/Subscribe/Unsubscribe/Disconnect: These methods are analogous to the Blocking API but return CompletableFuture instead of blocking.
• Per-Subscribe Consumption: You can provide a callback directly during subscription:

  client.subscribeWith()
        .topicFilter("test/topic")
        .callback(System.out::println)
        .send();

• Global Consumption:

  client.publishes(MqttGlobalPublishFilter.ALL, System.out::println);

Mqtt5AsyncClient client = Mqtt5Client.builder()
        .identifier(UUID.randomUUID().toString())
        .serverHost("broker.hivemq.com")
        .buildAsync();

client.connect()
        .thenCompose(connAck -> client.publishWith().topic("test/topic").payload("1".getBytes()).send())
        .thenCompose(publishResult -> client.disconnect());

The Reactive API uses RxJava types to handle MQTT streams. It is initialized via .buildRx() or switched via .toRx().

Key Reactive Types

• Single<T>: Represents a single asynchronous value (e.g., connect() returns Single<Mqtt5ConnAck>).
• Flowable<T>: Represents an asynchronous stream of values (e.g., publishes() returns a Flowable).
• Completable: Represents an asynchronous operation that completes without emitting a value (e.g., disconnect()).
• FlowableWithSingle<T, R>: A combination of a single value and a stream (e.g., subscribeStreamWith() returns a FlowableWithSingle containing the SubAck and the matching Publish messages).

Subscription Patterns

• Stream Subscription: Use subscribeStreamWith() to get a stream of messages directly.
• Global Stream: client.publishes(MqttGlobalPublishFilter.ALL) returns a Flowable of all incoming messages.

Mqtt5RxClient client = Mqtt5Client.builder()
        .identifier(UUID.randomUUID().toString())
        .serverHost("broker.hivemq.com")
        .buildRx();

// Example: Connecting and then subscribing to a stream
Completable connectScenario = client.connect()
        .doOnSuccess(connAck -> System.out.println("Connected"))
        .ignoreElement();

FlowableWithSingle<Mqtt5Publish, Mqtt5SubAck> subAckAndMatchingPublishes = client.subscribeStreamWith()
        .topicFilter("a/b/c").qos(MqttQos.AT_LEAST_ONCE)
        .applySubscribe();

Completable subscribeScenario = subAckAndMatchingPublishes
        .doOnNext(publish -> System.out.println("Received: " + publish.getTopic()))
        .ignoreElements();

connectScenario.andThen(subscribeScenario).blockingAwait();

The Blocking API is used for synchronous MQTT operations. You can initialize it via the builder using .buildBlocking() or switch from another client using .toBlocking().

Common Operations

• Connect: client.connect() or client.connectWith()...send().
• Publish: client.publishWith()...send() or client.publish(Mqtt5Publish).
• Subscribe: client.subscribeWith()...send() or client.subscribe(Mqtt5Subscribe).
• Unsubscribe: client.unsubscribeWith()...send() or client.unsubscribe(Mqtt5Unsubscribe).
• Disconnect: client.disconnect() or client.disconnectWith()...send().
• Consume Messages: Use client.publishes(MqttGlobalPublishFilter.ALL) to get a Mqtt5Publishes object. Call this before subscribing to ensure no messages are lost.

final Mqtt5BlockingClient client = Mqtt5Client.builder()
        .identifier(UUID.randomUUID().toString())
        .serverHost("broker.hivemq.com")
        .buildBlocking();

client.connect();

try (final Mqtt5Publishes publishes = client.publishes(MqttGlobalPublishFilter.ALL)) {
    client.subscribeWith().topicFilter("test/topic").qos(MqttQos.AT_LEAST_ONCE).send();

    // Receive with timeout
    publishes.receive(1, TimeUnit.SECONDS).ifPresent(System.out::println);
} finally {
    client.disconnect();
}