Added MQTT actor

2026-07-19 01:45:49 +03:00
parent a004c1a97e
commit 9bfee2978e
5 changed files with 103 additions and 0 deletions
@@ -0,0 +1,46 @@
+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()
+      }
+    }
+  }
+}