Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

668 aspect oriented middleware #669

Open
wants to merge 9 commits into
base: main
Choose a base branch
from
Open

668 aspect oriented middleware #669

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
0b639fc
initial work an aspects
ValdemarGr Sep 7, 2023
2aa969e
more work
ValdemarGr Sep 7, 2023
e00ab12
dsl
ValdemarGr Sep 7, 2023
8b954b2
generalize effect type
ValdemarGr Sep 13, 2023
dc5e409
work on api consistency
ValdemarGr Sep 13, 2023
7f574db
removed req/res typeclasses
ValdemarGr Sep 15, 2023
113bab3
test resource
ValdemarGr Sep 17, 2023
584c51a
very big test
ValdemarGr Sep 17, 2023
8f9b947
tracing test done
ValdemarGr Sep 17, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .scalafmt.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,4 +10,4 @@ fileOverride {
"glob:**/scala-3/**" {
runner.dialect = scala3
}
}
}
5 changes: 4 additions & 1 deletion build.sbt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,10 @@ lazy val e2e = (projectMatrix in file("e2e"))
.settings(
codeGenClasspath := (codeGenJVM212 / Compile / fullClasspath).value,
libraryDependencies := Nil,
libraryDependencies ++= List(scalaPbGrpcRuntime, scalaPbRuntime, scalaPbRuntime % "protobuf", ceMunit % Test),
libraryDependencies ++= List(grpcApi, scalaPbGrpcRuntime, scalaPbRuntime, scalaPbRuntime % "protobuf", ceMunit % Test),
libraryDependencies ++= Seq(
"io.grpc" % "grpc-services" % versions.grpc % Test,
),
Compile / PB.targets := Seq(
scalapb.gen() -> (Compile / sourceManaged).value / "scalapb",
genModule(codegenFullName + "$") -> (Compile / sourceManaged).value / "fs2-grpc"
Expand Down
83 changes: 59 additions & 24 deletions codegen/src/main/scala/fs2/grpc/codegen/Fs2GrpcServicePrinter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,48 +47,67 @@ class Fs2GrpcServicePrinter(service: ServiceDescriptor, serviceSuffix: String, d
})
}

private[this] def handleMethod(method: MethodDescriptor) = {
private[this] def methodName(method: MethodDescriptor) =
method.streamType match {
case StreamType.Unary => "unaryToUnaryCall"
case StreamType.ClientStreaming => "streamingToUnaryCall"
case StreamType.ServerStreaming => "unaryToStreamingCall"
case StreamType.Bidirectional => "streamingToStreamingCall"
case StreamType.Unary => "unaryToUnary"
case StreamType.ClientStreaming => "streamingToUnary"
case StreamType.ServerStreaming => "unaryToStreaming"
case StreamType.Bidirectional => "streamingToStreaming"
}
}

private[this] def handleMethod(method: MethodDescriptor) =
methodName(method) + "Call"

private[this] def visitMethod(method: MethodDescriptor) =
"visit" + methodName(method).capitalize

private[this] def createClientCall(method: MethodDescriptor) = {
val basicClientCall =
s"$Fs2ClientCall[F](channel, ${method.grpcDescriptor.fullName}, dispatcher, clientOptions)"
s"$Fs2ClientCall[G](channel, ${method.grpcDescriptor.fullName}, dispatcher, clientOptions)"
if (method.isServerStreaming)
s"$Stream.eval($basicClientCall)"
else
basicClientCall
}

private[this] def serviceMethodImplementation(method: MethodDescriptor): PrinterEndo = { p =>
val mkMetadata = if (method.isServerStreaming) s"$Stream.eval(mkMetadata(ctx))" else "mkMetadata(ctx)"
val inType = method.inputType.scalaType
val outType = method.outputType.scalaType
val descriptor = method.grpcDescriptor.fullName

p.add(serviceMethodSignature(method) + " = {")
.indent
.add(s"$mkMetadata.flatMap { m =>")
.indent
.add(s"${createClientCall(method)}.flatMap(_.${handleMethod(method)}(request, m))")
.outdent
.add("}")
.outdent
.add("}")
p
.add(serviceMethodSignature(method) + " =")
.indented {
_.addStringMargin(
s"""|clientAspect.${visitMethod(method)}[$inType, $outType](
| ${ClientCallContext}(ctx, $descriptor),
| request,
| (req, m) => ${createClientCall(method)}.flatMap(_.${handleMethod(method)}(req, m))
|)""".stripMargin
)
}
}

private[this] def serviceBindingImplementation(method: MethodDescriptor): PrinterEndo = { p =>
val inType = method.inputType.scalaType
val outType = method.outputType.scalaType
val descriptor = method.grpcDescriptor.fullName
val handler = s"$Fs2ServerCallHandler[F](dispatcher, serverOptions).${handleMethod(method)}[$inType, $outType]"
val handler = s"$Fs2ServerCallHandler[G](dispatcher, serverOptions).${handleMethod(method)}[$inType, $outType]"

val serviceCall = s"serviceImpl.${method.name}"
val eval = if (method.isServerStreaming) s"$Stream.eval(mkCtx(m))" else "mkCtx(m)"

p.add(s".addMethod($descriptor, $handler((r, m) => $eval.flatMap($serviceCall(r, _))))")
p.addStringMargin(
s"""|.addMethod(
| $descriptor,
| $handler{ (r, m) =>
| serviceAspect.${visitMethod(method)}[$inType, $outType](
| ${ServerCallContext}(m, $descriptor),
| r,
| (r, m) => $serviceCall(r, m)
| )
| }
|)"""
)
}

private[this] def serviceMethods: PrinterEndo = _.seq(service.methods.map(serviceMethodSignature))
Expand Down Expand Up @@ -116,17 +135,27 @@ class Fs2GrpcServicePrinter(service: ServiceDescriptor, serviceSuffix: String, d
.add("}")

private[this] def serviceClient: PrinterEndo = {
_.add(
s"def mkClient[F[_]: $Async, $Ctx](dispatcher: $Dispatcher[F], channel: $Channel, mkMetadata: $Ctx => F[$Metadata], clientOptions: $ClientOptions): $serviceNameFs2[F, $Ctx] = new $serviceNameFs2[F, $Ctx] {"
_.addStringMargin(
s"""|def mkClientFull[F[_], G[_]: $Async, $Ctx](
| dispatcher: $Dispatcher[G],
| channel: $Channel,
| clientAspect: ${ClientAspect}[F, G, $Ctx],
| clientOptions: $ClientOptions
|): $serviceNameFs2[F, $Ctx] = new $serviceNameFs2[F, $Ctx] {"""
).indent
.call(serviceMethodImplementations)
.outdent
.add("}")
}

private[this] def serviceBinding: PrinterEndo = {
_.add(
s"protected def serviceBinding[F[_]: $Async, $Ctx](dispatcher: $Dispatcher[F], serviceImpl: $serviceNameFs2[F, $Ctx], mkCtx: $Metadata => F[$Ctx], serverOptions: $ServerOptions): $ServerServiceDefinition = {"
_.addStringMargin(
s"""|protected def serviceBindingFull[F[_], G[_]: $Async, $Ctx](
| dispatcher: $Dispatcher[G],
| serviceImpl: $serviceNameFs2[F, $Ctx],
| serviceAspect: ${ServiceAspect}[F, G, $Ctx],
| serverOptions: $ServerOptions
|) = {"""
).indent
.add(s"$ServerServiceDefinition")
.call(serviceBindingImplementations)
Expand All @@ -152,6 +181,8 @@ object Fs2GrpcServicePrinter {
private val effPkg = "_root_.cats.effect"
private val fs2Pkg = "_root_.fs2"
private val fs2grpcPkg = "_root_.fs2.grpc"
private val fs2grpcServerPkg = "_root_.fs2.grpc.server"
private val fs2grpcClientPkg = "_root_.fs2.grpc.client"
private val grpcPkg = "_root_.io.grpc"

// /
Expand All @@ -173,6 +204,10 @@ object Fs2GrpcServicePrinter {
val Channel = s"$grpcPkg.Channel"
val Metadata = s"$grpcPkg.Metadata"

val ServiceAspect = s"${fs2grpcServerPkg}.ServiceAspect"
val ServerCallContext = s"${fs2grpcServerPkg}.ServerCallContext"
val ClientAspect = s"${fs2grpcClientPkg}.ClientAspect"
val ClientCallContext = s"${fs2grpcClientPkg}.ClientCallContext"
}

}
102 changes: 76 additions & 26 deletions e2e/src/test/resources/TestServiceFs2Grpc.scala.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,36 +11,86 @@ trait TestServiceFs2Grpc[F[_], A] {

object TestServiceFs2Grpc extends _root_.fs2.grpc.GeneratedCompanion[TestServiceFs2Grpc] {

def mkClient[F[_]: _root_.cats.effect.Async, A](dispatcher: _root_.cats.effect.std.Dispatcher[F], channel: _root_.io.grpc.Channel, mkMetadata: A => F[_root_.io.grpc.Metadata], clientOptions: _root_.fs2.grpc.client.ClientOptions): TestServiceFs2Grpc[F, A] = new TestServiceFs2Grpc[F, A] {
def noStreaming(request: hello.world.TestMessage, ctx: A): F[hello.world.TestMessage] = {
mkMetadata(ctx).flatMap { m =>
_root_.fs2.grpc.client.Fs2ClientCall[F](channel, hello.world.TestServiceGrpc.METHOD_NO_STREAMING, dispatcher, clientOptions).flatMap(_.unaryToUnaryCall(request, m))
}
}
def clientStreaming(request: _root_.fs2.Stream[F, hello.world.TestMessage], ctx: A): F[hello.world.TestMessage] = {
mkMetadata(ctx).flatMap { m =>
_root_.fs2.grpc.client.Fs2ClientCall[F](channel, hello.world.TestServiceGrpc.METHOD_CLIENT_STREAMING, dispatcher, clientOptions).flatMap(_.streamingToUnaryCall(request, m))
}
}
def serverStreaming(request: hello.world.TestMessage, ctx: A): _root_.fs2.Stream[F, hello.world.TestMessage] = {
_root_.fs2.Stream.eval(mkMetadata(ctx)).flatMap { m =>
_root_.fs2.Stream.eval(_root_.fs2.grpc.client.Fs2ClientCall[F](channel, hello.world.TestServiceGrpc.METHOD_SERVER_STREAMING, dispatcher, clientOptions)).flatMap(_.unaryToStreamingCall(request, m))
}
}
def bothStreaming(request: _root_.fs2.Stream[F, hello.world.TestMessage], ctx: A): _root_.fs2.Stream[F, hello.world.TestMessage] = {
_root_.fs2.Stream.eval(mkMetadata(ctx)).flatMap { m =>
_root_.fs2.Stream.eval(_root_.fs2.grpc.client.Fs2ClientCall[F](channel, hello.world.TestServiceGrpc.METHOD_BOTH_STREAMING, dispatcher, clientOptions)).flatMap(_.streamingToStreamingCall(request, m))
}
}
def mkClientFull[F[_], G[_]: _root_.cats.effect.Async, A](
dispatcher: _root_.cats.effect.std.Dispatcher[G],
channel: _root_.io.grpc.Channel,
clientAspect: _root_.fs2.grpc.client.ClientAspect[F, G, A],
clientOptions: _root_.fs2.grpc.client.ClientOptions
): TestServiceFs2Grpc[F, A] = new TestServiceFs2Grpc[F, A] {
def noStreaming(request: hello.world.TestMessage, ctx: A): F[hello.world.TestMessage] =
clientAspect.visitUnaryToUnary[hello.world.TestMessage, hello.world.TestMessage](
_root_.fs2.grpc.client.ClientCallContext(ctx, hello.world.TestServiceGrpc.METHOD_NO_STREAMING),
request,
(req, m) => _root_.fs2.grpc.client.Fs2ClientCall[G](channel, hello.world.TestServiceGrpc.METHOD_NO_STREAMING, dispatcher, clientOptions).flatMap(_.unaryToUnaryCall(req, m))
)
def clientStreaming(request: _root_.fs2.Stream[F, hello.world.TestMessage], ctx: A): F[hello.world.TestMessage] =
clientAspect.visitStreamingToUnary[hello.world.TestMessage, hello.world.TestMessage](
_root_.fs2.grpc.client.ClientCallContext(ctx, hello.world.TestServiceGrpc.METHOD_CLIENT_STREAMING),
request,
(req, m) => _root_.fs2.grpc.client.Fs2ClientCall[G](channel, hello.world.TestServiceGrpc.METHOD_CLIENT_STREAMING, dispatcher, clientOptions).flatMap(_.streamingToUnaryCall(req, m))
)
def serverStreaming(request: hello.world.TestMessage, ctx: A): _root_.fs2.Stream[F, hello.world.TestMessage] =
clientAspect.visitUnaryToStreaming[hello.world.TestMessage, hello.world.TestMessage](
_root_.fs2.grpc.client.ClientCallContext(ctx, hello.world.TestServiceGrpc.METHOD_SERVER_STREAMING),
request,
(req, m) => _root_.fs2.Stream.eval(_root_.fs2.grpc.client.Fs2ClientCall[G](channel, hello.world.TestServiceGrpc.METHOD_SERVER_STREAMING, dispatcher, clientOptions)).flatMap(_.unaryToStreamingCall(req, m))
)
def bothStreaming(request: _root_.fs2.Stream[F, hello.world.TestMessage], ctx: A): _root_.fs2.Stream[F, hello.world.TestMessage] =
clientAspect.visitStreamingToStreaming[hello.world.TestMessage, hello.world.TestMessage](
_root_.fs2.grpc.client.ClientCallContext(ctx, hello.world.TestServiceGrpc.METHOD_BOTH_STREAMING),
request,
(req, m) => _root_.fs2.Stream.eval(_root_.fs2.grpc.client.Fs2ClientCall[G](channel, hello.world.TestServiceGrpc.METHOD_BOTH_STREAMING, dispatcher, clientOptions)).flatMap(_.streamingToStreamingCall(req, m))
)
}

protected def serviceBinding[F[_]: _root_.cats.effect.Async, A](dispatcher: _root_.cats.effect.std.Dispatcher[F], serviceImpl: TestServiceFs2Grpc[F, A], mkCtx: _root_.io.grpc.Metadata => F[A], serverOptions: _root_.fs2.grpc.server.ServerOptions): _root_.io.grpc.ServerServiceDefinition = {
protected def serviceBindingFull[F[_], G[_]: _root_.cats.effect.Async, A](
dispatcher: _root_.cats.effect.std.Dispatcher[G],
serviceImpl: TestServiceFs2Grpc[F, A],
serviceAspect: _root_.fs2.grpc.server.ServiceAspect[F, G, A],
serverOptions: _root_.fs2.grpc.server.ServerOptions
) = {
_root_.io.grpc.ServerServiceDefinition
.builder(hello.world.TestServiceGrpc.SERVICE)
.addMethod(hello.world.TestServiceGrpc.METHOD_NO_STREAMING, _root_.fs2.grpc.server.Fs2ServerCallHandler[F](dispatcher, serverOptions).unaryToUnaryCall[hello.world.TestMessage, hello.world.TestMessage]((r, m) => mkCtx(m).flatMap(serviceImpl.noStreaming(r, _))))
.addMethod(hello.world.TestServiceGrpc.METHOD_CLIENT_STREAMING, _root_.fs2.grpc.server.Fs2ServerCallHandler[F](dispatcher, serverOptions).streamingToUnaryCall[hello.world.TestMessage, hello.world.TestMessage]((r, m) => mkCtx(m).flatMap(serviceImpl.clientStreaming(r, _))))
.addMethod(hello.world.TestServiceGrpc.METHOD_SERVER_STREAMING, _root_.fs2.grpc.server.Fs2ServerCallHandler[F](dispatcher, serverOptions).unaryToStreamingCall[hello.world.TestMessage, hello.world.TestMessage]((r, m) => _root_.fs2.Stream.eval(mkCtx(m)).flatMap(serviceImpl.serverStreaming(r, _))))
.addMethod(hello.world.TestServiceGrpc.METHOD_BOTH_STREAMING, _root_.fs2.grpc.server.Fs2ServerCallHandler[F](dispatcher, serverOptions).streamingToStreamingCall[hello.world.TestMessage, hello.world.TestMessage]((r, m) => _root_.fs2.Stream.eval(mkCtx(m)).flatMap(serviceImpl.bothStreaming(r, _))))
.addMethod(
hello.world.TestServiceGrpc.METHOD_NO_STREAMING,
_root_.fs2.grpc.server.Fs2ServerCallHandler[G](dispatcher, serverOptions).unaryToUnaryCall[hello.world.TestMessage, hello.world.TestMessage]{ (r, m) =>
serviceAspect.visitUnaryToUnary[hello.world.TestMessage, hello.world.TestMessage](
_root_.fs2.grpc.server.ServerCallContext(m, hello.world.TestServiceGrpc.METHOD_NO_STREAMING),
r,
(r, m) => serviceImpl.noStreaming(r, m)
)
}
)
.addMethod(
hello.world.TestServiceGrpc.METHOD_CLIENT_STREAMING,
_root_.fs2.grpc.server.Fs2ServerCallHandler[G](dispatcher, serverOptions).streamingToUnaryCall[hello.world.TestMessage, hello.world.TestMessage]{ (r, m) =>
serviceAspect.visitStreamingToUnary[hello.world.TestMessage, hello.world.TestMessage](
_root_.fs2.grpc.server.ServerCallContext(m, hello.world.TestServiceGrpc.METHOD_CLIENT_STREAMING),
r,
(r, m) => serviceImpl.clientStreaming(r, m)
)
}
)
.addMethod(
hello.world.TestServiceGrpc.METHOD_SERVER_STREAMING,
_root_.fs2.grpc.server.Fs2ServerCallHandler[G](dispatcher, serverOptions).unaryToStreamingCall[hello.world.TestMessage, hello.world.TestMessage]{ (r, m) =>
serviceAspect.visitUnaryToStreaming[hello.world.TestMessage, hello.world.TestMessage](
_root_.fs2.grpc.server.ServerCallContext(m, hello.world.TestServiceGrpc.METHOD_SERVER_STREAMING),
r,
(r, m) => serviceImpl.serverStreaming(r, m)
)
}
)
.addMethod(
hello.world.TestServiceGrpc.METHOD_BOTH_STREAMING,
_root_.fs2.grpc.server.Fs2ServerCallHandler[G](dispatcher, serverOptions).streamingToStreamingCall[hello.world.TestMessage, hello.world.TestMessage]{ (r, m) =>
serviceAspect.visitStreamingToStreaming[hello.world.TestMessage, hello.world.TestMessage](
_root_.fs2.grpc.server.ServerCallContext(m, hello.world.TestServiceGrpc.METHOD_BOTH_STREAMING),
r,
(r, m) => serviceImpl.bothStreaming(r, m)
)
}
)
.build()
}

Expand Down
Loading
Loading