This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-http.git
The following commit(s) were added to refs/heads/main by this push:
new 454e08d3b refactor: replace deprecated extends App with explicit main
method (#1131)
454e08d3b is described below
commit 454e08d3b51474d0b9e0b479c98adb0af83ca7a7
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jul 7 00:40:42 2026 +0800
refactor: replace deprecated extends App with explicit main method (#1131)
* refactor: replace deprecated `extends App` with explicit main method
Motivation:
`extends App` is deprecated in Scala 3. Replace with explicit `def main`
method for forward compatibility.
Modification:
Replace `object X extends App { ... }` with
`object X { def main(args: Array[String]): Unit = { ... } }`
in 14 test and example files.
Result:
No more usage of deprecated `scala.App` trait. Code is compatible with
Scala 3.
Tests:
Not run - docs only
References:
None - Scala 3 compatibility
* fix: fix forward reference errors in Http2ServerTest and TestServer
Motivation:
Moving lazy vals from object-level members into def main() body
introduced forward reference compilation errors since local vals
cannot reference later definitions in the same block.
Modification:
Move lazy val index/imagePage and def imagesBlock definitions before
the code that references them (syncHandler/routes).
Result:
No more forward reference compilation errors.
Tests:
sbt "http2-tests / Test / compile" "http-tests / Test / compile" - success
References:
Fixes CI failures in PR #1131
* fix: avoid forward reference after removing extends App
Motivation:
PR 1131 replaces deprecated Scala extends App usage with explicit main
methods, but moving Http2ClientApp logic into main turned the singleRequest
helper into a local method used before its definition.
Modification:
Move singleRequest back to object scope as a private helper while keeping
the explicit main method.
Result:
Http2ClientApp keeps the PR's explicit main entry point and compiles
without the local forward reference error.
Tests:
- rtk sbt "++2.13.18!" "docs / Test / compile" - passed
- rtk sbt "++3.3.8!" "docs / Test / compile" - passed
- rtk sbt "++3.8.4!" "docs / Test / compile" - passed
- rtk scalafmt --mode diff-ref=origin/main - passed
- rtk scalafmt --list --mode diff-ref=origin/main - passed
- rtk git diff --check - passed
- rtk sbt "docs / Test / headerCheck" - passed
References:
Refs #1131
---
.../scala/docs/http/scaladsl/Http2ClientApp.scala | 79 +++---
.../server/ExceptionHandlerExamplesSpec.scala | 40 +--
.../server/RejectionHandlerExamplesSpec.scala | 58 ++--
.../directives/JsonStreamingFullExamples.scala | 54 ++--
.../engine/parsing/HttpHeaderParserTestBed.scala | 56 ++--
.../http/impl/engine/ws/EchoTestClientApp.scala | 78 +++---
.../http/impl/engine/ws/WSClientAutobahnTest.scala | 306 +++++++++++----------
.../http/impl/engine/ws/WSServerAutobahnTest.scala | 56 ++--
.../apache/pekko/http/scaladsl/TestClient.scala | 168 +++++------
.../apache/pekko/http/scaladsl/TestServer.scala | 116 ++++----
.../pekko/http/scaladsl/TestSingleRequest.scala | 34 +--
.../pekko/http/scaladsl/server/TcpLeakApp.scala | 73 ++---
.../pekko/http/scaladsl/server/TestServer.scala | 178 ++++++------
.../pekko/http/scaladsl/Http2ServerTest.scala | 202 +++++++-------
14 files changed, 766 insertions(+), 732 deletions(-)
diff --git a/docs/src/test/scala/docs/http/scaladsl/Http2ClientApp.scala
b/docs/src/test/scala/docs/http/scaladsl/Http2ClientApp.scala
index af3d841ef..52d084250 100644
--- a/docs/src/test/scala/docs/http/scaladsl/Http2ClientApp.scala
+++ b/docs/src/test/scala/docs/http/scaladsl/Http2ClientApp.scala
@@ -34,50 +34,54 @@ import scala.concurrent.Future
import scala.concurrent.Promise
/** A small example app that shows how to use the HTTP/2 client API currently
against actual internet servers */
-object Http2ClientApp extends App {
- val config =
- ConfigFactory.parseString(
- """
- # pekko.loglevel = debug
- pekko.http.client.http2.log-frames = true
- pekko.http.client.parsing.max-content-length = 20m
- """).withFallback(ConfigFactory.defaultApplication())
-
- implicit val system: ActorSystem = ActorSystem("Http2ClientApp", config)
- implicit val ec: ExecutionContext = system.dispatcher
+object Http2ClientApp {
+ def main(args: Array[String]): Unit = {
+ val config =
+ ConfigFactory.parseString(
+ """
+ # pekko.loglevel = debug
+ pekko.http.client.http2.log-frames = true
+ pekko.http.client.parsing.max-content-length = 20m
+ """).withFallback(ConfigFactory.defaultApplication())
+
+ implicit val system: ActorSystem = ActorSystem("Http2ClientApp", config)
+ implicit val ec: ExecutionContext = system.dispatcher
+
+ // #response-future-association
+ val dispatch =
singleRequest(Http().connectionTo("pekko.apache.org").http2())
+
+ dispatch(
+ HttpRequest(
+ uri =
"https://pekko.apache.org/api/pekko/current/org/apache/pekko/actor/typed/scaladsl/index.html";,
+ headers = headers.`Accept-Encoding`(HttpEncodings.gzip) ::
Nil)).onComplete { res =>
+ println(s"[1] Got index.html: $res")
+ res.get.entity.dataBytes.runWith(Sink.ignore).onComplete(res =>
println(s"Finished reading [1] $res"))
+ }
- // #response-future-association
- val dispatch = singleRequest(Http().connectionTo("pekko.apache.org").http2())
-
- dispatch(
- HttpRequest(
- uri =
"https://pekko.apache.org/api/pekko/current/org/apache/pekko/actor/typed/scaladsl/index.html";,
- headers = headers.`Accept-Encoding`(HttpEncodings.gzip) ::
Nil)).onComplete { res =>
- println(s"[1] Got index.html: $res")
- res.get.entity.dataBytes.runWith(Sink.ignore).onComplete(res =>
println(s"Finished reading [1] $res"))
- }
+ // #response-future-association
- // #response-future-association
+ dispatch(
+ HttpRequest(
+ uri = "https://pekko.apache.org/api/pekko/current/index.js";,
+ headers = /*headers.`Accept-Encoding`(HttpEncodings.gzip) ::*/
Nil)).onComplete { res =>
+ println(s"[2] Got index.js: $res")
+ res.get.entity.dataBytes.runWith(Sink.ignore).onComplete(res =>
println(s"Finished reading [2] $res"))
+ }
- dispatch(
- HttpRequest(
- uri = "https://pekko.apache.org/api/pekko/current/index.js";,
- headers = /*headers.`Accept-Encoding`(HttpEncodings.gzip) ::*/
Nil)).onComplete { res =>
- println(s"[2] Got index.js: $res")
- res.get.entity.dataBytes.runWith(Sink.ignore).onComplete(res =>
println(s"Finished reading [2] $res"))
- }
+ dispatch(HttpRequest(uri =
"https://pekko.apache.org/api/pekko/current/lib/MaterialIcons-Regular.woff";))
+ .flatMap(_.toStrict(1.second))
+ .onComplete(res => println(s"[3] Got font: $res"))
- dispatch(HttpRequest(uri =
"https://pekko.apache.org/api/pekko/current/lib/MaterialIcons-Regular.woff";))
- .flatMap(_.toStrict(1.second))
- .onComplete(res => println(s"[3] Got font: $res"))
+ dispatch(HttpRequest(uri = "https://pekko.apache.org/favicon.ico";))
+ .flatMap(_.toStrict(1.second))
+ .onComplete(res => println(s"[4] Got favicon: $res"))
- dispatch(HttpRequest(uri = "https://pekko.apache.org/favicon.ico";))
- .flatMap(_.toStrict(1.second))
- .onComplete(res => println(s"[4] Got favicon: $res"))
+ }
// #response-future-association
- def singleRequest(
- connection: Flow[HttpRequest, HttpResponse, Any], bufferSize: Int =
100): HttpRequest => Future[HttpResponse] = {
+ private def singleRequest(
+ connection: Flow[HttpRequest, HttpResponse, Any], bufferSize: Int =
100)(implicit system: ActorSystem)
+ : HttpRequest => Future[HttpResponse] = {
val queue =
Source.queue(bufferSize)
.via(connection)
@@ -102,5 +106,4 @@ object Http2ClientApp extends App {
}
}
// #response-future-association
-
}
diff --git
a/docs/src/test/scala/docs/http/scaladsl/server/ExceptionHandlerExamplesSpec.scala
b/docs/src/test/scala/docs/http/scaladsl/server/ExceptionHandlerExamplesSpec.scala
index dc5aeb546..992d9f0b7 100644
---
a/docs/src/test/scala/docs/http/scaladsl/server/ExceptionHandlerExamplesSpec.scala
+++
b/docs/src/test/scala/docs/http/scaladsl/server/ExceptionHandlerExamplesSpec.scala
@@ -39,17 +39,19 @@ object MyExplicitExceptionHandler {
}
}
- object MyApp extends App {
+ object MyApp {
+ def main(args: Array[String]): Unit = {
- implicit val system: ActorSystem = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
- val route: Route =
- handleExceptions(myExceptionHandler) {
- // ... some route structure
- null // #hide
- }
+ val route: Route =
+ handleExceptions(myExceptionHandler) {
+ // ... some route structure
+ null // #hide
+ }
- Http().newServerAt("localhost", 8080).bind(route)
+ Http().newServerAt("localhost", 8080).bind(route)
+ }
}
//#explicit-handler-example
@@ -75,15 +77,17 @@ object MyImplicitExceptionHandler {
}
}
- object MyApp extends App {
+ object MyApp {
+ def main(args: Array[String]): Unit = {
- implicit val system: ActorSystem = ActorSystem()
+ implicit val system: ActorSystem = ActorSystem()
- val route: Route =
- // ... some route structure
- null // #hide
+ val route: Route =
+ // ... some route structure
+ null // #hide
- Http().newServerAt("localhost", 8080).bind(route)
+ Http().newServerAt("localhost", 8080).bind(route)
+ }
}
//#implicit-handler-example
@@ -162,10 +166,12 @@ object RespondWithHeaderExceptionHandlerExample {
}
}
- object MyApp extends App {
- implicit val system: ActorSystem = ActorSystem()
+ object MyApp {
+ def main(args: Array[String]): Unit = {
+ implicit val system: ActorSystem = ActorSystem()
- Http().newServerAt("localhost", 8080).bind(route)
+ Http().newServerAt("localhost", 8080).bind(route)
+ }
}
//#respond-with-header-exceptionhandler-example
}
diff --git
a/docs/src/test/scala/docs/http/scaladsl/server/RejectionHandlerExamplesSpec.scala
b/docs/src/test/scala/docs/http/scaladsl/server/RejectionHandlerExamplesSpec.scala
index dc06f9c01..2c33e0290 100644
---
a/docs/src/test/scala/docs/http/scaladsl/server/RejectionHandlerExamplesSpec.scala
+++
b/docs/src/test/scala/docs/http/scaladsl/server/RejectionHandlerExamplesSpec.scala
@@ -30,36 +30,38 @@ object MyRejectionHandler {
import StatusCodes._
import Directives._
- object MyApp extends App {
- def myRejectionHandler =
- RejectionHandler.newBuilder()
- .handle {
- case MissingCookieRejection(cookieName) =>
- complete(HttpResponse(BadRequest, entity = "No cookies, no
service!!!"))
- }
- .handle {
- case AuthorizationFailedRejection =>
- complete(Forbidden, "You're out of your depth!")
- }
- .handle {
- case ValidationRejection(msg, _) =>
- complete(InternalServerError, "That wasn't valid! " + msg)
- }
- .handleAll[MethodRejection] { methodRejections =>
- val names = methodRejections.map(_.supported.name)
- complete(MethodNotAllowed, s"Can't do that! Supported:
${names.mkString(" or ")}!")
- }
- .handleNotFound { complete((NotFound, "Not here!")) }
- .result()
-
- implicit val system: ActorSystem = ActorSystem()
+ object MyApp {
+ def main(args: Array[String]): Unit = {
+ def myRejectionHandler =
+ RejectionHandler.newBuilder()
+ .handle {
+ case MissingCookieRejection(cookieName) =>
+ complete(HttpResponse(BadRequest, entity = "No cookies, no
service!!!"))
+ }
+ .handle {
+ case AuthorizationFailedRejection =>
+ complete(Forbidden, "You're out of your depth!")
+ }
+ .handle {
+ case ValidationRejection(msg, _) =>
+ complete(InternalServerError, "That wasn't valid! " + msg)
+ }
+ .handleAll[MethodRejection] { methodRejections =>
+ val names = methodRejections.map(_.supported.name)
+ complete(MethodNotAllowed, s"Can't do that! Supported:
${names.mkString(" or ")}!")
+ }
+ .handleNotFound { complete((NotFound, "Not here!")) }
+ .result()
+
+ implicit val system: ActorSystem = ActorSystem()
+
+ val route: Route = handleRejections(myRejectionHandler) {
+ // ... some route structure
+ null // #hide
+ }
- val route: Route = handleRejections(myRejectionHandler) {
- // ... some route structure
- null // #hide
+ Http().newServerAt("localhost", 8080).bind(route)
}
-
- Http().newServerAt("localhost", 8080).bind(route)
}
// #custom-handler-example
}
diff --git
a/docs/src/test/scala/docs/http/scaladsl/server/directives/JsonStreamingFullExamples.scala
b/docs/src/test/scala/docs/http/scaladsl/server/directives/JsonStreamingFullExamples.scala
index cb2874a47..c2d93a26a 100644
---
a/docs/src/test/scala/docs/http/scaladsl/server/directives/JsonStreamingFullExamples.scala
+++
b/docs/src/test/scala/docs/http/scaladsl/server/directives/JsonStreamingFullExamples.scala
@@ -54,35 +54,37 @@ class JsonStreamingFullExamples extends AnyWordSpec {
})
}
- object ApiServer extends App with UserProtocol {
- implicit val system: ActorSystem = ActorSystem("api")
- implicit val executionContext: ExecutionContext = system.dispatcher
-
- implicit val jsonStreamingSupport: JsonEntityStreamingSupport =
EntityStreamingSupport.json()
- .withContentType(ct)
- .withParallelMarshalling(parallelism = 10, unordered = false)
-
- // (fake) async database query api
- def dummyUser(id: String) = User(s"User $id", id.toString)
-
- def fetchUsers(): Source[User, NotUsed] = Source.fromIterator(() =>
- Iterator.fill(10000) {
- val id = Random.nextInt()
- dummyUser(id.toString)
- })
-
- val route =
- pathPrefix("users") {
- get {
- complete(fetchUsers())
+ object ApiServer extends UserProtocol {
+ def main(args: Array[String]): Unit = {
+ implicit val system: ActorSystem = ActorSystem("api")
+ implicit val executionContext: ExecutionContext = system.dispatcher
+
+ implicit val jsonStreamingSupport: JsonEntityStreamingSupport =
EntityStreamingSupport.json()
+ .withContentType(ct)
+ .withParallelMarshalling(parallelism = 10, unordered = false)
+
+ // (fake) async database query api
+ def dummyUser(id: String) = User(s"User $id", id.toString)
+
+ def fetchUsers(): Source[User, NotUsed] = Source.fromIterator(() =>
+ Iterator.fill(10000) {
+ val id = Random.nextInt()
+ dummyUser(id.toString)
+ })
+
+ val route =
+ pathPrefix("users") {
+ get {
+ complete(fetchUsers())
+ }
}
- }
- val bindingFuture = Http().newServerAt("localhost", 8080).bind(route)
+ val bindingFuture = Http().newServerAt("localhost", 8080).bind(route)
- println(s"Server online at http://localhost:8080/\nPress RETURN to
stop...")
- StdIn.readLine()
- bindingFuture.flatMap(_.unbind()).onComplete(_ => system.terminate())
+ println(s"Server online at http://localhost:8080/\nPress RETURN to
stop...")
+ StdIn.readLine()
+ bindingFuture.flatMap(_.unbind()).onComplete(_ => system.terminate())
+ }
}
// #custom-content-type
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/parsing/HttpHeaderParserTestBed.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/parsing/HttpHeaderParserTestBed.scala
index 78daefe64..e8a4c522f 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/parsing/HttpHeaderParserTestBed.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/parsing/HttpHeaderParserTestBed.scala
@@ -18,32 +18,34 @@ import pekko.actor.ActorSystem
import pekko.http.scaladsl.settings.ParserSettings
import com.typesafe.config.{ Config, ConfigFactory }
-object HttpHeaderParserTestBed extends App {
-
- val testConf: Config = ConfigFactory.parseString("""
- pekko.event-handlers = ["org.apache.pekko.testkit.TestEventListener"]
- pekko.loglevel = ERROR
- pekko.http.parsing.max-header-name-length = 20
- pekko.http.parsing.max-header-value-length = 21
- pekko.http.parsing.header-cache.Host = 300""")
- val system: ActorSystem = ActorSystem("HttpHeaderParserTestBed", testConf)
-
- val parser = HttpHeaderParser.prime {
- HttpHeaderParser.unprimed(ParserSettings(system), system.log,
- warnOnIllegalHeader = info => system.log.warning(info.formatPretty))
+object HttpHeaderParserTestBed {
+ def main(args: Array[String]): Unit = {
+
+ val testConf: Config = ConfigFactory.parseString("""
+ pekko.event-handlers = ["org.apache.pekko.testkit.TestEventListener"]
+ pekko.loglevel = ERROR
+ pekko.http.parsing.max-header-name-length = 20
+ pekko.http.parsing.max-header-value-length = 21
+ pekko.http.parsing.header-cache.Host = 300""")
+ val system: ActorSystem = ActorSystem("HttpHeaderParserTestBed", testConf)
+
+ val parser = HttpHeaderParser.prime {
+ HttpHeaderParser.unprimed(ParserSettings(system), system.log,
+ warnOnIllegalHeader = info => system.log.warning(info.formatPretty))
+ }
+
+ println {
+ s"""
+ |HttpHeaderParser primed Trie
+ |----------------------------
+ |
+ |%TRIE%
+ |
+ |formatSizes: ${parser.formatSizes}
+ |contentHistogram: ${parser.contentHistogram.mkString("\n ", "\n ",
"\n")}
+ """.stripMargin.replace("%TRIE%", parser.formatTrie)
+ }
+
+ system.terminate()
}
-
- println {
- s"""
- |HttpHeaderParser primed Trie
- |----------------------------
- |
- |%TRIE%
- |
- |formatSizes: ${parser.formatSizes}
- |contentHistogram: ${parser.contentHistogram.mkString("\n ", "\n ",
"\n")}
- """.stripMargin.replace("%TRIE%", parser.formatTrie)
- }
-
- system.terminate()
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/EchoTestClientApp.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/EchoTestClientApp.scala
index de81c7678..a89b643de 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/EchoTestClientApp.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/EchoTestClientApp.scala
@@ -30,49 +30,51 @@ import scala.util.{ Failure, Success }
/**
* An example App that runs a quick test against the websocket server at
wss://echo.websocket.org
*/
-object EchoTestClientApp extends App {
- implicit val system: ActorSystem = ActorSystem()
- import system.dispatcher
+object EchoTestClientApp {
+ def main(args: Array[String]): Unit = {
+ implicit val system: ActorSystem = ActorSystem()
+ import system.dispatcher
- def delayedCompletion(delay: FiniteDuration): Source[Nothing, NotUsed] =
- Source.single(1)
- .mapAsync(1)(_ => pekko.pattern.after(delay,
system.scheduler)(Future(1)))
- .drop(1).asInstanceOf[Source[Nothing, NotUsed]]
+ def delayedCompletion(delay: FiniteDuration): Source[Nothing, NotUsed] =
+ Source.single(1)
+ .mapAsync(1)(_ => pekko.pattern.after(delay,
system.scheduler)(Future(1)))
+ .drop(1).asInstanceOf[Source[Nothing, NotUsed]]
- def messages: List[Message] =
- List(
- TextMessage("Test 1"),
- BinaryMessage(ByteString("abc")),
- TextMessage("Test 2"),
- BinaryMessage(ByteString("def")))
+ def messages: List[Message] =
+ List(
+ TextMessage("Test 1"),
+ BinaryMessage(ByteString("abc")),
+ TextMessage("Test 2"),
+ BinaryMessage(ByteString("def")))
- def source: Source[Message, NotUsed] =
- Source(messages) ++ delayedCompletion(1.second) // otherwise, we may start
closing too soon
+ def source: Source[Message, NotUsed] =
+ Source(messages) ++ delayedCompletion(1.second) // otherwise, we may
start closing too soon
- def sink: Sink[Message, Future[Seq[String]]] =
- Flow[Message]
- .mapAsync(1) {
- case tm: TextMessage =>
- tm.textStream.runWith(Sink.fold("")(_ + _)).map(str =>
s"TextMessage: '$str'")
- case bm: BinaryMessage =>
- bm.dataStream.runWith(Sink.fold(ByteString.empty)(_ ++ _)).map(bs =>
s"BinaryMessage: '${bs.utf8String}'")
- }
- .grouped(10000)
- .toMat(Sink.head)(Keep.right)
+ def sink: Sink[Message, Future[Seq[String]]] =
+ Flow[Message]
+ .mapAsync(1) {
+ case tm: TextMessage =>
+ tm.textStream.runWith(Sink.fold("")(_ + _)).map(str =>
s"TextMessage: '$str'")
+ case bm: BinaryMessage =>
+ bm.dataStream.runWith(Sink.fold(ByteString.empty)(_ ++ _)).map(bs
=> s"BinaryMessage: '${bs.utf8String}'")
+ }
+ .grouped(10000)
+ .toMat(Sink.head)(Keep.right)
- def echoClient = Flow.fromSinkAndSourceMat(sink, source)(Keep.left)
+ def echoClient = Flow.fromSinkAndSourceMat(sink, source)(Keep.left)
- val (upgrade, res) =
Http().singleWebSocketRequest("wss://echo.websocket.org", echoClient)
- res.onComplete {
- case Success(res) =>
- println("Run successful. Got these elements:")
- res.foreach(println)
- system.terminate()
- case Failure(e) =>
- println("Run failed.")
- e.printStackTrace()
- system.terminate()
- }
+ val (upgrade, res) =
Http().singleWebSocketRequest("wss://echo.websocket.org", echoClient)
+ res.onComplete {
+ case Success(res) =>
+ println("Run successful. Got these elements:")
+ res.foreach(println)
+ system.terminate()
+ case Failure(e) =>
+ println("Run failed.")
+ e.printStackTrace()
+ system.terminate()
+ }
- system.scheduler.scheduleOnce(10.seconds)(system.terminate())
+ system.scheduler.scheduleOnce(10.seconds)(system.terminate())
+ }
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WSClientAutobahnTest.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WSClientAutobahnTest.scala
index aec0fe995..5d002308d 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WSClientAutobahnTest.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WSClientAutobahnTest.scala
@@ -25,188 +25,190 @@ import pekko.http.scaladsl.Http
import pekko.http.scaladsl.model.Uri
import pekko.http.scaladsl.model.ws._
-object WSClientAutobahnTest extends App {
- implicit val system: ActorSystem = ActorSystem()
- import system.dispatcher
+object WSClientAutobahnTest {
+ def main(args: Array[String]): Unit = {
+ implicit val system: ActorSystem = ActorSystem()
+ import system.dispatcher
- val Agent = "pekko-http"
- val Parallelism = 4
+ val Agent = "pekko-http"
+ val Parallelism = 4
- val getCaseCountUri: Uri =
- s"ws://localhost:9001/getCaseCount"
+ val getCaseCountUri: Uri =
+ s"ws://localhost:9001/getCaseCount"
- def runCaseUri(caseIndex: Int, agent: String): Uri =
- s"ws://localhost:9001/runCase?case=$caseIndex&agent=$agent"
+ def runCaseUri(caseIndex: Int, agent: String): Uri =
+ s"ws://localhost:9001/runCase?case=$caseIndex&agent=$agent"
- def getCaseStatusUri(caseIndex: Int, agent: String): Uri =
- s"ws://localhost:9001/getCaseStatus?case=$caseIndex&agent=$agent"
+ def getCaseStatusUri(caseIndex: Int, agent: String): Uri =
+ s"ws://localhost:9001/getCaseStatus?case=$caseIndex&agent=$agent"
- def getCaseInfoUri(caseIndex: Int): Uri =
- s"ws://localhost:9001/getCaseInfo?case=$caseIndex"
+ def getCaseInfoUri(caseIndex: Int): Uri =
+ s"ws://localhost:9001/getCaseInfo?case=$caseIndex"
- def updateReportsUri(agent: String): Uri =
- s"ws://localhost:9001/updateReports?agent=$agent"
+ def updateReportsUri(agent: String): Uri =
+ s"ws://localhost:9001/updateReports?agent=$agent"
- def runCase(caseIndex: Int, agent: String = Agent): Future[CaseStatus] =
- runWs(runCaseUri(caseIndex, agent), echo).recover { case _ => () }.flatMap
{ _ =>
- getCaseStatus(caseIndex, agent)
- }
+ def runCase(caseIndex: Int, agent: String = Agent): Future[CaseStatus] =
+ runWs(runCaseUri(caseIndex, agent), echo).recover { case _ => ()
}.flatMap { _ =>
+ getCaseStatus(caseIndex, agent)
+ }
- def richRunCase(caseIndex: Int, agent: String = Agent): Future[CaseResult] =
{
- val info = getCaseInfo(caseIndex)
- val startMillis = System.currentTimeMillis()
- val status = runCase(caseIndex, agent).map { res =>
- val lastedMillis = System.currentTimeMillis() - startMillis
- (res, lastedMillis)
+ def richRunCase(caseIndex: Int, agent: String = Agent): Future[CaseResult]
= {
+ val info = getCaseInfo(caseIndex)
+ val startMillis = System.currentTimeMillis()
+ val status = runCase(caseIndex, agent).map { res =>
+ val lastedMillis = System.currentTimeMillis() - startMillis
+ (res, lastedMillis)
+ }
+ import Console._
+ info.flatMap { i =>
+ val prefix = f"$YELLOW${i.caseInfo.id}%-7s$RESET -
$RESET${i.caseInfo.description}$RESET ... "
+
+ status.onComplete {
+ case Success((CaseStatus(status), millis)) =>
+ val color = if (status == "OK") GREEN else RED
+ println(f"${color}$status%-15s$RESET$millis%5d ms $prefix")
+ case Failure(e) =>
+ println(s"$prefix${RED}failed with '${e.getMessage}'$RESET")
+ }
+
+ status.map(s => CaseResult(i.caseInfo, s._1))
+ }
}
- import Console._
- info.flatMap { i =>
- val prefix = f"$YELLOW${i.caseInfo.id}%-7s$RESET -
$RESET${i.caseInfo.description}$RESET ... "
- status.onComplete {
- case Success((CaseStatus(status), millis)) =>
- val color = if (status == "OK") GREEN else RED
- println(f"${color}$status%-15s$RESET$millis%5d ms $prefix")
- case Failure(e) =>
- println(s"$prefix${RED}failed with '${e.getMessage}'$RESET")
+ def getCaseCount(): Future[Int] =
+ runToSingleText(getCaseCountUri).map(_.toInt)
+
+ def getCaseInfo(caseId: Int): Future[IndexedCaseInfo] =
+
runToSingleJsonValue[CaseInfo](getCaseInfoUri(caseId)).map(IndexedCaseInfo(caseId,
_))
+
+ def getCaseStatus(caseId: Int, agent: String = Agent): Future[CaseStatus] =
+ runToSingleJsonValue[CaseStatus](getCaseStatusUri(caseId, agent))
+
+ def updateReports(agent: String = Agent): Future[Unit] =
+ runToSingleText(updateReportsUri(agent)).map(_ => ())
+
+ /**
+ * Map from textual case ID (like 1.1.1) to IndexedCaseInfo
+ * @return
+ */
+ def getCaseMap(): Future[Map[String, IndexedCaseInfo]] = {
+ val res =
+ getCaseCount().flatMap { count =>
+ println(s"Retrieving case info for $count cases...")
+ Future.traverse(1 to count)(getCaseInfo).map(_.map(e =>
e.caseInfo.id -> e).toMap)
+ }
+ res.foreach { res =>
+ println(s"Received info for ${res.size} cases")
}
-
- status.map(s => CaseResult(i.caseInfo, s._1))
+ res
}
- }
-
- def getCaseCount(): Future[Int] =
- runToSingleText(getCaseCountUri).map(_.toInt)
-
- def getCaseInfo(caseId: Int): Future[IndexedCaseInfo] =
-
runToSingleJsonValue[CaseInfo](getCaseInfoUri(caseId)).map(IndexedCaseInfo(caseId,
_))
-
- def getCaseStatus(caseId: Int, agent: String = Agent): Future[CaseStatus] =
- runToSingleJsonValue[CaseStatus](getCaseStatusUri(caseId, agent))
- def updateReports(agent: String = Agent): Future[Unit] =
- runToSingleText(updateReportsUri(agent)).map(_ => ())
+ def echo = Flow[Message].viaMat(completionSignal)(Keep.right)
- /**
- * Map from textual case ID (like 1.1.1) to IndexedCaseInfo
- * @return
- */
- def getCaseMap(): Future[Map[String, IndexedCaseInfo]] = {
- val res =
+ import Console._
+ if (args.size >= 1) {
+ // run one
+ val testId = args(0)
+ println(s"Trying to run test $testId")
+ getCaseMap().flatMap { map =>
+ val info = map(testId)
+ richRunCase(info.index)
+ }.onComplete {
+ case Success(res) =>
+ println(s"[OK] Run successfully finished!")
+ updateReportsAndShutdown()
+ case Failure(e) =>
+ println(s"[${RED}FAILED$RESET] Run failed with this exception: ")
+ e.printStackTrace()
+ updateReportsAndShutdown()
+ }
+ } else {
+ println("Running complete test suite")
getCaseCount().flatMap { count =>
- println(s"Retrieving case info for $count cases...")
- Future.traverse(1 to count)(getCaseInfo).map(_.map(e => e.caseInfo.id
-> e).toMap)
+ println(s"Found $count tests.")
+ Source(1 to
count).mapAsyncUnordered(Parallelism)(richRunCase(_)).grouped(count).runWith(Sink.head)
+ }.map { results =>
+ val grouped =
+ results.groupBy(_.status.behavior)
+
+ println(s"${results.size} tests run.")
+ println()
+ println(s"${GREEN}OK$RESET: ${grouped.getOrElse("OK", Nil).size}")
+ val notOk = grouped.filterNot(_._1 == "OK")
+ notOk.toSeq.sortBy(_._2.size).foreach {
+ case (status, cases) => println(s"$RED$status$RESET: ${cases.size}")
+ }
+ println()
+ println("Not OK tests: ")
+ println()
+ results.filterNot(_.status.behavior == "OK").foreach { r =>
+ println(f"$RED${r.status.behavior}%-20s$RESET
$YELLOW${r.info.id}%-7s$RESET - $RESET${r.info.description}")
+ }
+
+ ()
}
- res.foreach { res =>
- println(s"Received info for ${res.size} cases")
+ .onComplete(completion)
}
- res
- }
- def echo = Flow[Message].viaMat(completionSignal)(Keep.right)
-
- import Console._
- if (args.size >= 1) {
- // run one
- val testId = args(0)
- println(s"Trying to run test $testId")
- getCaseMap().flatMap { map =>
- val info = map(testId)
- richRunCase(info.index)
- }.onComplete {
+ def completion[T]: Try[T] => Unit = {
case Success(res) =>
- println(s"[OK] Run successfully finished!")
+ println(s"Run successfully finished!")
updateReportsAndShutdown()
case Failure(e) =>
- println(s"[${RED}FAILED$RESET] Run failed with this exception: ")
+ println("Run failed with this exception")
e.printStackTrace()
updateReportsAndShutdown()
}
- } else {
- println("Running complete test suite")
- getCaseCount().flatMap { count =>
- println(s"Found $count tests.")
- Source(1 to
count).mapAsyncUnordered(Parallelism)(richRunCase(_)).grouped(count).runWith(Sink.head)
- }.map { results =>
- val grouped =
- results.groupBy(_.status.behavior)
-
- println(s"${results.size} tests run.")
- println()
- println(s"${GREEN}OK$RESET: ${grouped.getOrElse("OK", Nil).size}")
- val notOk = grouped.filterNot(_._1 == "OK")
- notOk.toSeq.sortBy(_._2.size).foreach {
- case (status, cases) => println(s"$RED$status$RESET: ${cases.size}")
- }
- println()
- println("Not OK tests: ")
- println()
- results.filterNot(_.status.behavior == "OK").foreach { r =>
- println(f"$RED${r.status.behavior}%-20s$RESET
$YELLOW${r.info.id}%-7s$RESET - $RESET${r.info.description}")
+ def updateReportsAndShutdown(): Unit =
+ updateReports().onComplete { res =>
+ println("Reports should now be accessible at
http://localhost:8080/cwd/reports/clients/index.html";)
+ system.terminate()
}
- ()
+ import scala.concurrent.duration._
+ import system.dispatcher
+ system.scheduler.scheduleOnce(60.seconds)(system.terminate())
+
+ def runWs[T](uri: Uri, clientFlow: Flow[Message, Message, T]): T =
+ Http().singleWebSocketRequest(uri, clientFlow)._2
+
+ def completionSignal[T]: Flow[T, T, Future[Done]] =
+ Flow[T].watchTermination((_, res) => res)
+
+ /**
+ * The autobahn tests define a weird API where every request must be a
WebSocket request and
+ * they will send a single websocket message with the result. WebSocket
everywhere? Strange,
+ * but somewhat consistent.
+ */
+ def runToSingleText(uri: Uri): Future[String] = {
+ val sink = Sink.head[Message]
+ runWs(uri, Flow.fromSinkAndSourceMat(sink,
Source.maybe[Message])(Keep.left)).flatMap {
+ case tm: TextMessage => tm.textStream.runWith(Sink.fold("")(_ + _))
+ case other =>
+ throw new IllegalStateException(s"unexpected element of type
${other.getClass}")
+ }
}
- .onComplete(completion)
- }
+ def runToSingleJsonValue[T: JsonReader](uri: Uri): Future[T] =
+ runToSingleText(uri).map(_.parseJson.convertTo[T])
- def completion[T]: Try[T] => Unit = {
- case Success(res) =>
- println(s"Run successfully finished!")
- updateReportsAndShutdown()
- case Failure(e) =>
- println("Run failed with this exception")
- e.printStackTrace()
- updateReportsAndShutdown()
- }
- def updateReportsAndShutdown(): Unit =
- updateReports().onComplete { res =>
- println("Reports should now be accessible at
http://localhost:8080/cwd/reports/clients/index.html";)
- system.terminate()
- }
+ case class IndexedCaseInfo(index: Int, caseInfo: CaseInfo)
+ case class CaseResult(info: CaseInfo, status: CaseStatus)
- import scala.concurrent.duration._
- import system.dispatcher
- system.scheduler.scheduleOnce(60.seconds)(system.terminate())
-
- def runWs[T](uri: Uri, clientFlow: Flow[Message, Message, T]): T =
- Http().singleWebSocketRequest(uri, clientFlow)._2
-
- def completionSignal[T]: Flow[T, T, Future[Done]] =
- Flow[T].watchTermination((_, res) => res)
-
- /**
- * The autobahn tests define a weird API where every request must be a
WebSocket request and
- * they will send a single websocket message with the result. WebSocket
everywhere? Strange,
- * but somewhat consistent.
- */
- def runToSingleText(uri: Uri): Future[String] = {
- val sink = Sink.head[Message]
- runWs(uri, Flow.fromSinkAndSourceMat(sink,
Source.maybe[Message])(Keep.left)).flatMap {
- case tm: TextMessage => tm.textStream.runWith(Sink.fold("")(_ + _))
- case other =>
- throw new IllegalStateException(s"unexpected element of type
${other.getClass}")
+ // {"behavior": "OK"}
+ case class CaseStatus(behavior: String) {
+ def isSuccessful: Boolean = behavior == "OK"
+ }
+ object CaseStatus {
+ import DefaultJsonProtocol._
+ implicit def caseStatusFormat: JsonFormat[CaseStatus] =
jsonFormat1(CaseStatus.apply)
}
- }
- def runToSingleJsonValue[T: JsonReader](uri: Uri): Future[T] =
- runToSingleText(uri).map(_.parseJson.convertTo[T])
-
- case class IndexedCaseInfo(index: Int, caseInfo: CaseInfo)
- case class CaseResult(info: CaseInfo, status: CaseStatus)
-
- // {"behavior": "OK"}
- case class CaseStatus(behavior: String) {
- def isSuccessful: Boolean = behavior == "OK"
- }
- object CaseStatus {
- import DefaultJsonProtocol._
- implicit def caseStatusFormat: JsonFormat[CaseStatus] =
jsonFormat1(CaseStatus.apply)
- }
- // {"id": "1.1.1", "description": "Send text message with payload 0."}
- case class CaseInfo(id: String, description: String)
- object CaseInfo {
- import DefaultJsonProtocol._
- implicit def caseInfoFormat: JsonFormat[CaseInfo] =
jsonFormat2(CaseInfo.apply)
+ // {"id": "1.1.1", "description": "Send text message with payload 0."}
+ case class CaseInfo(id: String, description: String)
+ object CaseInfo {
+ import DefaultJsonProtocol._
+ implicit def caseInfoFormat: JsonFormat[CaseInfo] =
jsonFormat2(CaseInfo.apply)
+ }
}
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WSServerAutobahnTest.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WSServerAutobahnTest.scala
index 8fa78f7fc..39eff7427 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WSServerAutobahnTest.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WSServerAutobahnTest.scala
@@ -28,34 +28,36 @@ import pekko.stream.scaladsl.Flow
import scala.io.StdIn
-object WSServerAutobahnTest extends App {
- implicit val system: ActorSystem = ActorSystem("WSServerTest")
-
- val host = System.getProperty("pekko.ws-host", "127.0.0.1")
- val port = System.getProperty("pekko.ws-port", "9001").toInt
- val mode = System.getProperty("pekko.ws-mode", "read") // read or sleep
-
- try {
- val binding = Http().newServerAt(host, port).bindSync {
- case req @ HttpRequest(GET, Uri.Path("/"), _, _, _) if
req.attribute(webSocketUpgrade).isDefined =>
- req.attribute(webSocketUpgrade) match {
- case Some(upgrade) => upgrade.handleMessages(echoWebSocketService)
// needed for running the autobahn test suite
- case None => HttpResponse(400, entity = "Not a valid
websocket request!")
- }
- case _: HttpRequest => HttpResponse(404, entity = "Unknown resource!")
+object WSServerAutobahnTest {
+ def main(args: Array[String]): Unit = {
+ implicit val system: ActorSystem = ActorSystem("WSServerTest")
+
+ val host = System.getProperty("pekko.ws-host", "127.0.0.1")
+ val port = System.getProperty("pekko.ws-port", "9001").toInt
+ val mode = System.getProperty("pekko.ws-mode", "read") // read or sleep
+
+ try {
+ val binding = Http().newServerAt(host, port).bindSync {
+ case req @ HttpRequest(GET, Uri.Path("/"), _, _, _) if
req.attribute(webSocketUpgrade).isDefined =>
+ req.attribute(webSocketUpgrade) match {
+ case Some(upgrade) => upgrade.handleMessages(echoWebSocketService)
// needed for running the autobahn test suite
+ case None => HttpResponse(400, entity = "Not a valid
websocket request!")
+ }
+ case _: HttpRequest => HttpResponse(404, entity = "Unknown resource!")
+ }
+
+ Await.result(binding, 3.second) // throws if binding fails
+ println(s"Server online at http://$host:$port";)
+ mode match {
+ case "sleep" => while (true) Thread.sleep(1.minute.toMillis)
+ case "read" => StdIn.readLine("Press RETURN to stop...")
+ case _ => throw new Exception("pekko.ws-mode MUST be sleep or
read.")
+ }
+ } finally {
+ system.terminate()
}
- Await.result(binding, 3.second) // throws if binding fails
- println(s"Server online at http://$host:$port";)
- mode match {
- case "sleep" => while (true) Thread.sleep(1.minute.toMillis)
- case "read" => StdIn.readLine("Press RETURN to stop...")
- case _ => throw new Exception("pekko.ws-mode MUST be sleep or
read.")
- }
- } finally {
- system.terminate()
+ def echoWebSocketService: Flow[Message, Message, NotUsed] =
+ Flow[Message] // just let message flow directly to the output
}
-
- def echoWebSocketService: Flow[Message, Message, NotUsed] =
- Flow[Message] // just let message flow directly to the output
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/scaladsl/TestClient.scala
b/http-core/src/test/scala/org/apache/pekko/http/scaladsl/TestClient.scala
index 56f6fa6e1..dc97b49ca 100644
--- a/http-core/src/test/scala/org/apache/pekko/http/scaladsl/TestClient.scala
+++ b/http-core/src/test/scala/org/apache/pekko/http/scaladsl/TestClient.scala
@@ -30,94 +30,96 @@ import pekko.util.ByteString
import scala.concurrent.{ Await, Future }
import scala.concurrent.duration._
-object TestClient extends App {
- val testConf: Config = ConfigFactory.parseString("""
- pekko.loglevel = DEBUG
- pekko.log-dead-letters = off
- pekko.io.tcp.trace-logging = off""")
- implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
- import system.dispatcher
-
- installEventStreamLoggerFor[UnhandledMessage]
-
- val host = "github.com"
-
- fetchServerVersion1()
-
- // Console.readLine()
- // system.terminate()
-
- def fetchServerVersion1(): Unit = {
- println(s"Fetching HTTPS server version of host `$host` via a direct
low-level connection ...")
-
- val connection = Http().outgoingConnectionHttps(host)
- val result =
Source.single(HttpRequest()).via(connection).runWith(Sink.head)
- result.map(_.header[headers.Server]).onComplete {
- case Success(res) =>
- println(s"$host is running ${res.mkString(", ")}")
- println()
- fetchServerVersion2()
-
- case Failure(error) =>
- println(s"Error: $error")
- println()
- fetchServerVersion2()
- }
- }
+object TestClient {
+ def main(args: Array[String]): Unit = {
+ val testConf: Config = ConfigFactory.parseString("""
+ pekko.loglevel = DEBUG
+ pekko.log-dead-letters = off
+ pekko.io.tcp.trace-logging = off""")
+ implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
+ import system.dispatcher
- def fetchServerVersion2(): Unit = {
- println(s"Fetching HTTP server version of host `$host` via the high-level
API ...")
- val result = Http().singleRequest(HttpRequest(uri = s"https://$host/";))
- result.map(_.header[headers.Server]).onComplete {
- case Success(res) =>
- println(s"$host is running ${res.mkString(", ")}")
- Http().shutdownAllConnectionPools().onComplete { _ =>
system.log.info("STOPPED"); shutdown() }
-
- case Failure(error) =>
- println(s"Error: $error")
- shutdown()
- }
- }
+ installEventStreamLoggerFor[UnhandledMessage]
- // for gathering dumps of entity and headers from pekko http client
- // and curl in parallel to compare
- def fetchAndStoreABunchOfUrlsWithHttpAndCurl(urls: Seq[String]): Unit = {
- assert(urls.nonEmpty)
- assert(new File("/tmp/client-dumps/").exists(), "you need to create
/tmp/client-dumps/ before running")
+ val host = "github.com"
- val testConf: Config = ConfigFactory.parseString("""
- pekko.loglevel = DEBUG
- pekko.log-dead-letters = off
- pekko.io.tcp.trace-logging = off""")
- implicit val system = ActorSystem("ServerTest", testConf)
-
- try {
- val done = Future.traverse(urls.zipWithIndex) {
- case (url, index) =>
- Http().singleRequest(HttpRequest(uri = url)).map { response =>
- val path = new
File(s"/tmp/client-dumps/pekko-body-$index.dump").toPath
- val headersPath = new
File(s"/tmp/client-dumps/pekko-headers-$index.dump").toPath
-
- import scala.sys.process._
- (s"""curl -D /tmp/client-dumps/curl-headers-$index.dump $url""" #>
new File(
- s"/tmp/client-dumps/curl-body-$index.dump")).!
-
- val headers = Source(response.headers).map(header =>
ByteString(header.name + ": " + header.value + "\n"))
- .runWith(FileIO.toPath(headersPath))
-
- val body = response.entity.dataBytes
- .runWith(FileIO.toPath(path))
- .map(res => (url, path, res)): Future[(String, Path, IOResult)]
-
- headers.flatMap(_ => body)
- }
+ fetchServerVersion1()
+
+ // Console.readLine()
+ // system.terminate()
+
+ def fetchServerVersion1(): Unit = {
+ println(s"Fetching HTTPS server version of host `$host` via a direct
low-level connection ...")
+
+ val connection = Http().outgoingConnectionHttps(host)
+ val result =
Source.single(HttpRequest()).via(connection).runWith(Sink.head)
+ result.map(_.header[headers.Server]).onComplete {
+ case Success(res) =>
+ println(s"$host is running ${res.mkString(", ")}")
+ println()
+ fetchServerVersion2()
+
+ case Failure(error) =>
+ println(s"Error: $error")
+ println()
+ fetchServerVersion2()
}
+ }
- println("Fetched urls: " + Await.result(done, 10.minutes))
- } finally {
- Http().shutdownAllConnectionPools().flatMap(_ => system.terminate())
+ def fetchServerVersion2(): Unit = {
+ println(s"Fetching HTTP server version of host `$host` via the
high-level API ...")
+ val result = Http().singleRequest(HttpRequest(uri = s"https://$host/";))
+ result.map(_.header[headers.Server]).onComplete {
+ case Success(res) =>
+ println(s"$host is running ${res.mkString(", ")}")
+ Http().shutdownAllConnectionPools().onComplete { _ =>
system.log.info("STOPPED"); shutdown() }
+
+ case Failure(error) =>
+ println(s"Error: $error")
+ shutdown()
+ }
}
- }
- def shutdown(): Unit = system.terminate()
+ // for gathering dumps of entity and headers from pekko http client
+ // and curl in parallel to compare
+ def fetchAndStoreABunchOfUrlsWithHttpAndCurl(urls: Seq[String]): Unit = {
+ assert(urls.nonEmpty)
+ assert(new File("/tmp/client-dumps/").exists(), "you need to create
/tmp/client-dumps/ before running")
+
+ val testConf: Config = ConfigFactory.parseString("""
+ pekko.loglevel = DEBUG
+ pekko.log-dead-letters = off
+ pekko.io.tcp.trace-logging = off""")
+ implicit val system = ActorSystem("ServerTest", testConf)
+
+ try {
+ val done = Future.traverse(urls.zipWithIndex) {
+ case (url, index) =>
+ Http().singleRequest(HttpRequest(uri = url)).map { response =>
+ val path = new
File(s"/tmp/client-dumps/pekko-body-$index.dump").toPath
+ val headersPath = new
File(s"/tmp/client-dumps/pekko-headers-$index.dump").toPath
+
+ import scala.sys.process._
+ (s"""curl -D /tmp/client-dumps/curl-headers-$index.dump $url"""
#> new File(
+ s"/tmp/client-dumps/curl-body-$index.dump")).!
+
+ val headers = Source(response.headers).map(header =>
ByteString(header.name + ": " + header.value + "\n"))
+ .runWith(FileIO.toPath(headersPath))
+
+ val body = response.entity.dataBytes
+ .runWith(FileIO.toPath(path))
+ .map(res => (url, path, res)): Future[(String, Path, IOResult)]
+
+ headers.flatMap(_ => body)
+ }
+ }
+
+ println("Fetched urls: " + Await.result(done, 10.minutes))
+ } finally {
+ Http().shutdownAllConnectionPools().flatMap(_ => system.terminate())
+ }
+ }
+
+ def shutdown(): Unit = system.terminate()
+ }
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/scaladsl/TestServer.scala
b/http-core/src/test/scala/org/apache/pekko/http/scaladsl/TestServer.scala
index c7c5e8d0c..e1f65ced9 100644
--- a/http-core/src/test/scala/org/apache/pekko/http/scaladsl/TestServer.scala
+++ b/http-core/src/test/scala/org/apache/pekko/http/scaladsl/TestServer.scala
@@ -30,67 +30,69 @@ import com.typesafe.config.{ Config, ConfigFactory }
import scala.io.StdIn
-object TestServer extends App {
- val testConf: Config = ConfigFactory.parseString("""
- pekko.loglevel = INFO
- pekko.log-dead-letters = off
- pekko.stream.materializer.debug.fuzzing-mode = off
- pekko.actor.serialize-creators = off
- pekko.actor.serialize-messages = off
- pekko.actor.default-dispatcher.throughput = 1000
- """)
- implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
+object TestServer {
+ def main(args: Array[String]): Unit = {
+ val testConf: Config = ConfigFactory.parseString("""
+ pekko.loglevel = INFO
+ pekko.log-dead-letters = off
+ pekko.stream.materializer.debug.fuzzing-mode = off
+ pekko.actor.serialize-creators = off
+ pekko.actor.serialize-messages = off
+ pekko.actor.default-dispatcher.throughput = 1000
+ """)
+ implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
- implicit val fm: Materializer = Materializer(system)
- try {
- val binding = Http().newServerAt("localhost", 9001).bindSync {
- case req @ HttpRequest(GET, Uri.Path("/"), _, _, _) =>
- req.attribute(webSocketUpgrade) match {
- case Some(upgrade) => upgrade.handleMessages(echoWebSocketService)
// needed for running the autobahn test suite
- case None => index
- }
- case HttpRequest(GET, Uri.Path("/ping"), _, _, _) =>
HttpResponse(entity = "PONG!")
- case HttpRequest(GET, Uri.Path("/crash"), _, _, _) =>
sys.error("BOOM!")
- case req @ HttpRequest(GET, Uri.Path("/ws-greeter"), _, _, _) =>
- req.attribute(webSocketUpgrade) match {
- case Some(upgrade) => upgrade.handleMessages(greeterWebSocketService)
- case None => HttpResponse(400, entity = "Not a valid
websocket request!")
- }
- case _: HttpRequest => HttpResponse(404, entity = "Unknown resource!")
- }
+ implicit val fm: Materializer = Materializer(system)
+ try {
+ val binding = Http().newServerAt("localhost", 9001).bindSync {
+ case req @ HttpRequest(GET, Uri.Path("/"), _, _, _) =>
+ req.attribute(webSocketUpgrade) match {
+ case Some(upgrade) => upgrade.handleMessages(echoWebSocketService)
// needed for running the autobahn test suite
+ case None => index
+ }
+ case HttpRequest(GET, Uri.Path("/ping"), _, _, _) =>
HttpResponse(entity = "PONG!")
+ case HttpRequest(GET, Uri.Path("/crash"), _, _, _) =>
sys.error("BOOM!")
+ case req @ HttpRequest(GET, Uri.Path("/ws-greeter"), _, _, _) =>
+ req.attribute(webSocketUpgrade) match {
+ case Some(upgrade) =>
upgrade.handleMessages(greeterWebSocketService)
+ case None => HttpResponse(400, entity = "Not a valid
websocket request!")
+ }
+ case _: HttpRequest => HttpResponse(404, entity = "Unknown resource!")
+ }
- Await.result(binding, 1.second) // throws if binding fails
- println("Server online at http://localhost:9001";)
- println("Press RETURN to stop...")
- StdIn.readLine()
- } finally {
- system.terminate()
- }
+ Await.result(binding, 1.second) // throws if binding fails
+ println("Server online at http://localhost:9001";)
+ println("Press RETURN to stop...")
+ StdIn.readLine()
+ } finally {
+ system.terminate()
+ }
- ////////////// helpers //////////////
+ ////////////// helpers //////////////
- lazy val index = HttpResponse(
- entity = HttpEntity(
- ContentTypes.`text/html(UTF-8)`,
- """|<html>
- | <body>
- | <h1>Say hello to <i>pekko-http-core</i>!</h1>
- | <p>Defined resources:</p>
- | <ul>
- | <li><a href="/ping">/ping</a></li>
- | <li><a href="/crash">/crash</a></li>
- | </ul>
- | </body>
- |</html>""".stripMargin))
+ lazy val index = HttpResponse(
+ entity = HttpEntity(
+ ContentTypes.`text/html(UTF-8)`,
+ """|<html>
+ | <body>
+ | <h1>Say hello to <i>pekko-http-core</i>!</h1>
+ | <p>Defined resources:</p>
+ | <ul>
+ | <li><a href="/ping">/ping</a></li>
+ | <li><a href="/crash">/crash</a></li>
+ | </ul>
+ | </body>
+ |</html>""".stripMargin))
- def echoWebSocketService: Flow[Message, Message, NotUsed] =
- Flow[Message] // just let message flow directly to the output
+ def echoWebSocketService: Flow[Message, Message, NotUsed] =
+ Flow[Message] // just let message flow directly to the output
- def greeterWebSocketService: Flow[Message, Message, NotUsed] =
- Flow[Message]
- .collect {
- case TextMessage.Strict(name) => TextMessage(s"Hello '$name'")
- case tm: TextMessage => TextMessage(Source.single("Hello ")
++ tm.textStream)
- // ignore binary messages
- }
+ def greeterWebSocketService: Flow[Message, Message, NotUsed] =
+ Flow[Message]
+ .collect {
+ case TextMessage.Strict(name) => TextMessage(s"Hello '$name'")
+ case tm: TextMessage => TextMessage(Source.single("Hello ")
++ tm.textStream)
+ // ignore binary messages
+ }
+ }
}
diff --git
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/TestSingleRequest.scala
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/TestSingleRequest.scala
index cfd07a238..484b81fc9 100644
---
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/TestSingleRequest.scala
+++
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/TestSingleRequest.scala
@@ -24,26 +24,28 @@ import pekko.util.ByteString
import com.typesafe.config.{ Config, ConfigFactory }
-object TestSingleRequest extends App {
- val testConf: Config = ConfigFactory.parseString("""
- pekko.loglevel = INFO
- pekko.log-dead-letters = off
- pekko.stream.materializer.debug.fuzzing-mode = off
- """)
- implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
- import system.dispatcher
+object TestSingleRequest {
+ def main(args: Array[String]): Unit = {
+ val testConf: Config = ConfigFactory.parseString("""
+ pekko.loglevel = INFO
+ pekko.log-dead-letters = off
+ pekko.stream.materializer.debug.fuzzing-mode = off
+ """)
+ implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
+ import system.dispatcher
- val url = StdIn.readLine("url? ")
+ val url = StdIn.readLine("url? ")
- val x = Http().singleRequest(HttpRequest(uri = url))
+ val x = Http().singleRequest(HttpRequest(uri = url))
- val res = Await.result(x, 10.seconds)
+ val res = Await.result(x, 10.seconds)
- val response = res.entity.dataBytes.runFold(ByteString.empty)(_ ++
_).map(_.utf8String)
+ val response = res.entity.dataBytes.runFold(ByteString.empty)(_ ++
_).map(_.utf8String)
- println(" ------------ RESPONSE ------------")
- println(Await.result(response, 10.seconds))
- println(" -------- END OF RESPONSE ---------")
+ println(" ------------ RESPONSE ------------")
+ println(Await.result(response, 10.seconds))
+ println(" -------- END OF RESPONSE ---------")
- system.terminate()
+ system.terminate()
+ }
}
diff --git
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/TcpLeakApp.scala
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/TcpLeakApp.scala
index 5e0de71de..1fb945127 100644
---
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/TcpLeakApp.scala
+++
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/TcpLeakApp.scala
@@ -26,38 +26,43 @@ import pekko.util.ByteString
import com.typesafe.config.{ Config, ConfigFactory }
-object TcpLeakApp extends App {
- val testConf: Config = ConfigFactory.parseString(
- """
- pekko.loglevel = DEBUG
- pekko.log-dead-letters = on
- pekko.io.tcp.trace-logging = on""")
- implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
-
- import system.dispatcher
-
- val tcpFlow = Tcp(system).outgoingConnection(new
InetSocketAddress("127.0.0.1", 1234)).named("TCP-outgoingConnection")
- List
- .fill(100)(
- Source
- .single(ByteString("FOO"))
-
.log("outerFlow-beforeTcpFlow").withAttributes(ActorAttributes.logLevels(Logging.DebugLevel,
Logging.ErrorLevel,
- Logging.ErrorLevel))
- .via(tcpFlow)
-
.log("outerFlow-afterTcpFlow").withAttributes(ActorAttributes.logLevels(Logging.DebugLevel,
Logging.ErrorLevel,
- Logging.ErrorLevel))
- .toMat(Sink.head)(Keep.right).run())
- .last
- .onComplete {
- result =>
- println(s"Result: $result")
- Thread.sleep(10000)
- println("===================== \n\n" +
- system.asInstanceOf[
- ActorSystemImpl].printTree + "\n\n========================")
- }
-
- Thread.sleep(11000)
- StdIn.readLine("Press Enter to stop the application")
- system.terminate()
+object TcpLeakApp {
+ def main(args: Array[String]): Unit = {
+ val testConf: Config = ConfigFactory.parseString(
+ """
+ pekko.loglevel = DEBUG
+ pekko.log-dead-letters = on
+ pekko.io.tcp.trace-logging = on""")
+ implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
+
+ import system.dispatcher
+
+ val tcpFlow =
+ Tcp(system).outgoingConnection(new InetSocketAddress("127.0.0.1",
1234)).named("TCP-outgoingConnection")
+ List
+ .fill(100)(
+ Source
+ .single(ByteString("FOO"))
+
.log("outerFlow-beforeTcpFlow").withAttributes(ActorAttributes.logLevels(Logging.DebugLevel,
+ Logging.ErrorLevel,
+ Logging.ErrorLevel))
+ .via(tcpFlow)
+
.log("outerFlow-afterTcpFlow").withAttributes(ActorAttributes.logLevels(Logging.DebugLevel,
+ Logging.ErrorLevel,
+ Logging.ErrorLevel))
+ .toMat(Sink.head)(Keep.right).run())
+ .last
+ .onComplete {
+ result =>
+ println(s"Result: $result")
+ Thread.sleep(10000)
+ println("===================== \n\n" +
+ system.asInstanceOf[
+ ActorSystemImpl].printTree + "\n\n========================")
+ }
+
+ Thread.sleep(11000)
+ StdIn.readLine("Press Enter to stop the application")
+ system.terminate()
+ }
}
diff --git
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/TestServer.scala
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/TestServer.scala
index ecb98af38..0c35be3b6 100644
---
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/TestServer.scala
+++
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/TestServer.scala
@@ -32,99 +32,101 @@ import pekko.stream.scaladsl._
import com.typesafe.config.{ Config, ConfigFactory }
-object TestServer extends App {
- val testConf: Config = ConfigFactory.parseString("""
- pekko.loglevel = INFO
- pekko.log-dead-letters = off
- pekko.stream.materializer.debug.fuzzing-mode = off
- """)
-
- implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
- implicit val ec: ExecutionContext = system.dispatcher
-
- import spray.json.DefaultJsonProtocol._
-
- import pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
- final case class Tweet(message: String)
- implicit val tweetFormat: RootJsonFormat[Tweet] = jsonFormat1(Tweet.apply)
-
- implicit val jsonStreaming: JsonEntityStreamingSupport =
EntityStreamingSupport.json()
-
- import Directives._
- import ScalaXmlSupport._
-
- def auth: AuthenticatorPF[String] = {
- case p @ Credentials.Provided(name) if p.verify(name + "-password") => name
- }
-
- // format: OFF
- val routes = {
- get {
- path("") {
- withRequestTimeout(1.milli, _ => HttpResponse(
- StatusCodes.EnhanceYourCalm,
- entity = "Unable to serve response within time limit, please enhance
your calm.")) {
- Thread.sleep(1000)
- complete(index)
- }
- } ~
- path("secure") {
- authenticateBasicPF("My very secure site", auth) { user =>
- complete(<html> <body> Hello <b>{user}</b>. Access has been granted!
</body> </html>)
- }
- } ~
- path("ping") {
- complete("PONG!")
- } ~
- path("crash") {
- complete(sys.error("BOOM!"))
- } ~
- path("tweet") {
- complete(Tweet("Hello, world!"))
- } ~
- (path("tweets") & parameter("n".as[Int])) { n =>
- get {
- val tweets = Source.repeat(Tweet("Hello, world!")).take(n)
- complete(tweets)
+object TestServer {
+ def main(args: Array[String]): Unit = {
+ val testConf: Config = ConfigFactory.parseString("""
+ pekko.loglevel = INFO
+ pekko.log-dead-letters = off
+ pekko.stream.materializer.debug.fuzzing-mode = off
+ """)
+
+ implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
+ implicit val ec: ExecutionContext = system.dispatcher
+
+ import spray.json.DefaultJsonProtocol._
+
+ import pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
+ final case class Tweet(message: String)
+ implicit val tweetFormat: RootJsonFormat[Tweet] = jsonFormat1(Tweet.apply)
+
+ implicit val jsonStreaming: JsonEntityStreamingSupport =
EntityStreamingSupport.json()
+
+ import Directives._
+ import ScalaXmlSupport._
+
+ def auth: AuthenticatorPF[String] = {
+ case p @ Credentials.Provided(name) if p.verify(name + "-password") =>
name
+ }
+
+ // format: OFF
+ lazy val index =
+ <html>
+ <body>
+ <h1>Say hello to <i>pekko-http-core</i>!</h1>
+ <p>Defined resources:</p>
+ <ul>
+ <li><a href="/ping">/ping</a></li>
+ <li><a href="/secure">/secure</a> Use any username and
'<username>-password' as credentials</li>
+ <li><a href="/crash">/crash</a></li>
+ </ul>
+ </body>
+ </html>
+
+ val routes = {
+ get {
+ path("") {
+ withRequestTimeout(1.milli, _ => HttpResponse(
+ StatusCodes.EnhanceYourCalm,
+ entity = "Unable to serve response within time limit, please
enhance your calm.")) {
+ Thread.sleep(1000)
+ complete(index)
+ }
} ~
- post {
- entity(asSourceOf[Tweet]) { tweets =>
- onComplete(tweets.runFold(0)({ case (acc, t) => acc + 1 })) {
count =>
- complete(s"Total tweets received: " + count)
- }
+ path("secure") {
+ authenticateBasicPF("My very secure site", auth) { user =>
+ complete(<html> <body> Hello <b>{user}</b>. Access has been
granted! </body> </html>)
}
} ~
- put {
- // checking the alternative syntax also works:
- entity(as[Source[Tweet, NotUsed]]) { tweets =>
- onComplete(tweets.runFold(0)({ case (acc, t) => acc + 1 })) {
count =>
- complete(s"Total tweets received: " + count)
+ path("ping") {
+ complete("PONG!")
+ } ~
+ path("crash") {
+ complete(sys.error("BOOM!"))
+ } ~
+ path("tweet") {
+ complete(Tweet("Hello, world!"))
+ } ~
+ (path("tweets") & parameter("n".as[Int])) { n =>
+ get {
+ val tweets = Source.repeat(Tweet("Hello, world!")).take(n)
+ complete(tweets)
+ } ~
+ post {
+ entity(asSourceOf[Tweet]) { tweets =>
+ onComplete(tweets.runFold(0)({ case (acc, t) => acc + 1 })) {
count =>
+ complete(s"Total tweets received: " + count)
+ }
+ }
+ } ~
+ put {
+ // checking the alternative syntax also works:
+ entity(as[Source[Tweet, NotUsed]]) { tweets =>
+ onComplete(tweets.runFold(0)({ case (acc, t) => acc + 1 })) {
count =>
+ complete(s"Total tweets received: " + count)
+ }
}
}
}
- }
- } ~
- pathPrefix("inner")(getFromResourceDirectory("someDir"))
+ } ~
+ pathPrefix("inner")(getFromResourceDirectory("someDir"))
+ }
+ // format: ON
+
+ val bindingFuture = Http().newServerAt(interface = "0.0.0.0", port =
8080).bind(routes)
+
+ println(s"Server online at http://0.0.0.0:8080/\nPress RETURN to stop...")
+ StdIn.readLine()
+
+ bindingFuture.flatMap(_.unbind()).onComplete(_ => system.terminate())
}
- // format: ON
-
- val bindingFuture = Http().newServerAt(interface = "0.0.0.0", port =
8080).bind(routes)
-
- println(s"Server online at http://0.0.0.0:8080/\nPress RETURN to stop...")
- StdIn.readLine()
-
- bindingFuture.flatMap(_.unbind()).onComplete(_ => system.terminate())
-
- lazy val index =
- <html>
- <body>
- <h1>Say hello to <i>pekko-http-core</i>!</h1>
- <p>Defined resources:</p>
- <ul>
- <li><a href="/ping">/ping</a></li>
- <li><a href="/secure">/secure</a> Use any username and
'<username>-password' as credentials</li>
- <li><a href="/crash">/crash</a></li>
- </ul>
- </body>
- </html>
}
diff --git
a/http2-tests/src/test/scala/org/apache/pekko/http/scaladsl/Http2ServerTest.scala
b/http2-tests/src/test/scala/org/apache/pekko/http/scaladsl/Http2ServerTest.scala
index e56da5329..bff34438e 100644
---
a/http2-tests/src/test/scala/org/apache/pekko/http/scaladsl/Http2ServerTest.scala
+++
b/http2-tests/src/test/scala/org/apache/pekko/http/scaladsl/Http2ServerTest.scala
@@ -36,108 +36,108 @@ import com.typesafe.config.ConfigFactory
/**
* App to manually test an HTTP2 server
*/
-object Http2ServerTest extends App {
- val testConf: Config = ConfigFactory.parseString("""
- pekko.log-dead-letters = off
- pekko.stream.materializer.debug.fuzzing-mode = off
- pekko.actor.serialize-creators = off
- pekko.actor.serialize-messages = off
- #pekko.actor.default-dispatcher.throughput = 1000
- pekko.actor.default-dispatcher.fork-join-executor.parallelism-max=8""")
- implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
- implicit val ec: ExecutionContext = system.dispatcher
-
- def slowDown[T](millis: Int): T => Future[T] = { t =>
- pekko.pattern.after(millis.millis, system.scheduler)(Future.successful(t))
- }
-
- val syncHandler: HttpRequest => HttpResponse = {
- case HttpRequest(GET, Uri.Path("/"), _, _, _)
=> index
- case HttpRequest(GET, Uri.Path("/ping"), _, _, _)
=> HttpResponse(entity = "PONG!")
- case HttpRequest(GET, Uri.Path("/image-page"), _, _, _)
=> imagePage
- case HttpRequest(GET, Uri(_, _, p, _, _), _, _, _) if
p.toString.startsWith("/image1") =>
- HttpResponse(entity = HttpEntity(MediaTypes.`image/jpeg`,
- FileIO.fromPath(Paths.get("bigimage.jpg"),
100000).mapAsync(1)(slowDown(1))))
- case HttpRequest(GET, Uri(_, _, p, _, _), _, _, _) if
p.toString.startsWith("/image2") =>
- HttpResponse(entity = HttpEntity(MediaTypes.`image/jpeg`,
- FileIO.fromPath(Paths.get("bigimage2.jpg"),
150000).mapAsync(1)(slowDown(2))))
- case HttpRequest(GET, Uri.Path("/crash"), _, _, _) => sys.error("BOOM!")
- case _: HttpRequest => HttpResponse(404,
entity = "Unknown resource!")
- }
-
- val asyncHandler: HttpRequest => Future[HttpResponse] = {
- case HttpRequest(POST, Uri.Path("/upload"), _, entity, _) =>
- Unmarshal(entity).to[Multipart.FormData]
- .flatMap { formData =>
- formData.parts.runFoldAsync("") { (msg, part) =>
- part.entity.dataBytes.runFold(0)(_ + _.size)
- .map(dataSize =>
- msg +
- s"${part.name} ${part.filename} $dataSize
${part.entity.contentType} ${part.additionalDispositionParams}\n")
- }
- }
- .map { msg =>
- HttpResponse(entity = s"Got upload: $msg")
- }
- case req => Future.successful(syncHandler(req))
- }
-
- try {
- val bindings =
- for {
- httpsBinding <- Http().newServerAt(interface = "localhost", port =
9000).enableHttps(
- ExampleHttpContexts.exampleServerContext).bind(asyncHandler)
- plainBinding <- Http().newServerAt(interface = "localhost", port =
9002).bind(asyncHandler)
- } yield (httpsBinding, plainBinding)
-
- Await.result(bindings, 1.second) // throws if binding fails
- println(Console.BOLD + "Server (HTTP/1.1 and HTTP/2) online at
https://localhost:9000"; + Console.RESET)
- println(Console.BOLD + "Server (HTTP/1/1 and HTTP/2) online at
http://localhost:9002"; + Console.RESET)
- println("Press RETURN to stop...")
- StdIn.readLine()
- } finally {
- system.terminate()
- }
-
- ////////////// helpers //////////////
-
- lazy val index = HttpResponse(
- entity = HttpEntity(
- ContentTypes.`text/html(UTF-8)`,
- """|<html>
- | <body>
- | <h1>Say hello to <i>pekko-http-core</i>!</h1>
- | <p>Defined resources:</p>
- | <ul>
- | <li><a href="/ping">/ping</a></li>
- | <li><a href="/image-page">/image-page</a></li>
- | <li><a href="/crash">/crash</a></li>
- | </ul>
- | <div>
- | <form method="post" enctype="multipart/form-data"
action="upload">
- | <label for="file">File</label><input type="file" name="file"
multiple="true"/><br/>
- | <input type="submit" />
- | </form>
- | </div>
- | </body>
- |</html>""".stripMargin))
-
- def imagesBlock = {
- def one(): String =
- s"""<img width="80" height="60"
src="/image1?cachebuster=${Random.nextInt()}"></img>
- |<img width="80" height="60"
src="/image2?cachebuster=${Random.nextInt()}"></img>
- |""".stripMargin
-
- Seq.fill(20)(one()).mkString
- }
-
- lazy val imagePage = HttpResponse(
- entity = HttpEntity(
- ContentTypes.`text/html(UTF-8)`,
- s"""|<html>
+object Http2ServerTest {
+ def main(args: Array[String]): Unit = {
+ val testConf: Config = ConfigFactory.parseString("""
+ pekko.log-dead-letters = off
+ pekko.stream.materializer.debug.fuzzing-mode = off
+ pekko.actor.serialize-creators = off
+ pekko.actor.serialize-messages = off
+ #pekko.actor.default-dispatcher.throughput = 1000
+ pekko.actor.default-dispatcher.fork-join-executor.parallelism-max=8""")
+ implicit val system: ActorSystem = ActorSystem("ServerTest", testConf)
+ implicit val ec: ExecutionContext = system.dispatcher
+
+ def slowDown[T](millis: Int): T => Future[T] = { t =>
+ pekko.pattern.after(millis.millis,
system.scheduler)(Future.successful(t))
+ }
+
+ lazy val index = HttpResponse(
+ entity = HttpEntity(
+ ContentTypes.`text/html(UTF-8)`,
+ """|<html>
| <body>
- | <h1>Image Page</h1>
- | $imagesBlock
+ | <h1>Say hello to <i>pekko-http-core</i>!</h1>
+ | <p>Defined resources:</p>
+ | <ul>
+ | <li><a href="/ping">/ping</a></li>
+ | <li><a href="/image-page">/image-page</a></li>
+ | <li><a href="/crash">/crash</a></li>
+ | </ul>
+ | <div>
+ | <form method="post" enctype="multipart/form-data"
action="upload">
+ | <label for="file">File</label><input type="file"
name="file" multiple="true"/><br/>
+ | <input type="submit" />
+ | </form>
+ | </div>
| </body>
|</html>""".stripMargin))
+
+ def imagesBlock = {
+ def one(): String =
+ s"""<img width="80" height="60"
src="/image1?cachebuster=${Random.nextInt()}"></img>
+ |<img width="80" height="60"
src="/image2?cachebuster=${Random.nextInt()}"></img>
+ |""".stripMargin
+
+ Seq.fill(20)(one()).mkString
+ }
+
+ lazy val imagePage = HttpResponse(
+ entity = HttpEntity(
+ ContentTypes.`text/html(UTF-8)`,
+ s"""|<html>
+ | <body>
+ | <h1>Image Page</h1>
+ | $imagesBlock
+ | </body>
+ |</html>""".stripMargin))
+
+ val syncHandler: HttpRequest => HttpResponse = {
+ case HttpRequest(GET, Uri.Path("/"), _, _, _)
=> index
+ case HttpRequest(GET, Uri.Path("/ping"), _, _, _)
=> HttpResponse(entity = "PONG!")
+ case HttpRequest(GET, Uri.Path("/image-page"), _, _, _)
=> imagePage
+ case HttpRequest(GET, Uri(_, _, p, _, _), _, _, _) if
p.toString.startsWith("/image1") =>
+ HttpResponse(entity = HttpEntity(MediaTypes.`image/jpeg`,
+ FileIO.fromPath(Paths.get("bigimage.jpg"),
100000).mapAsync(1)(slowDown(1))))
+ case HttpRequest(GET, Uri(_, _, p, _, _), _, _, _) if
p.toString.startsWith("/image2") =>
+ HttpResponse(entity = HttpEntity(MediaTypes.`image/jpeg`,
+ FileIO.fromPath(Paths.get("bigimage2.jpg"),
150000).mapAsync(1)(slowDown(2))))
+ case HttpRequest(GET, Uri.Path("/crash"), _, _, _) => sys.error("BOOM!")
+ case _: HttpRequest => HttpResponse(404,
entity = "Unknown resource!")
+ }
+
+ val asyncHandler: HttpRequest => Future[HttpResponse] = {
+ case HttpRequest(POST, Uri.Path("/upload"), _, entity, _) =>
+ Unmarshal(entity).to[Multipart.FormData]
+ .flatMap { formData =>
+ formData.parts.runFoldAsync("") { (msg, part) =>
+ part.entity.dataBytes.runFold(0)(_ + _.size)
+ .map(dataSize =>
+ msg +
+ s"${part.name} ${part.filename} $dataSize
${part.entity.contentType} ${part.additionalDispositionParams}\n")
+ }
+ }
+ .map { msg =>
+ HttpResponse(entity = s"Got upload: $msg")
+ }
+ case req => Future.successful(syncHandler(req))
+ }
+
+ try {
+ val bindings =
+ for {
+ httpsBinding <- Http().newServerAt(interface = "localhost", port =
9000).enableHttps(
+ ExampleHttpContexts.exampleServerContext).bind(asyncHandler)
+ plainBinding <- Http().newServerAt(interface = "localhost", port =
9002).bind(asyncHandler)
+ } yield (httpsBinding, plainBinding)
+
+ Await.result(bindings, 1.second) // throws if binding fails
+ println(Console.BOLD + "Server (HTTP/1.1 and HTTP/2) online at
https://localhost:9000"; + Console.RESET)
+ println(Console.BOLD + "Server (HTTP/1/1 and HTTP/2) online at
http://localhost:9002"; + Console.RESET)
+ println("Press RETURN to stop...")
+ StdIn.readLine()
+ } finally {
+ system.terminate()
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
