Support compressed frames.

Author: alistairjevans

ext/hyper_ruby/src/grpc.rs

@@ -65,25 +65,20 @@ pub fn is_grpc_request(request: &HyperRequest<Bytes>) -> bool {
     true
 }
 
-pub fn decode_grpc_frame(bytes: &[u8]) -> Option<Bytes> {
+pub fn decode_grpc_frame(bytes: &Bytes) -> Option<(bool, Bytes)> {
     if bytes.len() < GRPC_HEADER_SIZE {
         return None;
     }
 
     // GRPC frame format:
     // Compressed-Flag (1 byte) | Message-Length (4 bytes) | Message
     let compressed = bytes[0] != 0;
-    if compressed {
-        // We don't support compression yet
-        return None;
-    }
-
     let message_len = u32::from_be_bytes([bytes[1], bytes[2], bytes[3], bytes[4]]) as usize;
     if bytes.len() < GRPC_HEADER_SIZE + message_len {
         return None;
     }
 
-    Some(Bytes::copy_from_slice(&bytes[GRPC_HEADER_SIZE..GRPC_HEADER_SIZE + message_len]))
+    Some((compressed, bytes.slice(GRPC_HEADER_SIZE..GRPC_HEADER_SIZE + message_len)))
 }
 
 pub fn encode_grpc_frame(message: &[u8]) -> Bytes {

ext/hyper_ruby/src/lib.rs

@@ -452,6 +452,7 @@ fn init(ruby: &Ruby) -> Result<(), MagnusError> {
     grpc_request_class.define_method("body", method!(GrpcRequest::body, 0))?;
     grpc_request_class.define_method("fill_body", method!(GrpcRequest::fill_body, 1))?;
     grpc_request_class.define_method("body_size", method!(GrpcRequest::body_size, 0))?;
+    grpc_request_class.define_method("compressed?", method!(GrpcRequest::is_compressed, 0))?;
     grpc_request_class.define_method("inspect", method!(GrpcRequest::inspect, 0))?;
 
     Ok(())

ext/hyper_ruby/src/request.rs

@@ -75,11 +75,15 @@ impl FillBuffer for Request {
 
 impl FillBuffer for GrpcRequest {
     fn get_body_bytes(&self) -> Bytes {
-        grpc::decode_grpc_frame(self.request.body()).unwrap_or_else(|| Bytes::new())
+        if let Some((_, message)) = grpc::decode_grpc_frame(self.request.body()) {
+            message
+        } else {
+            Bytes::new()
+        }
     }
 
     fn get_body_size(&self) -> usize {
-        if let Some(message) = grpc::decode_grpc_frame(self.request.body()) {
+        if let Some((_, message)) = grpc::decode_grpc_frame(self.request.body()) {
             message.len()
         } else {
             0
@@ -217,6 +221,14 @@ impl GrpcRequest {
         self.fill_buffer(buffer)
     }
 
+    pub fn is_compressed(&self) -> bool {
+        if let Some((compressed, _)) = grpc::decode_grpc_frame(self.request.body()) {
+            compressed
+        } else {
+            false
+        }
+    }
+
     pub fn inspect(&self) -> RString {
         let body_size = self.body_size();
         RString::new(&format!("#<HyperRuby::GrpcRequest service={} method={} body_size={}>", self.service, self.method, body_size))

ext/hyper_ruby/src/response.rs

@@ -194,7 +194,7 @@ impl GrpcResponse {
     pub fn body(&self) -> RString {
         // For gRPC responses, decode the frame
         let body = self.response.body().get_data();
-        if let Some(message) = grpc::decode_grpc_frame(body) {
+        if let Some((_, message)) = grpc::decode_grpc_frame(body) {
             RString::from_slice(message.as_ref())
         } else {
             RString::new("")

test/test_bad_http_requests.rb

@@ -104,20 +104,24 @@ def test_header_timeout
 
       # Sleep longer than the timeout
       sleep 1.5
-      
-      # Try to send the rest of the headers, but the connection should be closed
-      socket.write("Content-Length: 0\r\n")
-      socket.write("Connection: close\r\n")
-      socket.write("\r\n")
-      
-      # Attempt to read response - should be a timeout or connection closed
-      response = read_http_response(socket)
-      socket.close
-      
-      # The server might respond with a 408 timeout, or might just close the connection
-      # Both behaviors are acceptable according to HTTP/1.1 spec
-      if response[:status]
-        assert_equal 408, response[:status].split(" ")[1].to_i  # Request Timeout if we got a response
+      begin
+        # Try to send the rest of the headers, but the connection should be closed
+        socket.write("Content-Length: 0\r\n")
+        socket.write("Connection: close\r\n")
+        socket.write("\r\n")
+        
+        # Attempt to read response - should be a timeout or connection closed
+        response = read_http_response(socket)
+        socket.close
+        
+        # The server might respond with a 408 timeout, or might just close the connection
+        # Both behaviors are acceptable according to HTTP/1.1 spec
+        if response[:status]
+          assert_equal 408, response[:status].split(" ")[1].to_i  # Request Timeout if we got a response
+        end
+      rescue Errno::EPIPE
+      # This is expected if the server closes the connection due to timeout/error
+      # This is not an error, so we don't need to assert anything
       end
     end
   end

test/test_grpc.rb

@@ -1,9 +1,11 @@
 # frozen_string_literal: true
 
 require "test_helper"
+require 'zlib'
 require_relative "echo_pb"
 require_relative "echo_services_pb"
 
+
 class TestGrpc < HyperRubyTest
   def test_grpc_request
     buffer = String.new(capacity: 1024)
@@ -129,6 +131,28 @@ def test_grpc_over_unix_socket
     end
   end
 
+  def test_grpc_compression
+    buffer = String.new(capacity: 1024)
+    compression_options = GRPC::Core::CompressionOptions.new(default_algorithm: :gzip)
+    compression_channel_args = compression_options.to_channel_arg_hash
+
+    with_server(-> (request) { handler_grpc_compressed(request, buffer) }) do |_client|
+      stub = Echo::Echo::Stub.new(
+        "127.0.0.1:3010",
+        :this_channel_is_insecure,
+        channel_args: {
+          'grpc.enable_http_proxy' => 0,
+        }.merge(compression_channel_args)
+      )
+      
+      request = Echo::EchoRequest.new(message: "Hello Compressed GRPC " + ("a" * 10000))
+      response = stub.echo(request)
+      
+      assert_instance_of Echo::EchoResponse, response
+      assert_equal "Decompressed: Hello Compressed GRPC " + ("a" * 10000), response.message
+    end
+  end
+
   private
 
   def handler_grpc(request, buffer)
@@ -185,4 +209,26 @@ def handler_grpc_status(request)
       HyperRuby::GrpcResponse.error(2, "unknown error") # UNKNOWN = 2
     end
   end
+
+  def handler_grpc_compressed(request, buffer)
+    assert_equal "application/grpc", request.header("content-type")
+    assert_equal "echo.Echo", request.service
+    assert_equal "Echo", request.method
+    # Check if the message is compressed
+    assert request.compressed?, "Expected request to be compressed"
+    
+    # Get the compressed message
+    request.fill_body(buffer)
+    
+    decompressed = Zlib.gunzip(buffer)
+    echo_request = Echo::EchoRequest.decode(decompressed)
+    
+    echo_response = Echo::EchoResponse.new(message: "Decompressed: #{echo_request.message}")
+    response_data = Echo::EchoResponse.encode(echo_response)
+    
+    HyperRuby::GrpcResponse.new(0, response_data)
+  rescue => e
+    pp e
+    raise e
+  end
 end