Application.scala 3.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788
  1. package controllers
  2. import play.api.Play.current
  3. import play.api.mvc._
  4. import play.api.libs.json.Json
  5. import java.util.concurrent._
  6. import scala.concurrent._
  7. import models.{World, Fortune}
  8. import utils._
  9. import scala.concurrent.Future
  10. import play.api.libs.concurrent.Execution.Implicits._
  11. import play.core.NamedThreadFactory
  12. object Application extends Controller {
  13. private val MaxQueriesPerRequest = 20
  14. private val TestDatabaseRows = 10000
  15. private val partitionCount = current.configuration.getInt("db.default.partitionCount").getOrElse(2)
  16. private val maxConnections =
  17. partitionCount * current.configuration.getInt("db.default.maxConnectionsPerPartition").getOrElse(5)
  18. private val minConnections =
  19. partitionCount * current.configuration.getInt("db.default.minConnectionsPerPartition").getOrElse(5)
  20. private val tpe = new ThreadPoolExecutor(minConnections, maxConnections,
  21. 0L, TimeUnit.MILLISECONDS,
  22. new LinkedBlockingQueue[Runnable](),
  23. new NamedThreadFactory("dbEc"))
  24. private val dbEc = ExecutionContext.fromExecutorService(tpe)
  25. // A predicate for checking our ability to service database requests is determined by ensuring that the request
  26. // queue doesn't fill up beyond a certain threshold. For convenience we use the max number of connections * the max
  27. // # of db requests per web request to determine this threshold. It is a rough check as we don't know how many
  28. // queries we're going to make or what other threads are running in parallel etc. Nevertheless, the check is
  29. // adequate in order to throttle the acceptance of requests to the size of the pool.
  30. def isDbAvailable: Boolean = tpe.getQueue.size() < maxConnections * MaxQueriesPerRequest
  31. def db(queries: Int) = PredicatedAction(isDbAvailable, ServiceUnavailable) {
  32. Action.async {
  33. val random = ThreadLocalRandom.current()
  34. val worlds = Future.sequence((for {
  35. _ <- 1 to queries
  36. } yield Future(World.findById(random.nextInt(TestDatabaseRows) + 1))(dbEc)
  37. ).toList)
  38. worlds map {
  39. w => Ok(Json.toJson(w))
  40. }
  41. }
  42. }
  43. def fortunes() = PredicatedAction(isDbAvailable, ServiceUnavailable) {
  44. Action.async {
  45. Future(Fortune.getAll())(dbEc).map { fs =>
  46. val fortunes = Fortune(0.toLong, "Additional fortune added at request time.") +: fs
  47. Ok(views.html.fortune(fortunes))
  48. }
  49. }
  50. }
  51. def update(queries: Int) = PredicatedAction(isDbAvailable, ServiceUnavailable) {
  52. Action.async {
  53. val random = ThreadLocalRandom.current()
  54. val boundsCheckedQueries = queries match {
  55. case q if q > 500 => 500
  56. case q if q < 1 => 1
  57. case _ => queries
  58. }
  59. val worlds = Future.sequence((for {
  60. _ <- 1 to boundsCheckedQueries
  61. } yield Future {
  62. val world = World.findById(random.nextInt(TestDatabaseRows) + 1)
  63. val updatedWorld = world.copy(randomNumber = random.nextInt(TestDatabaseRows) + 1)
  64. World.updateRandom(updatedWorld)
  65. updatedWorld
  66. }(dbEc)
  67. ).toList)
  68. worlds.map {
  69. w => Ok(Json.toJson(w)).withHeaders("Server" -> "Netty")
  70. }
  71. }
  72. }
  73. }