Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
88b187e6a3 | ||
|
|
cab27eddab | ||
|
|
782ab79c42 | ||
|
|
81e126013b | ||
|
|
8582a92d85 | ||
|
|
9bfee2978e |
@@ -1 +0,0 @@
|
|||||||
{"name":"sbt","version":"1.12.13","bspVersion":"2.1.0-M1","languages":["scala"],"argv":["/usr/lib/jvm/java-21-openjdk-amd64/bin/java","-Xms100m","-Xmx100m","-classpath","/home/artemiy/.local/share/JetBrains/IntelliJIdea2026.1/Scala/launcher/sbt-launch.jar","-Dsbt.script=/home/artemiy/.local/bin/sbt","xsbt.boot.Boot","-bsp"]}
|
|
||||||
+3
-1
@@ -7,4 +7,6 @@ hs_err_pid*
|
|||||||
|
|
||||||
/project/project/target/
|
/project/project/target/
|
||||||
/project/target
|
/project/target
|
||||||
/target
|
/target
|
||||||
|
|
||||||
|
.bsp
|
||||||
|
|||||||
@@ -0,0 +1,103 @@
|
|||||||
|
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
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,41 @@
|
|||||||
|
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}
|
||||||
|
|
||||||
|
import java.security.SecureRandom
|
||||||
|
import javax.inject.{Inject, Singleton}
|
||||||
|
import scala.concurrent.{ExecutionContext, Future}
|
||||||
|
|
||||||
|
@Singleton
|
||||||
|
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(
|
||||||
|
errors => Future.successful(BadRequest(JsError.toJson(errors))),
|
||||||
|
auctionCreateData => {
|
||||||
|
for {
|
||||||
|
auction <- auctionsDAO.create(auctionCreateData.name, auctionCreateData.startingPrice, request.user)
|
||||||
|
json = Json.toJson(auction)
|
||||||
|
response = Created(json)
|
||||||
|
_ = auctionPublisher ! AuctionPublisher.Publish(auction, request.user)
|
||||||
|
} yield response
|
||||||
|
}
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
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)
|
||||||
|
}
|
||||||
@@ -2,34 +2,38 @@ package controllers.authentication
|
|||||||
|
|
||||||
import dao.AuthTokensDAO
|
import dao.AuthTokensDAO
|
||||||
import jakarta.inject.Inject
|
import jakarta.inject.Inject
|
||||||
import models.AuthToken
|
import models.{AuthToken, User}
|
||||||
import play.api.libs.json.*
|
import play.api.libs.json.*
|
||||||
import play.api.mvc.Results.Unauthorized
|
import play.api.mvc.Results.Unauthorized
|
||||||
import play.api.mvc.{ActionRefiner, Request, Result, WrappedRequest}
|
import play.api.mvc.{ActionRefiner, Request, Result, WrappedRequest}
|
||||||
|
|
||||||
import scala.concurrent.{ExecutionContext, Future}
|
import scala.concurrent.{ExecutionContext, Future}
|
||||||
|
|
||||||
class AuthenticatedRequest[A](val token: AuthToken, request: Request[A]) extends WrappedRequest[A](request)
|
class AuthenticatedRequest[A](val token: AuthToken, val user: User, request: Request[A]) extends WrappedRequest[A](request)
|
||||||
|
|
||||||
class AuthenticatedAction @Inject()(authTokensDao: AuthTokensDAO)(implicit val executionContext: ExecutionContext) extends ActionRefiner[Request, AuthenticatedRequest] {
|
class AuthenticatedAction @Inject()(authTokensDao: AuthTokensDAO)(implicit val executionContext: ExecutionContext) extends ActionRefiner[Request, AuthenticatedRequest] {
|
||||||
private final val tokenPrefix = "Token"
|
|
||||||
|
|
||||||
override protected def refine[A](request: Request[A]): Future[Either[Result, AuthenticatedRequest[A]]] = {
|
override protected def refine[A](request: Request[A]): Future[Either[Result, AuthenticatedRequest[A]]] = {
|
||||||
val unauthenticatedRequestError = Left(Unauthorized(Json.obj("error" -> "Unauthenticated")))
|
|
||||||
val authHeader = request.headers.get("Authorization")
|
val authHeader = request.headers.get("Authorization")
|
||||||
.flatMap { header =>
|
.map(_.split(" ", 2)(1))
|
||||||
if header.startsWith(tokenPrefix) then Some(header.substring(tokenPrefix.length).trim)
|
|
||||||
else None
|
|
||||||
}
|
|
||||||
.filter(_.nonEmpty)
|
.filter(_.nonEmpty)
|
||||||
|
|
||||||
authHeader match {
|
authHeader match {
|
||||||
case Some(token) =>
|
case Some(token) =>
|
||||||
authTokensDao.find(token).map {
|
authTokensDao.find(token).flatMap {
|
||||||
case Some(authToken) => Right(AuthenticatedRequest[A](authToken, request))
|
case Some(authToken) => findUser(request, authToken)
|
||||||
case None => unauthenticatedRequestError
|
case None => Future.successful(unauthenticatedRequestError)
|
||||||
}
|
}
|
||||||
case None => Future.successful(unauthenticatedRequestError)
|
case None => Future.successful(unauthenticatedRequestError)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private val unauthenticatedRequestError = Left(Unauthorized(Json.obj("error" -> "Unauthenticated")))
|
||||||
|
|
||||||
|
private def findUser[A](request: Request[A], token: AuthToken) = {
|
||||||
|
authTokensDao.getUser(token).map {
|
||||||
|
case Some(user) => Right(AuthenticatedRequest(token, user, request))
|
||||||
|
case None => unauthenticatedRequestError
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import jakarta.inject.{Inject, Singleton}
|
|||||||
import models.AuthToken
|
import models.AuthToken
|
||||||
import play.api.libs.json.*
|
import play.api.libs.json.*
|
||||||
import play.api.libs.functional.syntax.*
|
import play.api.libs.functional.syntax.*
|
||||||
import play.api.mvc.{AbstractController, Action, AnyContent, ControllerComponents}
|
import play.api.mvc.{AbstractController, ControllerComponents}
|
||||||
|
|
||||||
import scala.concurrent.{ExecutionContext, Future}
|
import scala.concurrent.{ExecutionContext, Future}
|
||||||
|
|
||||||
@@ -22,10 +22,7 @@ implicit val tokenWrites: Writes[AuthToken] = (token: AuthToken) => Json.obj(
|
|||||||
|
|
||||||
|
|
||||||
@Singleton
|
@Singleton
|
||||||
class SessionsController @Inject()(controllerComponents: ControllerComponents,
|
class SessionsController @Inject()(controllerComponents: ControllerComponents, usersDAO: UsersDAO, authTokensDAO: AuthTokensDAO, implicit val ec: ExecutionContext) extends AbstractController(controllerComponents) {
|
||||||
usersDAO: UsersDAO, authTokensDAO: AuthTokensDAO,
|
|
||||||
authenticatedAction: AuthenticatedAction,
|
|
||||||
implicit val ec: ExecutionContext) extends AbstractController(controllerComponents) {
|
|
||||||
def login() = Action.async(parse.json) { implicit request =>
|
def login() = Action.async(parse.json) { implicit request =>
|
||||||
val loginDataResult = request.body.validate[LoginData]
|
val loginDataResult = request.body.validate[LoginData]
|
||||||
loginDataResult.fold(
|
loginDataResult.fold(
|
||||||
@@ -44,7 +41,7 @@ class SessionsController @Inject()(controllerComponents: ControllerComponents,
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
def logout = (Action andThen authenticatedAction).async { request =>
|
def logout = Action.andThen(new AuthenticatedAction(authTokensDAO)).async { request =>
|
||||||
authTokensDAO.destroy(request.token).map{ _ => Ok("") }
|
authTokensDAO.destroy(request.token).map{ _ => Ok("") }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,40 @@
|
|||||||
|
package dao
|
||||||
|
|
||||||
|
import models.{Auction, AuctionStatus}
|
||||||
|
import schemas.Auctions
|
||||||
|
import schemas.auctionStatusColumnType
|
||||||
|
import play.api.db.slick.{DatabaseConfigProvider, HasDatabaseConfigProvider}
|
||||||
|
import slick.jdbc.PostgresProfile
|
||||||
|
import slick.jdbc.PostgresProfile.api.*
|
||||||
|
|
||||||
|
import java.time.Instant
|
||||||
|
import javax.inject.Inject
|
||||||
|
import scala.concurrent.{ExecutionContext, Future}
|
||||||
|
import scala.util.Random
|
||||||
|
|
||||||
|
class AuctionsDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider)(implicit val ec: ExecutionContext)
|
||||||
|
extends HasDatabaseConfigProvider[PostgresProfile] {
|
||||||
|
|
||||||
|
def create(name: String, startingPrice: BigDecimal, author: models.User): Future[Auction] = {
|
||||||
|
val auction = Auction(
|
||||||
|
id = None, name = name,
|
||||||
|
gid = Random.alphanumeric.take(24).mkString,
|
||||||
|
status = AuctionStatus.Created,
|
||||||
|
authorId = author.id.get,
|
||||||
|
startingPrice = startingPrice,
|
||||||
|
currentBid = None, currentBidderId = None, currentBidAt = None,
|
||||||
|
createdAt = Instant.now()
|
||||||
|
)
|
||||||
|
|
||||||
|
create(auction)
|
||||||
|
}
|
||||||
|
|
||||||
|
def create(auction: Auction) : Future[Auction] = {
|
||||||
|
db.run((Auctions returning Auctions) += auction)
|
||||||
|
}
|
||||||
|
|
||||||
|
def changeStatus(auctionId: Int, status: AuctionStatus): Future[Boolean] = {
|
||||||
|
db.run(Auctions.filter(_.id === auctionId).map(_.status).update(status))
|
||||||
|
.map(_ == 1)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -14,10 +14,9 @@ import models.AuthToken
|
|||||||
import schemas.AuthTokens
|
import schemas.AuthTokens
|
||||||
|
|
||||||
import scala.language.postfixOps
|
import scala.language.postfixOps
|
||||||
|
import scala.util.Random
|
||||||
|
|
||||||
class AuthTokensDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider)(implicit val ec: ExecutionContext) extends HasDatabaseConfigProvider[PostgresProfile] {
|
class AuthTokensDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProvider)(implicit val ec: ExecutionContext) extends HasDatabaseConfigProvider[PostgresProfile] {
|
||||||
private lazy val random = SecureRandom()
|
|
||||||
private lazy val base64 = Base64.getEncoder
|
|
||||||
private final lazy val tokenValidDuration = Duration.ofDays(30)
|
private final lazy val tokenValidDuration = Duration.ofDays(30)
|
||||||
|
|
||||||
def find(token: String): Future[Option[AuthToken]] = {
|
def find(token: String): Future[Option[AuthToken]] = {
|
||||||
@@ -26,12 +25,10 @@ class AuthTokensDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProv
|
|||||||
}
|
}
|
||||||
|
|
||||||
def createToken(user: models.User): Future[AuthToken] = {
|
def createToken(user: models.User): Future[AuthToken] = {
|
||||||
val tokenRaw = ByteBuffer.allocate(12)
|
|
||||||
random.nextBytes(tokenRaw.array())
|
|
||||||
val token = AuthToken(
|
val token = AuthToken(
|
||||||
id = None,
|
id = None,
|
||||||
userId = user.id.get,
|
userId = user.id.get,
|
||||||
token = base64.encodeToString(tokenRaw.array()),
|
token = Random.alphanumeric.take(12).mkString,
|
||||||
createdAt = Instant.now,
|
createdAt = Instant.now,
|
||||||
expiresAt = Instant.now.plus(tokenValidDuration)
|
expiresAt = Instant.now.plus(tokenValidDuration)
|
||||||
)
|
)
|
||||||
@@ -39,6 +36,11 @@ class AuthTokensDAO @Inject()(protected val dbConfigProvider: DatabaseConfigProv
|
|||||||
db.run((AuthTokens returning AuthTokens.map(_.id)) += token).map { tokenId => token.copy(id = Some(tokenId)) }
|
db.run((AuthTokens returning AuthTokens.map(_.id)) += token).map { tokenId => token.copy(id = Some(tokenId)) }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def getUser(token: AuthToken): Future[Option[models.User]] = {
|
||||||
|
val query = schemas.Users.filter(_.id === token.userId).take(1).result.headOption
|
||||||
|
db.run(query)
|
||||||
|
}
|
||||||
|
|
||||||
def destroy(token: AuthToken): Future[Unit] = {
|
def destroy(token: AuthToken): Future[Unit] = {
|
||||||
val q = AuthTokens.filter(_.token === token.token).delete
|
val q = AuthTokens.filter(_.token === token.token).delete
|
||||||
db.run(q).map{ _ => () }
|
db.run(q).map{ _ => () }
|
||||||
|
|||||||
+30
-1
@@ -1,7 +1,36 @@
|
|||||||
package models
|
package models
|
||||||
|
|
||||||
|
import play.api.libs.json.{JsValue, Json, Writes}
|
||||||
|
|
||||||
import java.time.Instant
|
import java.time.Instant
|
||||||
|
|
||||||
case class User(id: Option[Int], username: String, gid: String, passwordDigest: Option[String])
|
case class User(id: Option[Int], username: String, gid: String, passwordDigest: Option[String])
|
||||||
|
|
||||||
case class AuthToken(id: Option[Int], userId: Int, token: String, createdAt: Instant, expiresAt: Instant)
|
case class AuthToken(id: Option[Int], userId: Int, token: String, createdAt: Instant, expiresAt: Instant)
|
||||||
|
|
||||||
|
enum AuctionStatus(val value: Short) {
|
||||||
|
case Created extends AuctionStatus(0)
|
||||||
|
case Ready extends AuctionStatus(10)
|
||||||
|
case Finished extends AuctionStatus(10000)
|
||||||
|
}
|
||||||
|
|
||||||
|
object AuctionStatus {
|
||||||
|
lazy val byValue: Map[Short, AuctionStatus] = Map.from(AuctionStatus.values.map{it => (it.value, it)})
|
||||||
|
}
|
||||||
|
|
||||||
|
case class Auction(id: Option[Int], name: String, status: AuctionStatus, 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
|
||||||
|
)
|
||||||
|
}
|
||||||
@@ -0,0 +1,46 @@
|
|||||||
|
package mqttClient
|
||||||
|
|
||||||
|
import com.google.inject.Provides
|
||||||
|
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient
|
||||||
|
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish
|
||||||
|
import jakarta.inject.{Inject, Singleton}
|
||||||
|
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 java.nio.charset.StandardCharsets
|
||||||
|
|
||||||
|
object MessageProcessorActor extends ActorModule with Logging {
|
||||||
|
sealed trait Command
|
||||||
|
case object Stop extends Command
|
||||||
|
private case class MqttMessageReceived(publish: Mqtt5Publish) extends Command
|
||||||
|
|
||||||
|
override type Message = Command
|
||||||
|
|
||||||
|
@Singleton
|
||||||
|
class Reg @Inject()(private val mqttMessageProcessorActor: ActorRef[Command])
|
||||||
|
|
||||||
|
@Provides
|
||||||
|
def create(client: Mqtt5AsyncClient): Behavior[Command] = Behaviors.setup { context =>
|
||||||
|
client.connect().whenComplete { (_action, _throwable) =>
|
||||||
|
client.subscribeWith()
|
||||||
|
.topicFilter("test/+")
|
||||||
|
.callback{ (publish) =>
|
||||||
|
context.self ! MqttMessageReceived(publish)
|
||||||
|
}
|
||||||
|
.send()
|
||||||
|
}
|
||||||
|
|
||||||
|
Behaviors.receiveMessage[Command] {
|
||||||
|
case MqttMessageReceived(publish) =>
|
||||||
|
val topic = publish.getTopic
|
||||||
|
val payload = String(publish.getPayloadAsBytes, StandardCharsets.UTF_8)
|
||||||
|
logger.info(s"Received message $topic $payload")
|
||||||
|
Behaviors.same
|
||||||
|
case Stop => Behaviors.stopped{ () =>
|
||||||
|
client.disconnect()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
package mqttClient
|
||||||
|
|
||||||
|
import actors.auctions.AuctionPublisher
|
||||||
|
import com.google.inject.{AbstractModule, Provides, Singleton}
|
||||||
|
import com.hivemq.client.mqtt.mqtt5.{Mqtt5AsyncClient, Mqtt5Client}
|
||||||
|
import play.api.libs.concurrent.PekkoGuiceSupport
|
||||||
|
import play.api.{Configuration, Logging}
|
||||||
|
|
||||||
|
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 @Singleton
|
||||||
|
def mqtt5Client(config: Configuration): Mqtt5AsyncClient = MqttClientFactory.createClient(config)
|
||||||
|
}
|
||||||
@@ -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()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,9 +1,9 @@
|
|||||||
package schemas
|
package schemas
|
||||||
|
|
||||||
import slick.jdbc.PostgresProfile.api._
|
import slick.jdbc.PostgresProfile.api.*
|
||||||
import models._
|
import models.*
|
||||||
import slick.lifted.ProvenShape
|
import slick.lifted.ProvenShape
|
||||||
import slick.model.ForeignKeyAction.Cascade
|
import slick.model.ForeignKeyAction.{Cascade, Restrict}
|
||||||
|
|
||||||
import java.time.Instant
|
import java.time.Instant
|
||||||
|
|
||||||
@@ -31,4 +31,29 @@ class AuthTokensTable(tag: Tag) extends Table[AuthToken](tag, "auth_tokens") {
|
|||||||
def user = foreignKey("users", userId, Users)(_.id, onDelete = Cascade)
|
def user = foreignKey("users", userId, Users)(_.id, onDelete = Cascade)
|
||||||
}
|
}
|
||||||
|
|
||||||
lazy val AuthTokens = TableQuery[AuthTokensTable]
|
lazy val AuthTokens = TableQuery[AuthTokensTable]
|
||||||
|
|
||||||
|
implicit val auctionStatusColumnType: BaseColumnType[AuctionStatus] =
|
||||||
|
MappedColumnType.base[AuctionStatus, Short](
|
||||||
|
enumValue => enumValue.value,
|
||||||
|
intValue => AuctionStatus.byValue(intValue)
|
||||||
|
)
|
||||||
|
|
||||||
|
class AuctionsTable(tag: Tag) extends Table[Auction](tag, "auctions") {
|
||||||
|
def id = column[Int]("id", O.PrimaryKey, O.AutoInc)
|
||||||
|
def name = column[String]("name")
|
||||||
|
def status = column[AuctionStatus]("status")
|
||||||
|
def gid = column[String]("gid", O.Unique)
|
||||||
|
def authorId = column[Int]("author_id")
|
||||||
|
def startingPrice = column[BigDecimal]("starting_price")
|
||||||
|
def currentBid = column[Option[BigDecimal]]("current_bid")
|
||||||
|
def currentBidderId = column[Option[Int]]("current_bidder_id")
|
||||||
|
def currentBidAt = column[Option[Instant]]("current_bid_at")
|
||||||
|
def createdAt = column[Instant]("created_at")
|
||||||
|
|
||||||
|
override def * : ProvenShape[Auction] = (id.?, name, status, gid, authorId, startingPrice, currentBid, currentBidderId, currentBidAt, createdAt).mapTo[Auction]
|
||||||
|
|
||||||
|
def author = foreignKey("users", authorId, Users)(_.id, onDelete = Restrict)
|
||||||
|
}
|
||||||
|
|
||||||
|
lazy val Auctions = TableQuery[AuctionsTable]
|
||||||
@@ -21,6 +21,9 @@ libraryDependencies += "org.playframework" %% "play-slick-evolutions" % "6.2.0"
|
|||||||
// Source: https://mvnrepository.com/artifact/de.mkammerer/argon2-jvm
|
// Source: https://mvnrepository.com/artifact/de.mkammerer/argon2-jvm
|
||||||
libraryDependencies += "de.mkammerer" % "argon2-jvm" % "2.12"
|
libraryDependencies += "de.mkammerer" % "argon2-jvm" % "2.12"
|
||||||
|
|
||||||
|
// Source: https://mvnrepository.com/artifact/com.hivemq/hivemq-mqtt-client
|
||||||
|
libraryDependencies += "com.hivemq" % "hivemq-mqtt-client" % "1.3.17"
|
||||||
|
|
||||||
// Adds additional packages into Twirl
|
// Adds additional packages into Twirl
|
||||||
//TwirlKeys.templateImports += "me.artemis.controllers._"
|
//TwirlKeys.templateImports += "me.artemis.controllers._"
|
||||||
|
|
||||||
|
|||||||
@@ -2,6 +2,12 @@
|
|||||||
|
|
||||||
play.http.filters=controllers.filters.Filters
|
play.http.filters=controllers.filters.Filters
|
||||||
|
|
||||||
|
play.filters.csrf.header.bypassHeaders {
|
||||||
|
X-Requested-With = "*"
|
||||||
|
}
|
||||||
|
|
||||||
|
play.modules.enabled += "mqttClient.Module"
|
||||||
|
|
||||||
slick.dbs.default = {
|
slick.dbs.default = {
|
||||||
profile = "slick.jdbc.PostgresProfile$"
|
profile = "slick.jdbc.PostgresProfile$"
|
||||||
db = {
|
db = {
|
||||||
@@ -26,5 +32,20 @@ pekko {
|
|||||||
}
|
}
|
||||||
throughput = 1
|
throughput = 1
|
||||||
}
|
}
|
||||||
|
|
||||||
|
mqtt-dispatcher {
|
||||||
|
type = Dispatcher
|
||||||
|
executor = "fork-join-executor"
|
||||||
|
fork-join-executor {
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
mqtt {
|
||||||
|
host = "localhost"
|
||||||
|
port = 1883
|
||||||
|
clientId = "neon-vertigo-api"
|
||||||
|
username = "artemis"
|
||||||
|
password = "artemis"
|
||||||
}
|
}
|
||||||
@@ -2,13 +2,16 @@
|
|||||||
CREATE TABLE auctions
|
CREATE TABLE auctions
|
||||||
(
|
(
|
||||||
id BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY,
|
id BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY,
|
||||||
gid VARCHAR UNIQUE NOT NULL,
|
name VARCHAR NOT NULL,
|
||||||
author_id BIGINT NOT NULL REFERENCES users (id) ON DELETE RESTRICT,
|
gid VARCHAR UNIQUE NOT NULL,
|
||||||
description VARCHAR,
|
author_id BIGINT REFERENCES users (id) NOT NULL,
|
||||||
starting_price DECIMAL(12, 3) NOT NULL,
|
starting_price DECIMAL(20, 3) NOT NULL,
|
||||||
current_bid DECIMAL(12, 3),
|
current_bid DECIMAL(20, 3),
|
||||||
current_bidder_id BIGINT REFERENCES users (id) ON DELETE SET NULL
|
current_bidder_id BIGINT REFERENCES users (id),
|
||||||
|
current_bid_at TIMESTAMP,
|
||||||
|
created_at TIMESTAMP NOT NULL
|
||||||
);
|
);
|
||||||
|
|
||||||
-- !Downs
|
-- !Downs
|
||||||
DROP TABLE auctions;
|
|
||||||
|
DROP TABLE auctions
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
-- !Ups
|
||||||
|
ALTER TABLE auctions
|
||||||
|
ADD COLUMN status SMALLINT NOT NULL DEFAULT 0;
|
||||||
|
|
||||||
|
-- !Downs
|
||||||
|
ALTER TABLE auctions
|
||||||
|
DROP COLUMN status;
|
||||||
@@ -41,6 +41,8 @@
|
|||||||
|
|
||||||
<logger name="play" level="INFO"/>
|
<logger name="play" level="INFO"/>
|
||||||
<logger name="application" level="DEBUG"/>
|
<logger name="application" level="DEBUG"/>
|
||||||
|
<logger name="mqttClient" level="DEBUG"/>
|
||||||
|
<logger name="actors" level="INFO"/>
|
||||||
|
|
||||||
<root level="WARN">
|
<root level="WARN">
|
||||||
<appender-ref ref="ASYNCFILE"/>
|
<appender-ref ref="ASYNCFILE"/>
|
||||||
|
|||||||
+2
-2
@@ -7,9 +7,9 @@
|
|||||||
GET / controllers.HomeController.index()
|
GET / controllers.HomeController.index()
|
||||||
|
|
||||||
POST /login controllers.authentication.SessionsController.login()
|
POST /login controllers.authentication.SessionsController.login()
|
||||||
# TODO: enable it later
|
|
||||||
+nocsrf
|
|
||||||
DELETE /logout controllers.authentication.SessionsController.logout()
|
DELETE /logout controllers.authentication.SessionsController.logout()
|
||||||
|
|
||||||
# Map static resources from the /public folder to the /assets URL path
|
# Map static resources from the /public folder to the /assets URL path
|
||||||
GET /assets/*file controllers.Assets.versioned(path="/public", file: Asset)
|
GET /assets/*file controllers.Assets.versioned(path="/public", file: Asset)
|
||||||
|
|
||||||
|
POST /auctions controllers.AuctionsController.create()
|
||||||
Reference in New Issue
Block a user