#9, #10, #11, #12

build.sbt

@@ -44,6 +44,7 @@ libraryDependencies += "com.softwaremill.sttp.tapir" %% "tapir-swagger-ui-akka-h
 libraryDependencies += "com.softwaremill.sttp.tapir" %% "tapir-sttp-client" % "0.12.23"
 libraryDependencies += "com.softwaremill.sttp.client" %% "core" % "2.0.4"
 libraryDependencies += "com.softwaremill.sttp" %% "core" % "1.0.2"
+libraryDependencies += "com.softwaremill.sttp.tapir" %% "tapir-swagger-ui-http4s" % "0.12.23"
 
 val circeVersion = "0.12.3"
 
@@ -59,3 +60,22 @@ libraryDependencies += "com.github.ghik" % "silencer-plugin_2.12" % "1.4.2"
 libraryDependencies += "net.liftweb" %% "lift-json" % "3.4.1"
 
 libraryDependencies += "org.scalatest" %% "scalatest" % "3.0.8" % "test"
+
+libraryDependencies += "com.github.pureconfig" %% "pureconfig" % "0.12.3"
+
+libraryDependencies ++= Seq(
+
+    // Start with this one
+    "org.tpolecat" %% "doobie-core"      % "0.8.6",
+
+    // And add any of these as needed
+    "org.tpolecat" %% "doobie-h2"        % "0.8.6",          // H2 driver 1.4.200 + type mappings.
+    "org.tpolecat" %% "doobie-hikari"    % "0.8.6",          // HikariCP transactor.
+    "org.tpolecat" %% "doobie-postgres"  % "0.8.6",          // Postgres driver 42.2.9 + type mappings.
+    "org.tpolecat" %% "doobie-quill"     % "0.8.6",          // Support for Quill 3.4.10
+    "org.tpolecat" %% "doobie-specs2"    % "0.8.6" % "test", // Specs2 support for typechecking statements.
+    "org.tpolecat" %% "doobie-scalatest" % "0.8.6" % "test"  // ScalaTest support for typechecking statements.
+
+)
+
+scalacOptions += "-Ypartial-unification"

src/main/resources/application.conf

@@ -0,0 +1,16 @@
+development {
+    database {
+        data-source = "slick.jdbc.DatabaseUrlDataSource"
+        driver ="org.postgresql.Driver"
+        url = "jdbc:postgresql:world"
+        user = "postgres"
+        password = "*H1m9r4*"
+    }
+    server {
+        host = "localhost"
+        port = 8888
+    }
+}
+
+
+

src/main/scala/com/headstorm/Main.scala

@@ -1,24 +1,60 @@
 package com.headstorm
 
 import RouteGenerator._
-import akka.actor.ActorSystem
-import akka.http.scaladsl.Http
+import cats.effect._
+import cats.implicits._
+import com.typesafe.config.ConfigFactory
+import javax.sound.sampled.Port
+import org.http4s.HttpRoutes
+import org.http4s.server.Router
+import org.http4s.server.blaze.BlazeServerBuilder
+import org.http4s.syntax.kleisli._
 import sttp.tapir.docs.openapi._
 import sttp.tapir.openapi.OpenAPI
 import sttp.tapir.openapi.circe.yaml._
-import sttp.tapir.swagger.akkahttp.SwaggerAkka
+import sttp.tapir.swagger.http4s.SwaggerHttp4s
+import pureconfig.generic.auto._
+import pureconfig._
+import sttp.tapir.server.http4s.Http4sServerOptions
 
-import scala.concurrent.duration._
-import scala.concurrent.Await
-import akka.actor.ActorSystem
-import akka.http.scaladsl.Http
-import akka.http.scaladsl.server.Route
-import sttp.tapir._
-import sttp.tapir.server.akkahttp._
-import sttp.client._
+
+case class DBDetail(dataSource: String, driver: String, url: String, user: String, password: String)
+case class ServerDetail(host: String, port: Int)
+
+case class Config(database: DBDetail, server: ServerDetail)
+
+case class DevConfig(development: Config)
 
 object Main extends App {
 
+    val config = ConfigSource.default.load[DevConfig]
+
+    private val configuration = {
+        ConfigSource.default.load[DevConfig] match {
+            case Left(error) =>
+                println(s"There was an error loading the config, shutting down: ${error.toString}")
+                System.exit(1)
+            case Right(config) => config
+        }
+    }
+
+    val (serverHost: String,
+        serverPort: Int,
+        databaseDataSource: String,
+        databaseDriver: String,
+        databaseURL: String,
+        databaseUser: String,
+        databasePassword: String) = configuration match {
+        case DevConfig(r) => (r.server.host.toString,
+                            r.server.port,
+                            r.database.dataSource,
+                            r.database.driver,
+                            r.database.url,
+                            r.database.user,
+                            r.database.password)
+        case _ => println("Error Loading Config")
+    }
+
     // the endpoints' routes
     val helloWorldInit = new Endpoints("name", "hello", "get")
     val helloWorldEndpoint = helloWorldInit.aEndpoint
@@ -28,55 +64,42 @@ object Main extends App {
     val byeWorldEndpoint = byeWorldInit.aEndpoint
     val byeWorldRoute = byeWorldInit.aRoute
 
+    val databaseSelectInit = new DatabaseEndpoints(databaseDataSource, databaseDriver, databaseURL, databaseUser, databasePassword, "name", "dbselect", "select")
+    val databaseSelectEndpoint = databaseSelectInit.aEndpoint
+    def databaseSelectRoute[F[_]: Sync](implicit serverOption: Http4sServerOptions[F], fcs: ContextShift[F]) = databaseSelectInit.aRoute
+
+    val databaseInsertInit = new DatabaseEndpoints(databaseDataSource, databaseDriver, databaseURL, databaseUser, databasePassword, "name", "dbinsert", "insert")
+    val databaseInsertEndpoint = databaseInsertInit.aEndpoint
+    def databaseInsertRoute[F[_]: Sync](implicit serverOption: Http4sServerOptions[F], fcs: ContextShift[F]) = databaseInsertInit.aRoute
+
+    val databaseUpdateInit = new DatabaseEndpoints(databaseDataSource, databaseDriver, databaseURL, databaseUser, databasePassword, "name", "dbupdate", "update")
+    val databaseUpdateEndpoint = databaseUpdateInit.aEndpoint
+    def databaseUpdateRoute[F[_]: Sync](implicit serverOption: Http4sServerOptions[F], fcs: ContextShift[F]) = databaseUpdateInit.aRoute
+
     val streamInit = new StreamEndpoints("text", "stream", "get")
     val streamEndpoint = streamInit.streamingEndpoint
     val streamRoute = streamInit.streamingRoute
 
     // generating the documentation in yml; extension methods come from imported packages
-    val openApiDocs: OpenAPI = List(helloWorldEndpoint, byeWorldEndpoint, streamEndpoint).toOpenAPI("The tapir library", "1.0.0")
+    val routes: HttpRoutes[IO] = helloWorldRoute <+> byeWorldRoute <+> streamRoute <+> databaseSelectRoute <+> databaseInsertRoute <+> databaseUpdateRoute
+
+    // generating the documentation in yml; extension methods come from imported packages
+    val openApiDocs: OpenAPI = List(helloWorldEndpoint, byeWorldEndpoint, streamEndpoint, databaseSelectEndpoint, databaseInsertEndpoint, databaseUpdateEndpoint).toOpenAPI("The tapir library", "1.0.0")
     val openApiYml: String = openApiDocs.toYaml
 
     // starting the server
-    implicit val actorSystem: ActorSystem = ActorSystem()
-    import actorSystem.dispatcher
-
-    val routes = {
-        import akka.http.scaladsl.server.Directives._
-        helloWorldRoute ~ byeWorldRoute ~ streamRoute ~ new SwaggerAkka(openApiYml).routes
-    }
-
-    val bindAndCheck = Http().bindAndHandle(routes, "localhost", 8080).map { _ =>
-        implicit val backend: SttpBackend[Identity, Nothing, NothingT] = HttpURLConnectionBackend()
-
-        //test hello GET endpoint
-        val result: String = basicRequest.response(asStringAlways).get(uri"http://localhost:8080/hello?name=Frodo").send().body
-        //        println("Got result: " + result)
-        assert(result == "{\"code\":\"200\",\"msg\":\"hello : name : Frodo\"}")
-
-        //test bye POST endpoint
-        val result2: String = basicRequest.response(asStringAlways).post(uri"http://localhost:8080/bye?name=Frodo").send().body
-        //        println("Got result: " + result2)
-        assert(result2 == "{\"code\":\"200\",\"msg\":\"bye : name : Frodo\"}")
-
-        //test streaming GET endpoint
-        val result3: String = basicRequest.response(asStringAlways).get(uri"http://localhost:8080/stream?text=Frodo").send().body
-        //        println("Got result: " + result3)
-        assert(result3 == "{\n  \"code\" : \"200\",\n  \"msg\" : \"stream : text : Frodo\"\n}")
-
-        //test Not Found
-        val result4: String = basicRequest.response(asStringAlways).get(uri"http://localhost:8080/stream2?text=Frodo").send().statusText
-        //        println("Got result: " + result4)
-        assert(result4 == "Not Found")
-
-        //running server
-        println("Go to: http://localhost:8080/docs")
-        println("Press any key to exit ...")
-        scala.io.StdIn.readLine()
-    }
+    BlazeServerBuilder[IO]
+        .bindHttp(serverPort, serverHost)
+        .withHttpApp(Router("/" -> (routes <+> new SwaggerHttp4s(openApiYml).routes[IO])).orNotFound)
+        .resource
+        .use { _ =>
+            IO {
+                println(s"Go to: http://$serverHost:$serverPort/docs")
+                println("Press any key to exit ...")
+                scala.io.StdIn.readLine()
+            }
+        }
+        .unsafeRunSync()
 
-    // cleanup
-    Await.result(bindAndCheck.transformWith { r =>
-        actorSystem.terminate().transform(_ => r)
-    }, 1.minute)
 
 }

src/main/scala/com/headstorm/RouteGenerator.scala

@@ -1,16 +1,22 @@
 package com.headstorm
 
 import sttp.tapir.Endpoint
-
-import scala.concurrent.Future
+import sttp.model.StatusCode
+import io.circe.generic.auto._
+import io.circe.parser._
+import io.circe.syntax._
 import sttp.tapir.json.circe._
-import akka.http.scaladsl.server.Route
-import akka.stream.scaladsl.Source
-import akka.util.ByteString
+import cats.effect._
+import cats.implicits._
+import org.http4s.HttpRoutes
 import sttp.tapir._
-import sttp.tapir.server.akkahttp._
-import sttp.model.StatusCode
-import io.circe.generic.auto._, io.circe.syntax._
+import sttp.tapir.server.http4s._
+import fs2._
+import sttp.model.HeaderNames
+import doobie._
+import doobie.implicits._
+import cats.effect.IO
+import scala.concurrent.ExecutionContext
 
 object RouteGenerator {
 
@@ -23,6 +29,10 @@ object RouteGenerator {
     case class Messages(code: String, msg: String)
     val msgCodec = jsonBody[Messages]
 
+    implicit val ec: ExecutionContext = scala.concurrent.ExecutionContext.Implicits.global
+    implicit val contextShift: ContextShift[IO] = IO.contextShift(ec)
+    implicit val timer: Timer[IO] = IO.timer(ec)
+
     class Endpoints(paramName: String, endpointName: String, method: String) {
 
         val baseEndpoint = endpoint.errorOut(
@@ -42,7 +52,7 @@ object RouteGenerator {
 
         }
 
-        val aRoute = aEndpoint.toRoute(in => Future.successful(Right(Messages("200", s"$endpointName : $paramName : $in"))))
+        val aRoute = aEndpoint.toRoutes(in => IO(Messages("200", s"$endpointName : $paramName : $in").asRight[ErrorInfo with Product with Serializable]))
     }
 
     class StreamEndpoints(paramName: String, endpointName: String, method: String) {
@@ -56,19 +66,107 @@ object RouteGenerator {
             )
         )
 
-        val streamingEndpoint: Endpoint[String, ErrorInfo with Product with Serializable, Source[ByteString, Any], Source[ByteString, Any]] = {
+        val streamingEndpoint = {
             method match {
-                case "get" => baseEndpoint.get.in(endpointName).in(query[String](paramName)).out(streamBody[Source[ByteString, Any]](schemaFor[Messages], CodecFormat.TextPlain()))
-                case "post" => baseEndpoint.post.in(endpointName).in(query[String](paramName)).out(streamBody[Source[ByteString, Any]](schemaFor[Messages], CodecFormat.TextPlain()))
+                case "get" => baseEndpoint.get.in(endpointName).in(query[String](paramName)).out(header[Long](HeaderNames.ContentLength)).out(streamBody[Stream[IO, Byte]](schemaFor[String], CodecFormat.TextPlain()))
+                case "post" => baseEndpoint.post.in(endpointName).in(query[String](paramName)).out(header[Long](HeaderNames.ContentLength)).out(streamBody[Stream[IO, Byte]](schemaFor[String], CodecFormat.TextPlain()))
             }
         }
 
         def createStream(in: String) = {
-            val stringMsg = Messages("200", s"$endpointName : $paramName : $in").asJson.toString()
-            val streamMsg = Source.repeat(stringMsg).take(1).map(s => ByteString(s))
-            Future.successful(Right(streamMsg))
+            val size = 100L
+            val responseMsg = Messages("200", in)
+            val responseMsgJson = responseMsg.asJson.toString()
+            val listChar = responseMsgJson.toList
+            val streamProcess = Stream
+                .emit(listChar)
+//                .repeat
+                .flatMap(list => Stream.chunk(Chunk.seq(list)))
+//                .metered[IO](50.millis)
+                .take(size)
+                .covary[IO]
+                .map(_.toByte)
+                .pure[IO]
+                .map(s => Right((size, s)))
+            streamProcess
+        }
+
+        val streamingRoute: HttpRoutes[IO] = streamingEndpoint.toRoutes {createStream _}
+
+    }
+
+    class DatabaseEndpoints(dataSource: String, driver: String, url: String, user: String, password: String, paramName: String, endpointName: String, method: String) {
+
+        val xa = Transactor.fromDriverManager[IO](driver, url, user, password)
+        case class Country(code: String, name: String, pop: Int, gnp: Option[Double])
+        case class City(id: Int, name: String, countryCode: String, district: String, population: Int)
+        case class UpdateCity(idUpdate: Int, name: String, countryCode: String, district: String, population: Int)
+
+        def select(n: String): ConnectionIO[Option[Country]] =
+            sql"select code, name, population, gnp from country where name = $n".query[Country].option
+
+        def insert(c: City) =
+            sql"insert into city (id, name, countrycode, district, population) values (${c.id}, ${c.name}, ${c.countryCode}, ${c.district}, ${c.population})".update
+
+        def update(c: UpdateCity) =
+            sql"update city set name = ${c.name}, countrycode = ${c.countryCode}, district = ${c.district}, population = ${c.population} where id = ${c.idUpdate}".update
+
+        val baseEndpoint = endpoint.errorOut(
+            oneOf(
+                statusMapping(StatusCode.NotFound, jsonBody[NotFound].description("not found")),
+                statusMapping(StatusCode.Unauthorized, jsonBody[Unauthorized].description("unauthorized")),
+                statusMapping(StatusCode.NoContent, emptyOutput.map(_ => NoContent)(_ => ())),
+                statusDefaultMapping(jsonBody[Unknown].description("unknown"))
+            )
+        )
+
+        val aEndpoint: Endpoint[String, ErrorInfo with Product with Serializable, Messages, Nothing] = {
+            method match {
+                case "select" => baseEndpoint.get.in(endpointName).in(query[String](paramName)).out(jsonBody[Messages])
+                case "insert" => baseEndpoint.put.in(endpointName).in(stringBody).out(jsonBody[Messages])
+                case "update" => baseEndpoint.post.in(endpointName).in(stringBody).out(jsonBody[Messages])
+            }
+
+        }
+
+        def selectLogic(n: String) = {
+            val qResult = select(n).transact(xa).unsafeRunSync match {
+                case Some(x) =>
+                    x match {
+                        case in: Country => Messages("200", s"${in.asJson}")
+                    }
+                case _ => Messages("400", "Error")
+            }
+            qResult
+        }
+
+        def insertLogic(n: String) = {
+            val qResult = decode[City](n) match {
+                case Right(r) => insert(r).run.transact(xa).unsafeRunSync match {
+                    case x: Int => Messages("200", s"Updated $x record(s).")
+                    case _ => Messages("400", "Error")
+                }
+            }
+            qResult
+        }
+
+        def updateLogic(n: String) = {
+            val qResult = decode[UpdateCity](n) match {
+                case Right(r) => update(r).run.transact(xa).unsafeRunSync match {
+                    case x: Int => Messages("200", s"Updated $x record(s).")
+                    case _ => Messages("400", "Error")
+                }
+            }
+            qResult
+        }
+
+        def aRoute[F[_]: Sync](implicit serverOption: Http4sServerOptions[F], fcs: ContextShift[F]) = {
+            method match {
+                case "select" => aEndpoint.toRoutes ( in => selectLogic(in).asRight[ErrorInfo with Product with Serializable].pure[F] )
+                case "insert" => aEndpoint.toRoutes ( in => insertLogic(in).asRight[ErrorInfo with Product with Serializable].pure[F] )
+                case "update" => aEndpoint.toRoutes ( in => updateLogic(in).asRight[ErrorInfo with Product with Serializable].pure[F] )
+            }
         }
-        val streamingRoute: Route = streamingEndpoint.toRoute(createStream _)
 
     }
 }