add gRPC-client compression (#136)

* add upstream grpc-client compression

* fix review issues

docs/03-server-config.adoc

@@ -26,6 +26,8 @@ compression:
   grpc:
     server:
       enabled: false
+    client:
+      enabled: false
 
 cluster:
   include: "upstreams.yaml"
@@ -35,7 +37,8 @@ It configures following:
 
 - server is listening with gRCP API on `0.0.0.0:2449`
 - TLS is enabled
-- compression is disabled for gRPC server (enabled by default)
+- compression is disabled for gRPC-server (enabled by default)
+- compression is disabled for upstream gRPC-client (enabled by default)
 - server certificate is located at `server.crt` with the key for it at `server.p8.key`
 - the server requires a client authentication by TLS client certificate signed by `ca.crt` certificate
 - no JSON RPC is configured
@@ -72,12 +75,23 @@ a| `upstreams`
 |===
 
 === Compression
-Compression is enabled by default on the gRPC server
-(the server accepts compressed requests and can send compressed responses).
-Responses will be compressed if a client supports compression (sends relevant headers),
-otherwise, communication will be uncompressed.
-But if for some reason you need to disable compression forcibly,
-add the lines below to the config:
+Compression is enabled by default on the gRPC-server.
+Responses will be compressed if a client supports compression
+(sends relevant headers), otherwise, communication will be uncompressed.
+But if for some reason you need to disable responses compression
+forcibly (the server will still be able to accept compressed requests),
+add following lines to the config:
+[source,yaml]
+----
+compression:
+  grpc:
+    server:
+      enabled: false
+----
+Compression is enabled by default for gRPC-requests to upstreams.
+If you want to disable it,
+(gRPC-client will still be able to accept compressed responses),
+add following lines to the config:
 [source,yaml]
 ----
 compression:
@@ -85,6 +99,31 @@ compression:
     client:
       enabled: false
 ----
+Thus, all possible combinations of compression configuration for
+interacting dshackle grpc-client and grpc-server look like this:
+|===
+| Client | Server | Requests | Responses
+
+| enabled
+| enabled
+| compressed
+| compressed
+
+| enabled
+| disabled
+| compressed
+| plain
+
+| disabled
+| enabled
+| plain
+| compressed
+
+| disabled
+| disabled
+| plain
+| plain
+|===
 
 === Enabling JSON RPC proxy
 

src/main/kotlin/io/emeraldpay/dshackle/Config.kt

@@ -16,7 +16,16 @@
  */
 package io.emeraldpay.dshackle
 
-import io.emeraldpay.dshackle.config.*
+import io.emeraldpay.dshackle.config.CacheConfig
+import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.CompressionConfig
+import io.emeraldpay.dshackle.config.HealthConfig
+import io.emeraldpay.dshackle.config.MainConfig
+import io.emeraldpay.dshackle.config.MainConfigReader
+import io.emeraldpay.dshackle.config.MonitoringConfig
+import io.emeraldpay.dshackle.config.SignatureConfig
+import io.emeraldpay.dshackle.config.TokensConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig
 import org.bouncycastle.jce.provider.BouncyCastleProvider
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -110,6 +119,11 @@ open class Config(
         return mainConfig.upstreams
     }
 
+    @Bean
+    open fun compressionConfig(@Autowired mainConfig: MainConfig): CompressionConfig {
+        return mainConfig.compression
+    }
+
     @Bean
     open fun cacheConfig(@Autowired mainConfig: MainConfig): CacheConfig {
         return mainConfig.cache ?: CacheConfig()

src/main/kotlin/io/emeraldpay/dshackle/config/CompressionConfig.kt

@@ -1,6 +1,6 @@
 package io.emeraldpay.dshackle.config
 
-class CompressionConfig(
+open class CompressionConfig(
     var grpc: GRPC = GRPC()
 ) {
     /**

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.CompressionConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.*
@@ -62,6 +63,7 @@ import kotlin.math.abs
 open class ConfiguredUpstreams(
     private val fileResolver: FileResolver,
     private val config: UpstreamsConfig,
+    private val compressionConfig: CompressionConfig,
     private val callTargets: CallTargetsHolder,
     private val eventPublisher: ApplicationEventPublisher,
     @Qualifier("grpcChannelExecutor")
@@ -86,7 +88,7 @@ open class ConfiguredUpstreams(
             log.debug("Start upstream ${up.id}")
             if (up.connection is UpstreamsConfig.GrpcConnection) {
                 val options = up.options ?: UpstreamsConfig.Options()
-                buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options)
+                buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options, compressionConfig.grpc.clientEnabled)
             } else {
                 val chain = Global.chainById(up.chain)
                 if (chain == Chain.UNSPECIFIED) {
@@ -307,7 +309,8 @@ open class ConfiguredUpstreams(
     private fun buildGrpcUpstream(
         nodeId: Int?,
         config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
-        options: UpstreamsConfig.Options
+        options: UpstreamsConfig.Options,
+        compression: Boolean
     ) {
         if (!this::grpcUpstreamsScheduler.isInitialized) {
             grpcUpstreamsScheduler = Schedulers.fromExecutorService(
@@ -324,6 +327,7 @@ open class ConfiguredUpstreams(
             endpoint.host!!,
             endpoint.port,
             endpoint.auth,
+            compression,
             fileResolver,
             endpoint.upstreamRating,
             config.labels,

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt

@@ -34,6 +34,7 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
 import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
+import io.grpc.Codec
 import io.grpc.netty.NettyChannelBuilder
 import io.micrometer.core.instrument.Counter
 import io.micrometer.core.instrument.Metrics
@@ -63,6 +64,7 @@ class GrpcUpstreams(
     private val host: String,
     private val port: Int,
     private val auth: AuthConfig.ClientTlsAuth? = null,
+    private val compression: Boolean,
     private val fileResolver: FileResolver,
     private val nodeRating: Int,
     private val labels: UpstreamsConfig.Labels,
@@ -94,7 +96,10 @@ class GrpcUpstreams(
             chanelBuilder.usePlaintext()
         }
 
-        val client = ReactorBlockchainGrpc.newReactorStub(chanelBuilder.build())
+        var client = ReactorBlockchainGrpc.newReactorStub(chanelBuilder.build())
+        if (compression) {
+            client = client.withCompression(Codec.Gzip().messageEncoding)
+        }
         this.client = client
 
         val statusSubscription = AtomicReference<Disposable>()

src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy

@@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.startup
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.config.ChainsConfig
+import io.emeraldpay.dshackle.config.CompressionConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
 import io.emeraldpay.dshackle.upstream.CallTargetsHolder
@@ -20,6 +21,7 @@ class ConfiguredUpstreamsSpec extends Specification {
         def configurer = new ConfiguredUpstreams(
                 Stub(FileResolver),
                 Stub(UpstreamsConfig),
+                Stub(CompressionConfig),
                 callTargetsHolder,
                 Mock(ApplicationEventPublisher),
                 Executors.newFixedThreadPool(1),
@@ -47,6 +49,7 @@ class ConfiguredUpstreamsSpec extends Specification {
         def configurer = new ConfiguredUpstreams(
                 Stub(FileResolver),
                 Stub(UpstreamsConfig),
+                Stub(CompressionConfig),
                 callTargetsHolder,
                 Mock(ApplicationEventPublisher),
                 Executors.newFixedThreadPool(1),
@@ -73,6 +76,7 @@ class ConfiguredUpstreamsSpec extends Specification {
         def configurer = new ConfiguredUpstreams(
                 Stub(FileResolver),
                 Stub(UpstreamsConfig),
+                Stub(CompressionConfig),
                 callTargetsHolder,
                 Mock(ApplicationEventPublisher),
                 Executors.newFixedThreadPool(1),
@@ -94,6 +98,7 @@ class ConfiguredUpstreamsSpec extends Specification {
         def configurer = new ConfiguredUpstreams(
                 Stub(FileResolver),
                 Stub(UpstreamsConfig),
+                Stub(CompressionConfig),
                 callTargetsHolder,
                 Mock(ApplicationEventPublisher),
                 Executors.newFixedThreadPool(1),
@@ -120,6 +125,7 @@ class ConfiguredUpstreamsSpec extends Specification {
         def configurer = new ConfiguredUpstreams(
                 Stub(FileResolver),
                 Stub(UpstreamsConfig),
+                Stub(CompressionConfig),
                 callTargetsHolder,
                 Mock(ApplicationEventPublisher),
                 Executors.newFixedThreadPool(1),