Processing new bid MQTT messages

app/mqttClient/MessageProcessorActor.scala

@@ -8,8 +8,10 @@ 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 play.api.libs.json._
 
 import java.nio.charset.StandardCharsets
+import scala.util.matching.Regex
 
 object MessageProcessorActor extends ActorModule with Logging {
   sealed trait Command
@@ -18,14 +20,16 @@ object MessageProcessorActor extends ActorModule with Logging {
 
   override type Message = Command
 
+  private final val newBidPattern: Regex = """notify/new_bid/(\w+)""".r
+
   @Singleton
   class Reg @Inject()(private val mqttMessageProcessorActor: ActorRef[Command])
 
   @Provides
-  def create(client: Mqtt5AsyncClient): Behavior[Command] = Behaviors.setup { context =>
+  def create(client: Mqtt5AsyncClient, newBidService: services.auctions.NewBid): Behavior[Command] = Behaviors.setup { context =>
     client.connect().whenComplete { (_action, _throwable) =>
       client.subscribeWith()
-        .topicFilter("test/+")
+        .topicFilter("notify/new_bid/+")
         .callback{ (publish) =>
           context.self ! MqttMessageReceived(publish)
         }
@@ -37,6 +41,17 @@ object MessageProcessorActor extends ActorModule with Logging {
         val topic = publish.getTopic
         val payload = String(publish.getPayloadAsBytes, StandardCharsets.UTF_8)
         logger.info(s"Received message $topic $payload")
+
+        topic.toString match {
+          case newBidPattern(auctionGid) =>
+            Json.parse(payload).validate[services.auctions.NewBid.Request] match {
+              case JsSuccess(request, _) => newBidService.perform(auctionGid, request)
+              case JsError(errors) =>
+                logger.warn(errors.toString())
+            }
+            println(s"New bid $auctionGid")
+        }
+
         Behaviors.same
       case Stop => Behaviors.stopped{ () =>
         client.disconnect()