-
Notifications
You must be signed in to change notification settings - Fork 124
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add support for service client connection reuse.
Refactored ClientState into GrpcChannel. Modified generated service clients to use GrpcChannel instead of GrpcClientSettings, with an apply() shim to create a GrpcChannel from GrpcClientSettings. Added GrpChannelSpec to confirm that connection is being reused.
- Loading branch information
1 parent
c34a630
commit 911bd42
Showing
9 changed files
with
190 additions
and
160 deletions.
There are no files selected for viewing
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
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
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
89 changes: 89 additions & 0 deletions
89
interop-tests/src/test/scala/akka/grpc/scaladsl/GrpcChannelSpec.scala
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,89 @@ | ||
| /* | ||
| * Copyright (C) 2020-2021 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.grpc.scaladsl | ||
|
|
||
| import scala.concurrent.Await | ||
| import scala.concurrent.duration._ | ||
|
|
||
| import akka.actor.ActorSystem | ||
| import akka.grpc.scaladsl.tools.MutableServiceDiscovery | ||
| import akka.grpc.{ GrpcChannel, GrpcClientSettings } | ||
| import akka.http.scaladsl.Http | ||
| import akka.http.scaladsl.model.RemoteAddress | ||
| import akka.http.scaladsl.server.Directives | ||
| import akka.stream.SystemMaterializer | ||
| import com.typesafe.config.{ Config, ConfigFactory } | ||
| import example.myapp.helloworld.grpc.helloworld._ | ||
| import org.scalatest.BeforeAndAfterAll | ||
| import org.scalatest.concurrent.ScalaFutures | ||
| import org.scalatest.matchers.should.Matchers | ||
| import org.scalatest.time.Span | ||
| import org.scalatest.wordspec.AnyWordSpec | ||
|
|
||
| class GrpcClientSpecNetty extends GrpcChannelSpec() | ||
|
|
||
| class GrpcChannelSpec(config: Config = ConfigFactory.load()) | ||
| extends AnyWordSpec | ||
| with Matchers | ||
| with BeforeAndAfterAll | ||
| with ScalaFutures { | ||
| implicit val system = ActorSystem("GrpcChannelSpec", config) | ||
| implicit val mat = SystemMaterializer(system).materializer | ||
| implicit val ec = system.dispatcher | ||
|
|
||
| override implicit val patienceConfig: PatienceConfig = PatienceConfig(5.seconds, Span(10, org.scalatest.time.Millis)) | ||
|
|
||
| "GrpcChannel" should { | ||
| "create separate connections for separate channels" in { | ||
| val clientAddresses = scala.collection.concurrent.TrieMap.empty[RemoteAddress.IP, Unit] | ||
| val service = new CountingGreeterServiceImpl() | ||
| val handler = GreeterServiceHandler(service) | ||
| val route = Directives.extractClientIP { clientIp => | ||
| clientAddresses.put(clientIp.toIP.get, ()) | ||
| Directives.handle(handler) | ||
| } | ||
|
|
||
| val server = Http().newServerAt("127.0.0.1", 0).bind(route).futureValue | ||
|
|
||
| val discovery = MutableServiceDiscovery(List(server)) | ||
| val settings = GrpcClientSettings.usingServiceDiscovery("greeter", discovery).withTls(false) | ||
|
|
||
| val greeterClient1 = GreeterServiceClient(settings) | ||
| greeterClient1.sayHello(HelloRequest(s"Hello 1")).futureValue | ||
|
|
||
| val greeterClient2 = GreeterServiceClient(settings) | ||
| greeterClient2.sayHello(HelloRequest(s"Hello 2")).futureValue | ||
|
|
||
| clientAddresses.size should be(2) | ||
| } | ||
| "reuse a single connection for a shared channel" in { | ||
| val clientAddresses = scala.collection.concurrent.TrieMap.empty[RemoteAddress.IP, Unit] | ||
| val service = new CountingGreeterServiceImpl() | ||
| val handler = GreeterServiceHandler(service) | ||
| val route = Directives.extractClientIP { clientIp => | ||
| clientAddresses.put(clientIp.toIP.get, ()) | ||
| Directives.handle(handler) | ||
| } | ||
|
|
||
| val server = Http().newServerAt("127.0.0.1", 0).bind(route).futureValue | ||
|
|
||
| val discovery = MutableServiceDiscovery(List(server)) | ||
| val settings = GrpcClientSettings.usingServiceDiscovery("greeter", discovery).withTls(false) | ||
| val channel = GrpcChannel(settings) | ||
|
|
||
| val greeterClient1 = GreeterServiceClient(channel) | ||
| greeterClient1.sayHello(HelloRequest(s"Hello 0")).futureValue | ||
|
|
||
| val greeterClient2 = GreeterServiceClient(channel) | ||
| greeterClient2.sayHello(HelloRequest(s"Hello 1")).futureValue | ||
|
|
||
| clientAddresses.size should be(1) | ||
| } | ||
| } | ||
|
|
||
| override def afterAll(): Unit = { | ||
| Await.result(system.terminate(), 10.seconds) | ||
| } | ||
| } |
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| /* | ||
| * Copyright (C) 2018-2021 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.grpc | ||
|
|
||
| import java.util.concurrent.CompletionStage | ||
|
|
||
| import scala.compat.java8.FutureConverters._ | ||
| import scala.concurrent.Future | ||
|
|
||
| import akka.Done | ||
| import akka.actor.ClassicActorSystemProvider | ||
| import akka.annotation.InternalApi | ||
| import akka.grpc.internal.{ ChannelUtils, InternalChannel } | ||
| import akka.grpc.scaladsl.Grpc | ||
|
|
||
| class GrpcChannel(val settings: GrpcClientSettings, @InternalApi val internalChannel: InternalChannel)( | ||
| implicit sys: ClassicActorSystemProvider) { | ||
|
|
||
| Grpc(sys).registerChannel(this) | ||
|
|
||
| def closeCS(): CompletionStage[Done] = | ||
| close().toJava | ||
|
|
||
| def closedCS(): CompletionStage[Done] = | ||
| closed().toJava | ||
|
|
||
| def close(): Future[akka.Done] = { | ||
| Grpc(sys).deregisterChannel(this) | ||
| ChannelUtils.close(internalChannel) | ||
| } | ||
|
|
||
| def closed(): Future[akka.Done] = | ||
| internalChannel.done | ||
| } | ||
|
|
||
| object GrpcChannel { | ||
| def apply(settings: GrpcClientSettings)(implicit sys: ClassicActorSystemProvider): GrpcChannel = { | ||
| new GrpcChannel( | ||
| settings, | ||
| ChannelUtils.create(settings, akka.event.Logging(sys.classicSystem, classOf[GrpcChannel]))) | ||
| } | ||
| } |
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
66 changes: 0 additions & 66 deletions
66
runtime/src/main/scala/akka/grpc/internal/ClientState.scala
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.