Skip to content

Commit 64fd93c

Browse files
yankun1992yankun1992
authored
[Scala/otavia] Add new framework otavia: Your shiny new IO & Actor programming model! (#9158)
1 parent 848d684 commit 64fd93c

17 files changed

+832
-0
lines changed

‎frameworks/Scala/otavia/.mill-version

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
0.11.8

‎frameworks/Scala/otavia/.scalafmt.conf

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
version = "3.5.3"
2+
3+
runner.dialect = scala3
4+
maxColumn = 120
5+
docstrings.blankFirstLine = no
6+
docstrings.style = AsteriskSpace
7+
docstrings.removeEmpty = true
8+
docstrings.oneline = fold
9+
docstrings.wrap = yes
10+
docstrings.wrapMaxColumn = 120
11+
docstrings.forceBlankLineBefore = true
12+
align.preset = more
13+
14+
indent.main = 4
15+
16+
newlines.topLevelBodyIfMinStatements = [before,after]

‎frameworks/Scala/otavia/README.MD

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
## Introduction
2+
3+
[GitHub - otavia-projects/otavia : Your shiny new IO & Actor programming model!](https://github.com/otavia-projects/otavia)
4+
5+
`otavia` is an IO and Actor programming model power by Scala 3, it provides a toolkit to make writing high-performance
6+
concurrent programs more easily.
7+
8+
You can get a quick overview of the basic usage and core design of `otavia` in the following documentation:
9+
10+
- [Quick Start](https://github.com/otavia-projects/otavia/blob/main/docs/_docs/quick_start.md)
11+
- [Core Concepts and Design](https://github.com/otavia-projects/otavia/blob/main/docs/_docs/core_concept.md)
12+
13+
More document can be found at [website](https://otavia.cc/home.html)

‎frameworks/Scala/otavia/benchmark/src/app/controller/DBController.scala

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
package app.controller
2+
3+
import app.controller.DBController.*
4+
import app.model.World
5+
import cc.otavia.core.actor.{MessageOf, StateActor}
6+
import cc.otavia.core.address.Address
7+
import cc.otavia.core.message.{Ask, Reply}
8+
import cc.otavia.core.stack.helper.{FutureState, FuturesState, StartState}
9+
import cc.otavia.core.stack.{AskStack, StackState, StackYield}
10+
import cc.otavia.http.server.{HttpRequest, HttpResponse}
11+
import cc.otavia.sql.Connection
12+
import cc.otavia.sql.Statement.{ModifyRows, PrepareQuery}
13+
14+
import java.util.SplittableRandom
15+
16+
class DBController extends StateActor[REQ] {
17+
18+
private var connection: Address[MessageOf[Connection]] = _
19+
20+
private val random = new SplittableRandom()
21+
22+
override protected def afterMount(): Unit = connection = autowire[Connection]()
23+
24+
override protected def resumeAsk(stack: AskStack[REQ & Ask[? <: Reply]]): StackYield =
25+
stack match
26+
case stack: AskStack[SingleQueryRequest] if stack.ask.isInstanceOf[SingleQueryRequest] =>
27+
handleSingleQuery(stack)
28+
case stack: AskStack[MultipleQueryRequest] if stack.ask.isInstanceOf[MultipleQueryRequest] =>
29+
handleMultipleQuery(stack)
30+
case stack: AskStack[UpdateRequest] if stack.ask.isInstanceOf[UpdateRequest] =>
31+
handleUpdateQuery(stack)
32+
33+
// Test 2: Single database query
34+
private def handleSingleQuery(stack: AskStack[SingleQueryRequest]): StackYield = {
35+
stack.state match
36+
case _: StartState =>
37+
val state = FutureState[World]()
38+
connection.ask(PrepareQuery.fetchOne[World](SELECT_WORLD, Tuple1(randomWorld())), state.future)
39+
stack.suspend(state)
40+
case state: FutureState[World] =>
41+
stack.`return`(state.future.getNow)
42+
}
43+
44+
// Test 3: Multiple database queries
45+
private def handleMultipleQuery(stack: AskStack[MultipleQueryRequest]): StackYield = {
46+
stack.state match
47+
case _: StartState =>
48+
stack.suspend(selectWorlds(normalizeQueries(stack.ask.params)))
49+
case state: FuturesState[World] =>
50+
val response = HttpResponse.builder.setContent(state.futures.map(_.getNow)).build()
51+
stack.`return`(response)
52+
}
53+
54+
// Test 5: Database updates
55+
private def handleUpdateQuery(stack: AskStack[UpdateRequest]): StackYield = {
56+
stack.state match
57+
case _: StartState =>
58+
stack.suspend(selectWorlds(normalizeQueries(stack.ask.params)))
59+
case state: FuturesState[World] =>
60+
val worlds = state.futures.map(_.getNow)
61+
stack.attach(worlds)
62+
val newState = FutureState[ModifyRows]()
63+
val newWorlds = worlds.sortBy(_.id).map(_.copy(randomNumber = randomWorld()))
64+
connection.ask(PrepareQuery.update(UPDATE_WORLD, newWorlds), newState.future)
65+
stack.suspend(newState)
66+
case state: FutureState[ModifyRows] =>
67+
if (state.future.isFailed) state.future.causeUnsafe.printStackTrace()
68+
val response = HttpResponse.builder.setContent(stack.attach[Seq[World]]).build()
69+
stack.`return`(response)
70+
}
71+
72+
private def selectWorlds(queries: Int): StackState = {
73+
val state = FuturesState[World](queries)
74+
for (future <- state.futures)
75+
connection.ask(PrepareQuery.fetchOne[World](SELECT_WORLD, Tuple1(randomWorld())), future)
76+
state
77+
}
78+
79+
private def randomWorld(): Int = 1 + random.nextInt(10000)
80+
81+
private def normalizeQueries(params: Map[String, String]): Int = {
82+
params.get("queries") match
83+
case Some(value) =>
84+
try {
85+
val queries = value.toInt
86+
if (queries < 1) 1 else if (queries > 500) 500 else queries
87+
} catch {
88+
case e: Throwable => 1
89+
}
90+
case None => 1
91+
}
92+
93+
}
94+
95+
object DBController {
96+
97+
type REQ = SingleQueryRequest | MultipleQueryRequest | UpdateRequest
98+
99+
class SingleQueryRequest extends HttpRequest[Nothing, World]
100+
class MultipleQueryRequest extends HttpRequest[Nothing, HttpResponse[Seq[World]]]
101+
class UpdateRequest extends HttpRequest[Nothing, HttpResponse[Seq[World]]]
102+
103+
private val SELECT_WORLD = "SELECT id, randomnumber from WORLD where id=$1"
104+
private val UPDATE_WORLD = "update world set randomnumber=$2 where id=$1"
105+
106+
}

‎frameworks/Scala/otavia/benchmark/src/app/controller/FortuneController.scala

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
package app.controller
2+
3+
import app.controller.FortuneController.*
4+
import app.model.Fortune
5+
import cc.otavia.core.actor.{MessageOf, StateActor}
6+
import cc.otavia.core.address.Address
7+
import cc.otavia.core.stack.helper.{FutureState, StartState}
8+
import cc.otavia.core.stack.{AskStack, StackState, StackYield}
9+
import cc.otavia.http.server.{HttpRequest, HttpResponse}
10+
import cc.otavia.sql.Statement.PrepareQuery
11+
import cc.otavia.sql.{Connection, RowSet}
12+
13+
class FortuneController extends StateActor[FortuneRequest] {
14+
15+
private var connection: Address[MessageOf[Connection]] = _
16+
17+
override protected def afterMount(): Unit = connection = autowire[Connection]()
18+
19+
// Test 4: Fortunes
20+
override protected def resumeAsk(stack: AskStack[FortuneRequest]): StackYield = {
21+
stack.state match
22+
case _: StartState =>
23+
val state = FutureState[RowSet[Fortune]]()
24+
connection.ask(PrepareQuery.fetchAll[Fortune](SELECT_FORTUNE), state.future)
25+
stack.suspend(state)
26+
case state: FutureState[RowSet[Fortune]] =>
27+
val fortunes = (state.future.getNow.rows :+ Fortune(0, "Additional fortune added at request time."))
28+
.sortBy(_.message)
29+
val response = HttpResponse.builder.setContent(fortunes).build()
30+
stack.`return`(response)
31+
}
32+
33+
}
34+
35+
object FortuneController {
36+
37+
class FortuneRequest extends HttpRequest[Nothing, HttpResponse[Seq[Fortune]]]
38+
39+
private val SELECT_FORTUNE = "SELECT id, message from FORTUNE"
40+
41+
}

‎frameworks/Scala/otavia/benchmark/src/app/model/Fortune.scala

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package app.model
2+
3+
import cc.otavia.json.JsonSerde
4+
import cc.otavia.sql.{Row, RowDecoder}
5+
6+
/** The model for the "fortune" database table. */
7+
case class Fortune(id: Int, message: String) extends Row derives RowDecoder, JsonSerde

‎frameworks/Scala/otavia/benchmark/src/app/model/Message.scala

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
package app.model
2+
3+
import cc.otavia.json.JsonSerde
4+
5+
case class Message(message: String) derives JsonSerde

‎frameworks/Scala/otavia/benchmark/src/app/model/World.scala

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
package app.model
2+
3+
import cc.otavia.json.JsonSerde
4+
import cc.otavia.serde.annotation.rename
5+
import cc.otavia.sql.{Row, RowDecoder}
6+
7+
/** The model for the "world" database table. */
8+
case class World(id: Int, @rename("randomnumber") randomNumber: Int) extends Row derives RowDecoder, JsonSerde

‎frameworks/Scala/otavia/benchmark/src/app/startup.scala

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
package app
2+
3+
import app.controller.DBController.*
4+
import app.controller.FortuneController.*
5+
import app.controller.{DBController, FortuneController}
6+
import app.model.*
7+
import app.util.FortunesRender
8+
import cc.otavia.core.actor.ChannelsActor.{Bind, ChannelEstablished}
9+
import cc.otavia.core.actor.MainActor
10+
import cc.otavia.core.slf4a.LoggerFactory
11+
import cc.otavia.core.stack.helper.{FutureState, StartState}
12+
import cc.otavia.core.stack.{NoticeStack, StackYield}
13+
import cc.otavia.core.system.ActorSystem
14+
import cc.otavia.http.HttpMethod.*
15+
import cc.otavia.http.MediaType
16+
import cc.otavia.http.MediaType.*
17+
import cc.otavia.http.server.*
18+
import cc.otavia.http.server.Router.*
19+
import cc.otavia.json.JsonSerde
20+
import cc.otavia.serde.helper.BytesSerde
21+
import cc.otavia.sql.Connection
22+
23+
import java.io.File
24+
import java.nio.charset.StandardCharsets.UTF_8
25+
import java.nio.file.Path
26+
27+
private class ServerMain(val port: Int = 8080) extends MainActor(Array.empty) {
28+
29+
override def main0(stack: NoticeStack[MainActor.Args]): StackYield = stack.state match
30+
case _: StartState =>
31+
val worldResponseSerde = HttpResponseSerde.json(summon[JsonSerde[World]])
32+
val worldsResponseSerde = HttpResponseSerde.json(JsonSerde.derived[Seq[World]])
33+
val fortunesResponseSerde = HttpResponseSerde(new FortunesRender(), MediaType.TEXT_HTML_UTF8)
34+
35+
val dbController = autowire[DBController]()
36+
val fortuneController = autowire[FortuneController]()
37+
38+
val routers = Seq(
39+
// Test 6: plaintext
40+
constant[Array[Byte]](GET, "/plaintext", "Hello, World!".getBytes(UTF_8), BytesSerde, TEXT_PLAIN_UTF8),
41+
// Test 1: JSON serialization
42+
constant[Message](GET, "/json", Message("Hello, World!"), summon[JsonSerde[Message]], APP_JSON),
43+
// Test 2: Single database query.
44+
get("/db", dbController, () => new SingleQueryRequest(), worldResponseSerde),
45+
// Test 3: Multiple database queries
46+
get("/queries", dbController, () => new MultipleQueryRequest(), worldsResponseSerde),
47+
// Test 5: Database updates
48+
get("/updates", dbController, () => new UpdateRequest(), worldsResponseSerde),
49+
// Test 4: Fortunes
50+
get("/fortunes", fortuneController, () => new FortuneRequest(), fortunesResponseSerde)
51+
)
52+
val server = system.buildActor(() => new HttpServer(system.actorWorkerSize, routers))
53+
val state = FutureState[ChannelEstablished]()
54+
server.ask(Bind(port), state.future)
55+
stack.suspend(state)
56+
case state: FutureState[ChannelEstablished] =>
57+
if (state.future.isFailed) state.future.causeUnsafe.printStackTrace()
58+
logger.info(s"http server bind port $port success")
59+
stack.`return`()
60+
61+
}
62+
63+
@main def startup(url: String, user: String, password: String, poolSize: Int): Unit =
64+
val system = ActorSystem()
65+
val logger = LoggerFactory.getLogger("startup", system)
66+
logger.info("starting http server")
67+
system.buildActor(() => new Connection(url, user, password), global = true, num = poolSize)
68+
system.buildActor(() => new DBController(), global = true, num = system.actorWorkerSize)
69+
system.buildActor(() => new FortuneController(), global = true, num = system.actorWorkerSize)
70+
system.buildActor(() => new ServerMain())

‎frameworks/Scala/otavia/benchmark/src/app/util/FortunesRender.scala

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
package app.util
2+
3+
import app.model.Fortune
4+
import cc.otavia.buffer.{Buffer, BufferUtils}
5+
import cc.otavia.serde.Serde
6+
7+
import java.nio.charset.StandardCharsets
8+
import scala.annotation.switch
9+
10+
class FortunesRender extends Serde[Seq[Fortune]] {
11+
12+
private val text1 =
13+
"<!DOCTYPE html><html><head><title>Fortunes</title></head><body><table><tr><th>id</th><th>message</th></tr>"
14+
.getBytes(StandardCharsets.UTF_8)
15+
16+
private val text2 = "<tr><td>".getBytes(StandardCharsets.UTF_8)
17+
18+
private val text3 = "</td><td>".getBytes(StandardCharsets.UTF_8)
19+
20+
private val text4 = "</td></tr>".getBytes(StandardCharsets.UTF_8)
21+
22+
private val text5 = "</table></body></html>".getBytes(StandardCharsets.UTF_8)
23+
24+
private val lt = "&lt;".getBytes()
25+
private val gt = "&gt;".getBytes()
26+
private val quot = "&quot;".getBytes()
27+
private val squot = "&#39;".getBytes()
28+
private val amp = "&amp;".getBytes()
29+
30+
override def serialize(fortunes: Seq[Fortune], out: Buffer): Unit = {
31+
out.writeBytes(text1)
32+
for (fortune <- fortunes) {
33+
out.writeBytes(text2)
34+
BufferUtils.writeIntAsString(out, fortune.id)
35+
out.writeBytes(text3)
36+
writeEscapeMessage(out, fortune.message)
37+
out.writeBytes(text4)
38+
}
39+
out.writeBytes(text5)
40+
}
41+
42+
override def deserialize(in: Buffer): Seq[Fortune] = throw new UnsupportedOperationException()
43+
44+
private def writeEscapeMessage(buffer: Buffer, message: String): Unit = {
45+
var i = 0
46+
while (i < message.length) {
47+
val ch = message.charAt(i)
48+
writeChar(buffer, ch)
49+
i += 1
50+
}
51+
}
52+
53+
private def writeChar(buffer: Buffer, ch: Char): Unit = (ch: @switch) match
54+
case '<' => buffer.writeBytes(lt)
55+
case '>' => buffer.writeBytes(gt)
56+
case '"' => buffer.writeBytes(quot)
57+
case '\'' => buffer.writeBytes(squot)
58+
case '&' => buffer.writeBytes(amp)
59+
case _ =>
60+
if (ch < 0x80) buffer.writeByte(ch.toByte)
61+
else if (ch < 0x800) buffer.writeShortLE((ch >> 6 | (ch << 8 & 0x3f00) | 0x80c0).toShort)
62+
else buffer.writeMediumLE(ch >> 12 | (ch << 2 & 0x3f00) | (ch << 16 & 0x3f0000) | 0x8080e0)
63+
64+
}

0 commit comments

Comments
 (0)