diff --git a/app/actors/auctions/AuctionPublisher.scala b/app/actors/auctions/AuctionPublisher.scala new file mode 100644 index 0000000..e539191 --- /dev/null +++ b/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 + ) + } +} diff --git a/app/controllers/AuctionsController.scala b/app/controllers/AuctionsController.scala index 9ce8b33..a41082b 100644 --- a/app/controllers/AuctionsController.scala +++ b/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 - ) - } } diff --git a/app/models/models.scala b/app/models/models.scala index c4fcda8..806e6fd 100644 --- a/app/models/models.scala +++ b/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]) @@ -9,4 +11,16 @@ case class AuthToken(id: Option[Int], userId: Int, token: String, createdAt: Ins 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) \ No newline at end of file + 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 + ) +} \ No newline at end of file diff --git a/app/mqttClient/Module.scala b/app/mqttClient/Module.scala index 5bab635..8e8cd48 100644 --- a/app/mqttClient/Module.scala +++ b/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 - def mqtt5Client(config: Configuration): Mqtt5AsyncClient = { - 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() - } + @Provides @Singleton + def mqtt5Client(config: Configuration): Mqtt5AsyncClient = MqttClientFactory.createClient(config) } diff --git a/app/mqttClient/MqttClientFactory.scala b/app/mqttClient/MqttClientFactory.scala new file mode 100644 index 0000000..4cf28de --- /dev/null +++ b/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() + } +} diff --git a/conf/logback.xml b/conf/logback.xml index 61ea438..5aa4d20 100644 --- a/conf/logback.xml +++ b/conf/logback.xml @@ -42,6 +42,7 @@ +