Merge pull request #10 from dtuchs/master

Added support for snappy compression / decompression

Author: Adven27

README.md

@@ -90,7 +90,27 @@ docker run -e GRPC_SERVER_MAXHEADERLISTSIZE=1000 adven27/grpc-wiremock
 In case you don't need to change proto files, you can build your own image with precompiled protos.  
 See an [example](/example/Dockerfile)
 
-#### 3. Use in load testing
+#### 3. Use with snappy compresser/decompresser
+
+Snappy support can be enabled using EXT_CODECS env variable as follows:
+```posh
+docker run -e EXT_CODECS=snappy adven27/grpc-wiremock
+```
+Also in docker-compose:
+```posh
+    image: adven27/grpc-wiremock
+    ports:
+      - "12085:50000" # grpc port
+      - "8088:8888" # http serve port
+    volumes:
+      - ./example/proto:/proto
+    environment:
+      - EXT_CODECS=snappy
+```
+<sub>*gzip compression supported by default</sub>
+
+
+#### 4. Use in load testing
 
 To increase performance some Wiremock related options may be tuned either directly or by enabling the "load" profile. 
 Next two commands are identical:

build.gradle

@@ -21,6 +21,7 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter'
     implementation 'org.springframework.boot:spring-boot-starter-aop'
     implementation "io.grpc:grpc-all:${grpcVersion}"
+    implementation 'org.xerial.snappy:snappy-java:1.1.8.4'
     implementation 'com.github.tomakehurst:wiremock-standalone:2.27.2'
 }
 

src/main/java/io/adven/grpc/wiremock/CodecRegistry.java

@@ -0,0 +1,46 @@
+package io.adven.grpc.wiremock;
+
+import io.grpc.Codec;
+import io.grpc.CompressorRegistry;
+import io.grpc.DecompressorRegistry;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Map;
+
+@Component
+@PropertySource("classpath:application.properties")
+public class CodecRegistry {
+
+    @Value("#{'${external.codecs}'.split(',')}")
+    private List<String> externalCodecs;
+    private Map<String, Codec> codecs;
+
+    public CodecRegistry(Map<String, Codec> codecs) {
+        this.codecs = codecs;
+    }
+
+    public CompressorRegistry compressorRegistry() {
+        CompressorRegistry compressorRegistry = CompressorRegistry.getDefaultInstance();
+        for (String codec : externalCodecs) {
+            Codec codecImpl = codecs.get(codec);
+            if (codecImpl != null) {
+                compressorRegistry.register(codecs.get(codec));
+            }
+        }
+        return compressorRegistry;
+    }
+
+    public DecompressorRegistry decompressorRegistry() {
+        DecompressorRegistry decompressorRegistry = DecompressorRegistry.getDefaultInstance();
+        for (String codec : externalCodecs) {
+            Codec codecImpl = codecs.get(codec);
+            if (codecImpl != null) {
+                decompressorRegistry = decompressorRegistry.with(codecs.get(codec), true);
+            }
+        }
+        return decompressorRegistry;
+    }
+}

src/main/java/io/adven/grpc/wiremock/GrpcWiremock.java

@@ -40,21 +40,24 @@ public static class GrpcServer {
         private final ServerProperties serverProperties;
         private final List<BindableService> services;
         private Server server;
+        private CodecRegistry codecRegistry;
         private final CountDownLatch latch;
 
-        public GrpcServer(ServerProperties serverProperties, List<BindableService> services) {
+        public GrpcServer(ServerProperties serverProperties, CodecRegistry codecRegistry, List<BindableService> services) {
             this.serverProperties = serverProperties;
+            this.codecRegistry = codecRegistry;
             this.services = services;
             this.latch = new CountDownLatch(1);
         }
 
         public void start(int port) throws IOException {
             NettyServerBuilder builder = NettyServerBuilder.forPort(port)
                 .intercept(new ExceptionHandler())
+                .compressorRegistry(codecRegistry.compressorRegistry())
+                .decompressorRegistry(codecRegistry.decompressorRegistry())
                 .addService(ProtoReflectionService.newInstance());
 
             setProperties(builder);
-
             services.forEach(builder::addService);
             server = builder.build().start();
             LOG.info(summary(server));
@@ -64,29 +67,29 @@ public void start(int port) throws IOException {
         private void setProperties(NettyServerBuilder builder) {
             if (serverProperties.getMaxHeaderListSize() != null) {
                 int val = Math.toIntExact(serverProperties.getMaxHeaderListSize().toBytes());
-                LOG.info("Set maxHeaderListSize = {}" , val);
+                LOG.info("Set maxHeaderListSize = {}", val);
                 builder.maxHeaderListSize(val);
             }
             if (serverProperties.getMaxMessageSize() != null) {
                 int val = Math.toIntExact(serverProperties.getMaxMessageSize().toBytes());
-                LOG.info("Set maxMessageSize = {}" , val);
+                LOG.info("Set maxMessageSize = {}", val);
                 builder.maxMessageSize(val);
             }
             if (serverProperties.getMaxInboundMetadataSize() != null) {
                 int val = Math.toIntExact(serverProperties.getMaxInboundMetadataSize().toBytes());
-                LOG.info("Set maxInboundMetadataSize = {}" , val);
+                LOG.info("Set maxInboundMetadataSize = {}", val);
                 builder.maxInboundMetadataSize(val);
             }
             if (serverProperties.getMaxInboundMessageSize() != null) {
                 int val = Math.toIntExact(serverProperties.getMaxInboundMessageSize().toBytes());
-                LOG.info("Set maxInboundMessageSize = {}" , val);
+                LOG.info("Set maxInboundMessageSize = {}", val);
                 builder.maxInboundMessageSize(val);
             }
         }
 
         private String summary(Server server) {
             return "Started " + server + "\nRegistered services:\n" +
-                server.getServices().stream().map(s -> " * " + s.getServiceDescriptor().getName()).collect(joining("\n"));
+                    server.getServices().stream().map(s -> " * " + s.getServiceDescriptor().getName()).collect(joining("\n"));
         }
 
         private void startDaemonAwaitThread() {

src/main/java/io/adven/grpc/wiremock/codec/SnappyCodec.java

@@ -0,0 +1,28 @@
+package io.adven.grpc.wiremock.codec;
+
+import io.grpc.Codec;
+import org.springframework.stereotype.Component;
+import org.xerial.snappy.SnappyFramedInputStream;
+import org.xerial.snappy.SnappyFramedOutputStream;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+@Component("snappy")
+public class SnappyCodec implements Codec {
+    @Override
+    public String getMessageEncoding() {
+        return "snappy";
+    }
+
+    @Override
+    public InputStream decompress(InputStream is) throws IOException {
+        return new SnappyFramedInputStream(is);
+    }
+
+    @Override
+    public OutputStream compress(OutputStream os) throws IOException {
+        return new SnappyFramedOutputStream(os);
+    }
+}

src/main/resources/application.properties

@@ -7,7 +7,8 @@ wiremock.server.asynchronousResponseEnabled=false
 wiremock.server.asynchronousResponseThreads=10
 wiremock.server.stubRequestLoggingDisabled=false
 wiremock.server.verbose=true
-
+#
+external.codecs=${EXT_CODECS:#{""}}
 #---
 spring.config.activate.on-profile=load
 #