mqtt

dependencies {
    // The client (pulls in buffer + socket transitively).
    implementation("com.ditchoom:mqtt-client:<version>")

    // Add the protocol version(s) you need — or both:
    implementation("com.ditchoom:mqtt-4-models:<version>") // MQTT 3.1.1 (v4)
    implementation("com.ditchoom:mqtt-5-models:<version>") // MQTT 5.0
}

import com.ditchoom.mqtt.client.MqttClient
import com.ditchoom.mqtt.connection.MqttConnectionOptions
import com.ditchoom.mqtt.controlpacket.QualityOfService
import com.ditchoom.mqtt.controlpacket.TopicFilter
import com.ditchoom.mqtt.controlpacket.OpaquePublishPayloadCodec
import com.ditchoom.mqtt3.controlpacket.ConnectionRequest // v4; use com.ditchoom.mqtt5.* for v5
import com.ditchoom.buffer.Charset
import com.ditchoom.buffer.toReadBuffer
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import kotlinx.coroutines.flow.take

val scope = CoroutineScope(Dispatchers.Default)

// 1. Describe how/where to connect. A broker may list several options for high-availability failover.
val connection = MqttConnectionOptions.SocketConnection(host = "test.mosquitto.org", port = 1883)
val connectionRequest = ConnectionRequest(clientId = "my-client", keepAliveSeconds = 60, cleanSession = true)

// 2. Choose persistence (in-memory here) and register the broker.
val persistence = connectionRequest.controlPacketFactory.defaultPersistence(inMemory = true)
val broker = persistence.addBroker(connection, connectionRequest)

// 3. Start the client (stays connected until you shut it down).
val client = MqttClient.start(scope = scope, broker = broker, persistence = persistence)
client.awaitConnectivity()

// 4. Observe messages on a topic filter. A Codec<P> decodes the payload; OpaquePublishPayloadCodec
//    hands you the raw bytes. Register the observer before/at subscribe time.
val messages = client.observe(TopicFilter.fromOrThrow("test/+"), OpaquePublishPayloadCodec)
scope.launch {
    messages.take(1).collect { (publish, payload) ->
        val text = payload.handle.asReadBuffer().let { it.readString(it.remaining(), Charset.UTF8) }
        println("Received on ${publish.topic}: $text")
    }
}

// 5. Subscribe, then publish.
client.subscribe("test/+", OpaquePublishPayloadCodec, QualityOfService.AT_LEAST_ONCE).subAck.await()
client.publish(
    topicName = "test/123",
    qos = QualityOfService.EXACTLY_ONCE,
    payload = "hello".toReadBuffer(Charset.UTF8),
)

// 6. Clean up (drain = true waits for in-flight QoS 1/2 acks first).
client.unsubscribe("test/+").unsubAck.await()
client.shutdown(sendDisconnect = true, drain = true)

// Untyped / raw bytes:
client.publish(topicName = "t", payload = bytes /* ReadBuffer */, qos = QualityOfService.AT_LEAST_ONCE)

// Typed — supply a Codec<P> (hand-written or KSP-generated via @ProtocolMessage):
client.publish(topic = "t", payload = myValue, payloadCodec = MyCodec, qos = QualityOfService.AT_LEAST_ONCE)
client.observe(TopicFilter.fromOrThrow("t"), MyCodec).collect { (publish, value) -> /* value: P */ }