61 lines
1.9 KiB
Kotlin
61 lines
1.9 KiB
Kotlin
import org.eclipse.paho.client.mqttv3.*
|
|
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
|
|
import java.time.Instant
|
|
|
|
class MqttSubscriber : MqttCallbackExtended {
|
|
private val broker = System.getenv("BROKER") ?: "tcp://mqtt.k8s.kmlabz.com:1883"
|
|
private val clientId = MqttClient.generateClientId()
|
|
private val persistence = MemoryPersistence()
|
|
private val mqttClient = MqttClient(broker, clientId, persistence)
|
|
private val benchValues = LinkedHashMap<Int, Instant>()
|
|
private val connOpts = MqttConnectOptions()
|
|
|
|
init {
|
|
connOpts.isCleanSession = true
|
|
connOpts.userName = System.getenv("USERNAME") ?: "birbnetes"
|
|
connOpts.password = System.getenv("PASSWORD").toCharArray()
|
|
connOpts.isAutomaticReconnect = true
|
|
mqttClient.setCallback(this)
|
|
}
|
|
|
|
override fun connectionLost(arg0: Throwable) {
|
|
System.err.println("connection lost")
|
|
}
|
|
|
|
override fun deliveryComplete(arg0: IMqttDeliveryToken) {
|
|
}
|
|
|
|
@Throws(Exception::class)
|
|
override fun messageArrived(topic: String, message: MqttMessage) {
|
|
benchValues[topic.split("/")[1].toInt()] = Instant.now()
|
|
}
|
|
|
|
override fun connectComplete(reconnect: Boolean, serverURI: String?) {
|
|
println("Subscribing to all topics")
|
|
try {
|
|
mqttClient.subscribe("#", 0)
|
|
}catch (e: MqttException) {
|
|
e.printStackTrace();
|
|
}
|
|
}
|
|
|
|
fun connect() {
|
|
println("Connecting to broker")
|
|
try {
|
|
mqttClient.connect(connOpts)
|
|
}catch (e: MqttException) {
|
|
e.printStackTrace();
|
|
}
|
|
}
|
|
|
|
fun disconnect(): LinkedHashMap<Int, Instant> {
|
|
println("Disconnecting from broker")
|
|
try {
|
|
mqttClient.disconnect()
|
|
}catch (e: MqttException) {
|
|
e.printStackTrace();
|
|
}
|
|
mqttClient.close()
|
|
return benchValues
|
|
}
|
|
} |