neon-vertigo/api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sending actor data to MQTT

Browse Source
This commit is contained in:
artemis
2026-08-27 00:22:15 +03:00
parent 782ab79c42
commit e99a194353
6 changed files with 123 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/actors/auctions/AuctionPublisher.scala
+66
View File
@@ -0,0 +1,66 @@
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
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.Json
import java.nio.charset.StandardCharsets
import javax.inject.Singleton
object AuctionPublisher extends ActorModule with Logging {
sealed trait Command
private case object Ready extends Command
case class Publish(auction: Auction) 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[Auction] = List()): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
case Publish(auction) =>
idle(client, waitingPublishes :+ auction)
case Ready =>
for { auction <- waitingPublishes } publishAuction(client, auction)
ready(client)
}
private def ready(client: Mqtt5AsyncClient): Behaviors.Receive[Command] = Behaviors.receiveMessage[Command] {
case Publish(auction) =>
publishAuction(client, auction)
ready(client)
case Ready =>
logger.warn("Message Ready received after ")
ready(client)
}
private def publishAuction(client: Mqtt5AsyncClient, auction: Auction): Unit = {
val payload = Json.toBytes(Json.toJson(auction))
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)
}
}
app/controllers/AuctionsController.scala
+9 -13
View File
@@ -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)
} 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
+14
View File
@@ -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
+5 -23
View File
@@ -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)
}
app/mqttClient/MqttClientFactory.scala
+27
View File
@@ -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
+1
View File
@@ -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"/>