Sending actor data to MQTT

app/actors/auctions/AuctionPublisher.scala

@@ -0,0 +1,73 @@
+package actors.auctions
+
+import com.google.inject.Provides
+import com.hivemq.client.mqtt.datatypes.MqttQos
+import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish
+import models.{Auction, User}
+import mqttClient.MqttClientFactory
+import org.apache.pekko.actor.typed.Behavior
+import org.apache.pekko.actor.typed.scaladsl.Behaviors
+import play.api.{Configuration, Logging}
+import play.api.libs.concurrent.ActorModule
+import play.api.libs.json.{JsObject, Json}
+
+import javax.inject.Singleton
+
+object AuctionPublisher extends ActorModule with Logging {
+  sealed trait Command
+  private case object Ready extends Command
+  case class Publish(auction: Auction, author: User) extends Command
+
+  private final val topic = "auctions/created"
+
+  override type Message = Command
+
+  @Provides
+  @Singleton
+  def apply(configuration: Configuration): Behavior[Message] = Behaviors.setup { context =>
+    val client = MqttClientFactory.createClient(configuration, "-auction-publisher")
+    client.connect().whenComplete { (_action, _throwable) =>
+      context.self ! Ready
+    }
+
+    idle(client)
+  }
+
+  private def idle(client: Mqtt5AsyncClient, waitingPublishes: Seq[Publish] = List()): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
+    case p: Publish =>
+      idle(client, waitingPublishes :+ p)
+    case Ready =>
+      for { publish <- waitingPublishes } publishAuction(client, publish.auction, publish.author)
+      ready(client)
+  }
+
+  private def ready(client: Mqtt5AsyncClient): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
+    case Publish(auction, author) =>
+      publishAuction(client, auction, author)
+      ready(client)
+    case Ready =>
+      logger.warn("Message Ready received after having already received it before")
+      ready(client)
+  }
+
+  private def publishAuction(client: Mqtt5AsyncClient, auction: Auction, author: User): Unit = {
+    val payload = Json.toBytes(auction_payload(auction, author))
+
+    val publish = Mqtt5Publish.builder()
+      .topic(topic)
+      .payload(payload)
+      .qos(MqttQos.EXACTLY_ONCE)
+      .build()
+    logger.info(s"Publishing auction ${auction.id.getOrElse(-1)}")
+    client.publish(publish)
+  }
+
+  private def auction_payload(auction: Auction, author: User): JsObject = {
+    Json.obj(
+      "gid" -> auction.gid,
+      "author_gid" -> author.gid,
+      "starting_price" -> auction.startingPrice
+    )
+  }
+}

app/controllers/AuctionsController.scala

@@ -1,8 +1,10 @@
 package controllers
 
+import actors.auctions.AuctionPublisher
 import controllers.authentication.AuthenticatedAction
 import dao.{AuctionsDAO, AuthTokensDAO}
 import models.Auction
+import org.apache.pekko.actor.typed.ActorRef
 import play.api.libs.functional.syntax.toFunctionalBuilderOps
 import play.api.libs.json.{JsError, JsPath, JsValue, Json, Reads, Writes}
 import play.api.mvc.{BaseController, ControllerComponents}
@@ -12,7 +14,11 @@ import javax.inject.{Inject, Singleton}
 import scala.concurrent.{ExecutionContext, Future}
 
 @Singleton
-class AuctionsController @Inject()(val controllerComponents: ControllerComponents, val auctionsDAO: AuctionsDAO, val authTokensDAO: AuthTokensDAO, implicit val ec: ExecutionContext) extends BaseController {
+class AuctionsController @Inject()(val controllerComponents: ControllerComponents,
+                                   val auctionsDAO: AuctionsDAO,
+                                   val authTokensDAO: AuthTokensDAO,
+                                   val auctionPublisher: ActorRef[AuctionPublisher.Command],
+                                   implicit val ec: ExecutionContext) extends BaseController {
   def create = Action.andThen(AuthenticatedAction(authTokensDAO)).async(parse.json) { implicit request =>
     val auctionCreateDataResult = request.body.validate[AuctionCreateData]
     auctionCreateDataResult.fold(
@@ -22,24 +28,14 @@ class AuctionsController @Inject()(val controllerComponents: ControllerComponent
           auction <- auctionsDAO.create(auctionCreateData.name, auctionCreateData.startingPrice, request.user)
           json = Json.toJson(auction)
           response = Created(json)
+          _ = auctionPublisher ! AuctionPublisher.Publish(auction, request.user)
         } yield response
       }
     )
   }
 
-  case class AuctionCreateData(name: String, startingPrice: BigDecimal)
+  private case class AuctionCreateData(name: String, startingPrice: BigDecimal)
   private implicit val auctionCreateDataReads: Reads[AuctionCreateData] = (
     (JsPath \ "name").read[String] and (JsPath \ "starting_price").read[BigDecimal]
   )(AuctionCreateData.apply)
-  private implicit val auctionWrites: Writes[Auction] = new Writes[Auction] {
-    override def writes(a: Auction): JsValue = Json.obj(
-      "id" -> a.id,
-      "name" -> a.name,
-      "gid" -> a.gid,
-      "starting_price" -> a.startingPrice,
-      "current_bid" -> a.currentBid,
-      "current_bidder_id" -> a.currentBidderId,
-      "current_bid_at" -> a.currentBidAt
-    )
-  }
 }

app/models/models.scala

@@ -1,5 +1,7 @@
 package models
 
+import play.api.libs.json.{JsValue, Json, Writes}
+
 import java.time.Instant
 
 case class User(id: Option[Int], username: String, gid: String, passwordDigest: Option[String])
@@ -10,3 +12,15 @@ case class Auction(id: Option[Int], name: String, gid: String, authorId: Int,
                    startingPrice: BigDecimal, currentBid: Option[BigDecimal],
                    currentBidderId: Option[Int], currentBidAt: Option[Instant],
                    createdAt: Instant)
+
+object Auction {
+  implicit val writes: Writes[Auction] = (a: Auction) => Json.obj(
+    "id" -> a.id,
+    "name" -> a.name,
+    "gid" -> a.gid,
+    "starting_price" -> a.startingPrice,
+    "current_bid" -> a.currentBid,
+    "current_bidder_id" -> a.currentBidderId,
+    "current_bid_at" -> a.currentBidAt
+  )
+}

app/mqttClient/Module.scala

@@ -1,36 +1,18 @@
 package mqttClient
 
-import com.google.inject.{AbstractModule, Provides}
+import actors.auctions.AuctionPublisher
+import com.google.inject.{AbstractModule, Provides, Singleton}
 import com.hivemq.client.mqtt.mqtt5.{Mqtt5AsyncClient, Mqtt5Client}
-import jakarta.inject.Inject
 import play.api.libs.concurrent.PekkoGuiceSupport
 import play.api.{Configuration, Logging}
 
-import java.nio.charset.StandardCharsets
-
 class Module extends AbstractModule with PekkoGuiceSupport with Logging {
   override def configure(): Unit = {
     bindTypedActor(MessageProcessorActor, "mqtt-message-processor-actor")
     bind(classOf[MessageProcessorActor.Reg]).asEagerSingleton()
+    bindTypedActor(AuctionPublisher, "auction-publisher") // TODO move somewhere else
   }
 
-  @Provides
+  @Provides @Singleton
-  def mqtt5Client(config: Configuration): Mqtt5AsyncClient = {
+  def mqtt5Client(config: Configuration): Mqtt5AsyncClient = MqttClientFactory.createClient(config)
-    val mqttConfig = config.get[Configuration]("mqtt")
-
-    var builder = Mqtt5Client.builder()
-      .identifier(mqttConfig.get[String]("clientId"))
-      .serverHost(mqttConfig.get[String]("host"))
-      .serverPort(mqttConfig.get("port"))
-
-    mqttConfig.getOptional[String]("username").foreach{username =>
-      val password = mqttConfig.get[String]("password").getBytes(StandardCharsets.US_ASCII)
-      builder = builder.simpleAuth()
-        .username(username)
-        .password(password)
-        .applySimpleAuth()
-    }
-
-    builder.buildAsync()
-  }
 }

app/mqttClient/MqttClientFactory.scala

@@ -0,0 +1,27 @@
+package mqttClient
+
+import com.hivemq.client.mqtt.mqtt5.{Mqtt5AsyncClient, Mqtt5Client}
+import play.api.Configuration
+
+import java.nio.charset.StandardCharsets
+
+object MqttClientFactory {
+  def createClient(configuration: Configuration, suffix: String = ""): Mqtt5AsyncClient = {
+    val mqttConfig = configuration.get[Configuration]("mqtt")
+
+    var builder = Mqtt5Client.builder()
+      .identifier(mqttConfig.get[String]("clientId") + suffix)
+      .serverHost(mqttConfig.get[String]("host"))
+      .serverPort(mqttConfig.get("port"))
+
+    mqttConfig.getOptional[String]("username").foreach { username =>
+      val password = mqttConfig.get[String]("password").getBytes(StandardCharsets.US_ASCII)
+      builder = builder.simpleAuth()
+        .username(username)
+        .password(password)
+        .applySimpleAuth()
+    }
+
+    builder.buildAsync()
+  }
+}

conf/logback.xml

@@ -42,6 +42,7 @@
   <logger name="play" level="INFO"/>
   <logger name="application" level="DEBUG"/>
   <logger name="mqttClient" level="DEBUG"/>
+  <logger name="actors" level="INFO"/>
 
   <root level="WARN">
     <appender-ref ref="ASYNCFILE"/>