47 lines
1.4 KiB
Scala
47 lines
1.4 KiB
Scala
package mqttClient
|
|
|
|
import com.google.inject.Provides
|
|
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient
|
|
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish
|
|
import jakarta.inject.{Inject, Singleton}
|
|
import org.apache.pekko.actor.typed.{ActorRef, Behavior}
|
|
import org.apache.pekko.actor.typed.scaladsl.Behaviors
|
|
import play.api.Logging
|
|
import play.api.libs.concurrent.ActorModule
|
|
|
|
import java.nio.charset.StandardCharsets
|
|
|
|
object MessageProcessorActor extends ActorModule with Logging {
|
|
sealed trait Command
|
|
case object Stop extends Command
|
|
private case class MqttMessageReceived(publish: Mqtt5Publish) extends Command
|
|
|
|
override type Message = Command
|
|
|
|
@Singleton
|
|
class Reg @Inject()(private val mqttMessageProcessorActor: ActorRef[Command])
|
|
|
|
@Provides
|
|
def create(client: Mqtt5AsyncClient): Behavior[Command] = Behaviors.setup { context =>
|
|
client.connect().whenComplete { (_action, _throwable) =>
|
|
client.subscribeWith()
|
|
.topicFilter("test/+")
|
|
.callback{ (publish) =>
|
|
context.self ! MqttMessageReceived(publish)
|
|
}
|
|
.send()
|
|
}
|
|
|
|
Behaviors.receiveMessage[Command] {
|
|
case MqttMessageReceived(publish) =>
|
|
val topic = publish.getTopic
|
|
val payload = String(publish.getPayloadAsBytes, StandardCharsets.UTF_8)
|
|
logger.info(s"Received message $topic $payload")
|
|
Behaviors.same
|
|
case Stop => Behaviors.stopped{ () =>
|
|
client.disconnect()
|
|
}
|
|
}
|
|
}
|
|
}
|