104 lines
4.0 KiB
Scala
104 lines
4.0 KiB
Scala
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 dao.AuctionsDAO
|
|
import models.{Auction, AuctionStatus, User}
|
|
import mqttClient.MqttClientFactory
|
|
import org.apache.pekko.actor.typed.{ActorRef, 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.{JsBoolean, JsObject, JsValue, Json}
|
|
|
|
import java.nio.charset.StandardCharsets
|
|
import javax.inject.Singleton
|
|
import scala.jdk.OptionConverters.RichOptional
|
|
import scala.util.Random
|
|
|
|
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 case class AuctionConfirmed(id: Int, responseTopic: String) extends Command
|
|
|
|
private final val topic = "auctions/created"
|
|
|
|
override type Message = Command
|
|
|
|
@Provides
|
|
@Singleton
|
|
def apply(configuration: Configuration, auctionsDAO: AuctionsDAO): Behavior[Message] = Behaviors.setup { context =>
|
|
val client = MqttClientFactory.createClient(configuration, "-auction-publisher")
|
|
client.connect().whenComplete { (_action, _throwable) =>
|
|
context.self ! Ready
|
|
}
|
|
|
|
idle(context.self, client, auctionsDAO)
|
|
}
|
|
|
|
private def idle(self: ActorRef[Message], client: Mqtt5AsyncClient, auctionsDAO: AuctionsDAO, waitingPublishes: Seq[Publish] = List()): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
|
|
case p: Publish =>
|
|
idle(self, client, auctionsDAO, waitingPublishes :+ p)
|
|
case Ready =>
|
|
for { publish <- waitingPublishes } publishAuction(client, publish.auction, publish.author, self)
|
|
ready(self, client, auctionsDAO)
|
|
case message @ AuctionConfirmed(_, _) =>
|
|
logger.warn(s"Received impossible message $message")
|
|
idle(self, client, auctionsDAO, waitingPublishes)
|
|
}
|
|
|
|
private def ready(self: ActorRef[Message], client: Mqtt5AsyncClient, auctionsDAO: AuctionsDAO): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
|
|
case Publish(auction, author) =>
|
|
publishAuction(client, auction, author, self)
|
|
ready(self, client, auctionsDAO)
|
|
case AuctionConfirmed(auctionId, responseTopic) =>
|
|
client.unsubscribeWith().topicFilter(responseTopic).send()
|
|
auctionsDAO.changeStatus(auctionId, AuctionStatus.Ready)
|
|
ready(self, client, auctionsDAO)
|
|
case Ready =>
|
|
logger.warn("Message Ready received after having already received it before")
|
|
ready(self, client, auctionsDAO)
|
|
}
|
|
|
|
private def publishAuction(client: Mqtt5AsyncClient, auction: Auction, author: User, self: ActorRef[Command]): Unit = {
|
|
val payload = Json.toBytes(auction_payload(auction, author))
|
|
val responseTopic = s"auction-validate/${auction.gid}-${Random.alphanumeric.take(8).mkString}"
|
|
|
|
client.subscribeWith()
|
|
.topicFilter(responseTopic)
|
|
.callback{ message =>
|
|
logger.info(s"Receiving message on ${message.getTopic}")
|
|
logger.info(message.getPayloadAsBytes.mkString("Array(", ", ", ")"))
|
|
message.getPayload.toScala
|
|
.map(StandardCharsets.UTF_8.decode(_).toString)
|
|
.map(Json.parse)
|
|
.flatMap(v => (v \ "ok").toOption)
|
|
.map{
|
|
case b: JsBoolean if b.value => self ! AuctionConfirmed(auction.id.get, responseTopic)
|
|
// TODO: add error handling
|
|
}
|
|
}
|
|
.send()
|
|
|
|
val publish = Mqtt5Publish.builder()
|
|
.topic(topic)
|
|
.payload(payload)
|
|
.responseTopic(responseTopic)
|
|
.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
|
|
)
|
|
}
|
|
}
|