package xyz.driver.core
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.model.headers.RawHeader
import akka.http.scaladsl.model.{HttpResponse, StatusCodes}
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.server.RouteResult._
import akka.http.scaladsl.server.{ExceptionHandler, Route, RouteConcatenation}
import akka.stream.ActorMaterializer
import com.typesafe.config.Config
import org.slf4j.LoggerFactory
import spray.json.DefaultJsonProtocol
import xyz.driver.core
import xyz.driver.core.logging.{Logger, TypesafeScalaLogger}
import xyz.driver.core.rest.{ContextHeaders, Swagger}
import xyz.driver.core.stats.SystemStats
import xyz.driver.core.time.Time
import xyz.driver.core.time.provider.{SystemTimeProvider, TimeProvider}
import scala.compat.Platform.ConcurrentModificationException
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
object app {
class DriverApp(version: String,
gitHash: String,
modules: Seq[Module],
time: TimeProvider = new SystemTimeProvider(),
log: Logger = new TypesafeScalaLogger(
com.typesafe.scalalogging.Logger(LoggerFactory.getLogger(classOf[DriverApp]))),
config: Config = core.config.loadDefaultConfig,
interface: String = "::0",
baseUrl: String = "localhost:8080",
port: Int = 8080) {
implicit private lazy val actorSystem = ActorSystem("spray-routing", config)
implicit private lazy val executionContext = actorSystem.dispatcher
implicit private lazy val materializer = ActorMaterializer()(actorSystem)
private lazy val http = Http()(actorSystem)
def run(): Unit = {
activateServices(modules)
scheduleServicesDeactivation(modules)
bindHttp(modules)
Console.print(s"${this.getClass.getName} App is started\n")
}
def stop(): Unit = {
http.shutdownAllConnectionPools().onComplete { _ =>
val _ = actorSystem.terminate()
val terminated = Await.result(actorSystem.whenTerminated, 30.seconds)
val addressTerminated = if (terminated.addressTerminated) "is" else "is not"
Console.print(s"${this.getClass.getName} App $addressTerminated stopped ")
}
}
protected def bindHttp(modules: Seq[Module]): Unit = {
val serviceTypes = modules.flatMap(_.routeTypes)
val swaggerService = new Swagger(baseUrl, version, actorSystem, serviceTypes, config)
val swaggerRoutes = swaggerService.routes ~ swaggerService.swaggerUI
val versionRt = versionRoute(version, gitHash, time.currentTime())
val _ = Future {
http.bindAndHandle(route2HandlerFlow(handleExceptions(ExceptionHandler(exceptionHandler)) { ctx =>
val trackingId = rest.extractTrackingId(ctx)
val contextWithTrackingId =
ctx.withRequest(ctx.request.withHeaders(RawHeader(ContextHeaders.TrackingIdHeader, trackingId)))
log.audit(s"Received request ${ctx.request} with tracking id $trackingId")
modules.map(_.route).foldLeft(versionRt ~ healthRoute ~ swaggerRoutes)(_ ~ _)(contextWithTrackingId)
}), interface, port)(materializer)
}
}
protected def exceptionHandler = PartialFunction[Throwable, Route] {
case is: IllegalStateException =>
ctx =>
val trackingId = rest.extractTrackingId(ctx)
log.debug(s"Request is not allowed to ${ctx.request.uri} ($trackingId)", is)
complete(
HttpResponse(BadRequest, entity = s"""{ "trackingId": "$trackingId", "message": "${is.getMessage}" }"""))(
ctx)
case cm: ConcurrentModificationException =>
ctx =>
val trackingId = rest.extractTrackingId(ctx)
val concurrentModificationMessage = "Resource was changed concurrently, try requesting a newer version"
log.audit(s"Concurrent modification of the resource ${ctx.request.uri} ($trackingId)", cm)
complete(
HttpResponse(
Conflict,
entity = s"""{ "trackingId": "$trackingId", "message": "$concurrentModificationMessage" }"""))(ctx)
case t: Throwable =>
ctx =>
val trackingId = rest.extractTrackingId(ctx)
log.error(s"Request to ${ctx.request.uri} could not be handled normally ($trackingId)", t)
complete(
HttpResponse(InternalServerError,
entity = s"""{ "trackingId": "$trackingId", "message": "${t.getMessage}" }"""))(ctx)
}
protected def versionRoute(version: String, gitHash: String, startupTime: Time): Route = {
import DefaultJsonProtocol._
import SprayJsonSupport._
path("version") {
val currentTime = time.currentTime().millis
complete(
Map(
"version" -> version,
"gitHash" -> gitHash,
"modules" -> modules.map(_.name).mkString(", "),
"startupTime" -> startupTime.millis.toString,
"serverTime" -> currentTime.toString,
"uptime" -> (currentTime - startupTime.millis).toString
))
}
}
protected def healthRoute: Route = {
import DefaultJsonProtocol._
import SprayJsonSupport._
import spray.json._
val memoryUsage = SystemStats.memoryUsage
val gcStats = SystemStats.garbageCollectorStats
path("health") {
complete(
Map(
"availableProcessors" -> SystemStats.availableProcessors.toJson,
"memoryUsage" -> Map(
"free" -> memoryUsage.free.toJson,
"total" -> memoryUsage.total.toJson,
"max" -> memoryUsage.max.toJson
).toJson,
"gcStats" -> Map(
"garbageCollectionTime" -> gcStats.garbageCollectionTime.toJson,
"totalGarbageCollections" -> gcStats.totalGarbageCollections.toJson
).toJson,
"fileSystemSpace" -> SystemStats.fileSystemSpace.map { f =>
Map("path" -> f.path.toJson,
"freeSpace" -> f.freeSpace.toJson,
"totalSpace" -> f.totalSpace.toJson,
"usableSpace" -> f.usableSpace.toJson)
}.toJson,
"operatingSystem" -> SystemStats.operatingSystemStats.toJson
))
}
}
/**
* Initializes services
*/
protected def activateServices(services: Seq[Module]): Unit = {
services.foreach { service =>
Console.print(s"Service ${service.name} starts ...")
try {
service.activate()
} catch {
case t: Throwable =>
log.fatal(s"Service ${service.name} failed to activate", t)
Console.print(" Failed! (check log)")
}
Console.print(" Done\n")
}
}
/**
* Schedules services to be deactivated on the app shutdown
*/
protected def scheduleServicesDeactivation(services: Seq[Module]) = {
Runtime.getRuntime.addShutdownHook(new Thread() {
override def run(): Unit = {
services.foreach { service =>
Console.print(s"Service ${service.name} shutting down ...")
try {
service.deactivate()
} catch {
case t: Throwable =>
log.fatal(s"Service ${service.name} failed to deactivate", t)
Console.print(" Failed! (check log)")
}
Console.print(" Done\n")
}
}
})
}
}
import scala.reflect.runtime.universe._
trait Module {
val name: String
def route: Route
def routeTypes: Seq[Type]
def activate(): Unit = {}
def deactivate(): Unit = {}
}
class EmptyModule extends Module {
val name = "Nothing"
def route: Route = complete(StatusCodes.OK)
def routeTypes = Seq.empty[Type]
}
class SimpleModule(val name: String, val route: Route, routeType: Type) extends Module {
def routeTypes: Seq[Type] = Seq(routeType)
}
/**
* Module implementation which may be used to composed a few
*
* @param name more general name of the composite module,
* must be provided as there is no good way to automatically
* generalize the name from the composed modules' names
* @param modules modules to compose into a single one
*/
class CompositeModule(val name: String, modules: Seq[Module]) extends Module with RouteConcatenation {
def route: Route = modules.map(_.route).reduce(_ ~ _)
def routeTypes = modules.flatMap(_.routeTypes)
override def activate() = modules.foreach(_.activate())
override def deactivate() = modules.reverse.foreach(_.deactivate())
}
}