add lightweight toUnary client runtime #506
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
naoh87
force-pushed
the
add_unary_client
branch
3 times, most recently
from
6002735 to
b552e9e
Compare
| object ReceiveState { | ||
| def init[F[_]: Sync, R]( | ||
| callback: Either[Throwable, R] => Unit, | ||
| pf: PartialFunction[StatusRuntimeException, Exception] |
There was a problem hiding this comment.
error adapter is Throwable. I will take a look tomorrow
There was a problem hiding this comment.
There was a problem hiding this comment.
Ok, then we are good. I thought I noticed a Throwable signature somewhere
naoh87
force-pushed
the
add_unary_client
branch
2 times, most recently
from
7380a60 to
542ee92
Compare
naoh87
force-pushed
the
add_unary_client
branch
2 times, most recently
from
19dd92c to
90fe995
Compare
naoh87
force-pushed
the
add_unary_client
branch
from
90fe995 to
aa8bfca
Compare
| import fs2.grpc.client.ClientOptions | ||
| import io.grpc._ | ||
|
|
||
| object Fs2UnaryCallHandler { |
There was a problem hiding this comment.
Suggested change
| object Fs2UnaryCallHandler { | |
| private[client] object Fs2UnaryCallHandler { |
| def sendError(error: Throwable): SyncIO[ReceiveState[R]] = | ||
| SyncIO(callback(Left(error))).as(new Done[R]) | ||
|
|
||
| def done(): SyncIO[ReceiveState[R]] = SyncIO(callback(Right(message))).as(new Done[R]) |
There was a problem hiding this comment.
Suggested change
| def done(): SyncIO[ReceiveState[R]] = SyncIO(callback(Right(message))).as(new Done[R]) | |
| def done: SyncIO[ReceiveState[R]] = SyncIO(callback(Right(message))).as(new Done[R]) |
| Ref.in(new PendingMessage[R]({ | ||
| case r: Right[Throwable, R] => callback(r) | ||
| case Left(e: StatusRuntimeException) => callback(Left(pf.lift(e).getOrElse(e))) | ||
| case l: Left[Throwable, R] => callback(l) |
There was a problem hiding this comment.
Perhaps Left(NonFatal(e)) => callback(e)?
There was a problem hiding this comment.
Never happens fatal error because this callback is called by ClientCall.Listener.
| override def onClose(status: Status, trailers: Metadata): Unit = { | ||
| if (status.isOk) { | ||
| state.get.flatMap { | ||
| case expected: PendingHalfClose[Response] => expected.done().flatMap(state.set) |
There was a problem hiding this comment.
Suggested change
| case expected: PendingHalfClose[Response] => expected.done().flatMap(state.set) | |
| case expected: PendingHalfClose[Response] => | |
| expected.done().flatMap(state.set) |
Merged
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What this PR do?
This PR eliminate using Dispatcher[F] from toUnary client runtime.
Benchmark
50 pararell request in 30sec with 2core from MBP to other server
This PR
main