birbnetes/mqtt-benchmark-component
birbnetes/mqtt-benchmark-component
4
0
Fork 0
Code Pull Requests Releases Activity

something is fucked
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Torma Kristóf 2020-10-21 18:34:52 +02:00
parent 872c84d3f3
commit 56cf63bb0a
Signed by: tormakris
GPG Key ID: DC83C4F2C41B1047
1 changed files with 26 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
src/main/kotlin/MQTTSubscriber.kt
View File

@ -2,16 +2,24 @@ import org.eclipse.paho.client.mqttv3.*
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
import java.time.Instant import java.time.Instant
class MqttSubscriber : MqttCallbackExtended { class MqttSubscriber() : MqttCallbackExtended {
private val broker = "tcp://mqtt.k8s.kmlabz.com:1883" private val broker = "tcp://mqtt.k8s.kmlabz.com:1883"
private val clientId = MqttClient.generateClientId() private val clientId = MqttClient.generateClientId()
private val persistence = MemoryPersistence() private val persistence = MemoryPersistence()
private val mqttClient = MqttAsyncClient(broker, clientId, persistence) private val mqttClient = MqttClient(broker, clientId, persistence)
private val benchValues = HashMap<Int, Instant>() private val benchValues = HashMap<Int, Instant>()
private val connOpts = MqttConnectOptions()
init {
connOpts.isCleanSession = true
connOpts.userName = "birbnetes"
connOpts.password = "de4d2182".toCharArray()
connOpts.isAutomaticReconnect = true
mqttClient.setCallback(MqttSubscriber())
}
override fun connectionLost(arg0: Throwable) { override fun connectionLost(arg0: Throwable) {
System.err.println("connection lost") System.err.println("connection lost")
mqttClient.reconnect()
} }
override fun deliveryComplete(arg0: IMqttDeliveryToken) { override fun deliveryComplete(arg0: IMqttDeliveryToken) {
@ -26,22 +34,29 @@ class MqttSubscriber : MqttCallbackExtended {
override fun connectComplete(reconnect: Boolean, serverURI: String?) { override fun connectComplete(reconnect: Boolean, serverURI: String?) {
println("Subscribing to all topics") println("Subscribing to all topics")
try {
mqttClient.subscribe("#", 0) mqttClient.subscribe("#", 0)
}catch (e: MqttException) {
e.printStackTrace();
}
} }
fun connect() { fun connect() {
println("Connecting to broker") println("Connecting to broker")
val connOpts = MqttConnectOptions() try {
connOpts.isCleanSession = true
connOpts.userName = "birbnetes"
connOpts.password = "de4d2182".toCharArray()
mqttClient.setCallback(MqttSubscriber())
mqttClient.connect(connOpts) mqttClient.connect(connOpts)
}catch (e: MqttException) {
e.printStackTrace();
}
} }
fun disconnect(): HashMap<Int, Instant> { fun disconnect(): HashMap<Int, Instant> {
println("Disconnecting from broker") println("Disconnecting from broker")
try {
mqttClient.disconnect() mqttClient.disconnect()
}catch (e: MqttException) {
e.printStackTrace();
}
mqttClient.close() mqttClient.close()
return benchValues return benchValues
} }