hato

Hato

An MQTT client for Java.

Easy on your application, from its handlers to its tests.

Add to 
implementation("org.mqtt4j:hato-client:0.9.0")
Protocol
MQTT 5.0, 3.1.1
Transports
TCP, TLS, WebSocket
Java
25 or later
Concurrency
Virtual threads
Dependencies
None
Reflection
None

Write message handlers like the rest of your code.

Each subscription hands its messages to your handler away from the network threads, so the handler can block, call a database, or publish and wait while the connection and other subscriptions keep going.

  • Reconnects and subscribes again by itself.
  • Join a result, or chain the next step.
  • A client you can call from any thread.
Handler 
client.subscribe(
        Subscribe.of(new TopicFilter("orders/+"), Qos.AT_LEAST_ONCE),
        delivery -> {
            Order order = Order.parse(delivery.message().payload());
            orders.save(order);
            client.publish(confirmation(order)).join();
        }
);

One message at a time, until you need more.

The defaults suit most applications, with nothing to tune. When a subscription needs more throughput, choose how its messages reach your code instead of building thread pools.

Running your code apart from the network has a price. Hato keeps it small, under a microsecond for each QoS 0 message it receives.

Serial Default
One at a time, in arrival order.
Striped
In order per key, parallel across keys.
Parallel
As many at once as you allow.
Hand-off
Straight into your queue, in order.

Handlers can run on your own executor, too.

Rehearse the bad days in your tests.

Test how your application handles interrupted connections and timeouts. The test kit runs on the same API you ship.

  • Unit-test handlers with fakes, without a broker.
  • Start an embedded broker for each test.
  • Cut the connection at any moment.
  • Delay incoming or outgoing traffic.
  • Advance time instead of sleeping.
DashboardTest.java 
@Test
void showsOfflineWhenTheConnectionDrops() throws IOException {
    var dashboard = new Dashboard();
    var time = new ManualTimerScheduler();
    var transport = new AtomicReference<FaultyTransport>();
    try (
            MqttBroker broker = MqttBroker.start();
            MqttClient client = MqttClient.builder()
                    .server(URI.create("mqtt://localhost:" + broker.port()))
                    .timerScheduler(time)
                    .transportFactory(clientId -> {
                        var faulty = new FaultyTransport(Transport.tcp(clientId), time);
                        transport.set(faulty);
                        return faulty;
                    })
                    .connectionListener(dashboard::onDisconnected)
                    .build()
    ) {
        client.connect().join();

        transport.get().disconnect(new IOException("cable pulled"));

        await().until(dashboard::isOffline);
    }
}

Guides and reference