Added remote address in ConnectionInfo for pekko (#4847)

Relatively simple implementation of ConnectionInfo.remote and
ConnectionInfo.secure for pekko-http.

Maybe better than nothing. I am open to suggestions, but do not see how
I can do it better with pekko now.

One small step would be to use HostDirectives.extractHost in
PekkoHttpServerInterpreter - for a better a local connection info.

Author: jarekratajski

server/pekko-http-server/src/main/scala/sttp/tapir/server/pekkohttp/PekkoServerRequest.scala

@@ -1,18 +1,35 @@
 package sttp.tapir.server.pekkohttp
 
-import org.apache.pekko.http.scaladsl.model.{Uri => PekkoUri}
+import org.apache.pekko.http.scaladsl.model.{AttributeKeys, Uri => PekkoUri}
 import org.apache.pekko.http.scaladsl.server.RequestContext
 import sttp.model.Uri.{Authority, FragmentSegment, HostSegment, PathSegments, QuerySegment}
 import sttp.model.{Header, HeaderNames, Method, QueryParams, Uri}
 import sttp.tapir.model.{ConnectionInfo, ServerRequest}
 import sttp.tapir.{AttributeKey, AttributeMap}
 
+import java.net.InetSocketAddress
 import scala.annotation.tailrec
 import scala.collection.immutable.Seq
 
 private[pekkohttp] case class PekkoServerRequest(ctx: RequestContext, attributes: AttributeMap = AttributeMap.Empty) extends ServerRequest {
   override def protocol: String = ctx.request.protocol.value
-  override lazy val connectionInfo: ConnectionInfo = ConnectionInfo(None, None, None)
+  private lazy val remote = ctx
+    .request
+    .attribute(AttributeKeys.remoteAddress)
+    .flatMap(_.toIP)
+
+  override def connectionInfo: ConnectionInfo = {
+    // simple / naive deduction of secure property, as pekko uses only this 4 schemes
+    // @see org.apache.pekko.http.scaladsl.model.HttpRequest#verifyUri
+    val secure = ctx.request.uri.scheme match {
+      case "https" | "wss" => Some(true)
+      case "http" | "ws" => Some(false)
+      case _ => None
+    }
+    ConnectionInfo(None, remote.map( addr =>
+      new InetSocketAddress(addr.ip, addr.port.getOrElse(0))
+    ) , secure)
+  }
   override def underlying: Any = ctx
 
   override lazy val pathSegments: List[String] = {

server/pekko-http-server/src/test/scala/sttp/tapir/server/pekkohttp/PekkoHttpServerTest.scala

@@ -163,7 +163,21 @@ class PekkoHttpServerTest extends TestSuite with EitherValues {
               basicRequest.get(uri"http://localhost:$port/api/custom_code").send(backend).map(_.code shouldBe StatusCode(800))
             }
             .unsafeToFuture()
-        }
+        },
+        Test("Server reads remote address and secure from connectionInfo") {
+          val e = endpoint.get.in("test").in(extractFromRequest(req => req.connectionInfo) ).out(stringBody)
+            .serverLogic{connectionInfo =>
+              val remote = connectionInfo.remote
+              val secure = connectionInfo.secure
+              s"$remote $secure".asRight[Unit].unit}
+          val route = Directives.pathPrefix("api")(PekkoHttpServerInterpreter().toRoute(e))
+          interpreter
+            .server(NonEmptyList.of(route))
+            .use { port =>
+              basicRequest.get(uri"http://localhost:$port/api/test").send(backend).map(_.body.value should fullyMatch regex """Some\(/127\.0\.0\.1:\d+\) Some\(false\)""")
+            }
+            .unsafeToFuture()
+        },
       )
       def drainPekko(stream: PekkoStreams.BinaryStream): Future[Unit] =
         stream.runWith(Sink.ignore).map(_ => ())

server/pekko-http-server/src/test/scala/sttp/tapir/server/pekkohttp/PekkoHttpTestServerInterpreter.scala

@@ -27,7 +27,9 @@ class PekkoHttpTestServerInterpreter(implicit actorSystem: ActorSystem)
       routes: NonEmptyList[Route],
       gracefulShutdownTimeout: Option[FiniteDuration]
   ): Resource[IO, Port] = {
-    val bind = IO.fromFuture(IO(Http().newServerAt("localhost", 0).bind(concat(routes.toList: _*))))
+    val bind = IO.fromFuture(IO(Http().newServerAt("localhost", 0)
+      .adaptSettings( setts => setts.withRemoteAddressAttribute(true))
+      .bind(concat(routes.toList: _*))))
 
     Resource
       .make(bind)(server => IO.fromFuture(IO(server.terminate(gracefulShutdownTimeout.getOrElse(50.millis)))).void)