More tests for the grpc behaviour, but looking good so far....

Author: alistairjevans

ext/hyper_ruby/src/grpc.rs

@@ -124,5 +124,5 @@ pub fn create_grpc_error_response(http_status: u16, grpc_status: u32, message: &
     }
 
     // Create response with custom body that includes trailers
-    builder.body(BodyWithTrailers::new(Bytes::new(), trailers)).unwrap()
+    builder.body(BodyWithTrailers::new(Bytes::new(), Some(trailers))).unwrap()
 } 

ext/hyper_ruby/src/lib.rs

@@ -23,7 +23,7 @@ use tokio::task::JoinHandle;
 use crossbeam_channel;
 
 use hyper::service::service_fn;
-use hyper::{Error, Request as HyperRequest, Response as HyperResponse, StatusCode, Method, header::HeaderMap};
+use hyper::{Error, Request as HyperRequest, Response as HyperResponse, StatusCode};
 use hyper::body::Incoming;
 use hyper_util::rt::TokioIo;
 use hyper_util::server::conn::auto;
@@ -35,6 +35,9 @@ use log::{debug, info, warn};
 
 use env_logger;
 use crate::response::BodyWithTrailers;
+use std::sync::Once;
+
+static LOGGER_INIT: Once = Once::new();
 
 #[global_allocator]
 static GLOBAL: Jemalloc = Jemalloc;
@@ -43,13 +46,15 @@ static GLOBAL: Jemalloc = Jemalloc;
 struct ServerConfig {
     bind_address: String,
     tokio_threads: Option<usize>,
+    debug: bool,
 }
 
 impl ServerConfig {
     fn new() -> Self {
         Self {
             bind_address: String::from("127.0.0.1:3000"),
             tokio_threads: None,
+            debug: false,
         }
     }
 }
@@ -92,6 +97,19 @@ impl Server {
             server_config.tokio_threads = Some(usize::try_convert(tokio_threads)?);
         }
 
+        if let Some(debug) = config.get(magnus::Symbol::new("debug")) {
+            server_config.debug = bool::try_convert(debug)?;
+        }
+
+        // Initialize logging if debug is enabled, but only do it once
+        if server_config.debug {
+            LOGGER_INIT.call_once(|| {
+                env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("hyper=debug,h2=debug"))
+                    .write_style(env_logger::WriteStyle::Always)
+                    .init();
+            });
+        }
+
         Ok(())
     }
 
@@ -116,26 +134,26 @@ impl Server {
                     Ok(work_request) => {
                         let hyper_request = work_request.request;
 
-                        println!("\nProcessing request:");
-                        println!("  Method: {}", hyper_request.method());
-                        println!("  Path: {}", hyper_request.uri().path());
-                        println!("  Headers: {:?}", hyper_request.headers());
+                        debug!("Processing request:");
+                        debug!("  Method: {}", hyper_request.method());
+                        debug!("  Path: {}", hyper_request.uri().path());
+                        debug!("  Headers: {:?}", hyper_request.headers());
 
                         // Convert to appropriate request type
                         let value = if grpc::is_grpc_request(&hyper_request) {
-                            println!("Request identified as gRPC");
+                            debug!("Request identified as gRPC");
                             if let Some(grpc_request) = GrpcRequest::new(hyper_request) {
                                 grpc_request.into_value()
                             } else {
-                                println!("Failed to create GrpcRequest");
+                                warn!("Failed to create GrpcRequest");
                                 // Invalid gRPC request path
                                 let response = GrpcResponse::error(3_u32.into_value(), RString::new("Invalid gRPC request path")).unwrap()
                                     .into_hyper_response();
-                                work_request.response_tx.send(response).unwrap_or_else(|e| println!("Failed to send response: {:?}", e));
+                                work_request.response_tx.send(response).unwrap_or_else(|e| warn!("Failed to send response: {:?}", e));
                                 continue;
                             }
                         } else {
-                            println!("Request identified as HTTP");
+                            debug!("Request identified as HTTP");
                             Request::new(hyper_request).into_value()
                         };
 
@@ -147,19 +165,19 @@ impl Server {
                                 } else if let Ok(http_response) = Obj::<Response>::try_convert(result) {
                                     (*http_response).clone().into_hyper_response()
                                 } else {
-                                    println!("Block returned invalid response type");
+                                    warn!("Block returned invalid response type");
                                     create_error_response("Internal server error")
                                 }
                             },
                             Err(e) => {
-                                println!("Block call failed: {:?}", e);
+                                warn!("Block call failed: {:?}", e);
                                 create_error_response("Internal server error")
                             }
                         };
 
                         match work_request.response_tx.send(hyper_response) {
                             Ok(_) => (),
-                            Err(e) => println!("Failed to send response: {:?}", e),
+                            Err(e) => warn!("Failed to send response: {:?}", e),
                         }
                     }
                     Err(_) => {
@@ -253,7 +271,7 @@ impl Server {
         if bind_address.starts_with("unix:") {
             let path = bind_address.trim_start_matches("unix:");
             std::fs::remove_file(path).unwrap_or_else(|e| {
-                println!("Failed to remove socket file: {:?}", e);
+                warn!("Failed to remove socket file: {:?}", e);
             });
         }
 
@@ -349,20 +367,13 @@ fn create_error_response(error_message: &str) -> HyperResponse<BodyWithTrailers>
     let builder = HyperResponse::builder()
         .status(StatusCode::INTERNAL_SERVER_ERROR)
         .header("content-type", "text/plain");
-
-    let trailers = HeaderMap::new();
 
-    builder.body(BodyWithTrailers::new(Bytes::from(error_message.to_string()), trailers))
+    builder.body(BodyWithTrailers::new(Bytes::from(error_message.to_string()), None))
         .unwrap()
 }
 
 #[magnus::init]
 fn init(ruby: &Ruby) -> Result<(), MagnusError> {
-    // Initialize logging
-    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("hyper=debug,h2=debug"))
-        .write_style(env_logger::WriteStyle::Always)
-        .init();
-
     let module = ruby.define_module("HyperRuby")?;
 
     let server_class = module.define_class("Server", ruby.class_object())?;

ext/hyper_ruby/src/request.rs

@@ -8,6 +8,7 @@ use hyper::Request as HyperRequest;
 use rb_sys::{rb_str_set_len, rb_str_modify, rb_str_modify_expand, rb_str_capacity, RSTRING_PTR, VALUE};
 
 use crate::grpc;
+use log::debug;
 
 // Trait for common buffer filling behavior
 trait FillBuffer {
@@ -144,15 +145,15 @@ impl Request {
 
 impl GrpcRequest {
     pub fn new(request: HyperRequest<Bytes>) -> Option<Self> {
-        println!("Creating GrpcRequest from path: {}", request.uri().path());
+        debug!("Creating GrpcRequest from path: {}", request.uri().path());
 
         // Path format could be "/Echo" or "/echo.Echo/Echo" - handle both
         let path = request.uri().path();
         let parts: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
-        println!("  Path parts: {:?}", parts);
+        debug!("  Path parts: {:?}", parts);
 
         if parts.is_empty() {
-            println!("  Failed: Empty path");
+            debug!("  Failed: Empty path");
             return None;
         }
 
@@ -164,7 +165,7 @@ impl GrpcRequest {
             (format!("echo.{}", parts[0]), parts[0].to_string())
         };
 
-        println!("  Extracted service: {}, method: {}", service, method);
+        debug!("  Extracted service: {}, method: {}", service, method);
 
         Some(Self {
             request,

ext/hyper_ruby/src/response.rs

@@ -21,11 +21,11 @@ impl std::error::Error for ResponseError {}
 pub struct BodyWithTrailers {
     data: Bytes,
     trailers_sent: bool,
-    trailers: HeaderMap,
+    trailers: Option<HeaderMap>,
 }
 
 impl BodyWithTrailers {
-    pub fn new(data: Bytes, trailers: HeaderMap) -> Self {
+    pub fn new(data: Bytes, trailers: Option<HeaderMap>) -> Self {
         Self {
             data,
             trailers_sent: false,
@@ -54,7 +54,9 @@ impl Body for BodyWithTrailers {
 
         if !self.trailers_sent {
             self.trailers_sent = true;
-            return std::task::Poll::Ready(Some(Ok(Frame::trailers(self.trailers.clone()))));
+            if let Some(trailers) = &self.trailers {
+                return std::task::Poll::Ready(Some(Ok(Frame::trailers(trailers.clone()))));
+            }
         }
 
         std::task::Poll::Ready(None)
@@ -85,19 +87,16 @@ impl Response {
             Ok(ForEach::Continue)
         }).unwrap();
 
-        let mut trailers = HeaderMap::new();
-        trailers.insert("grpc-status", "0".parse().unwrap());
-
         if body.len() > 0 {
             unsafe {
                 let rust_body = Bytes::copy_from_slice(body.as_slice());
-                match builder.body(BodyWithTrailers::new(rust_body, trailers)) {
+                match builder.body(BodyWithTrailers::new(rust_body, None)) {
                     Ok(response) => Ok(Self { response }),
                     Err(_) => Err(MagnusError::new(magnus::exception::runtime_error(), "Failed to create response"))
                 }
             }
         } else {
-            match builder.body(BodyWithTrailers::new(Bytes::new(), trailers)) {
+            match builder.body(BodyWithTrailers::new(Bytes::new(), None)) {
                 Ok(response) => Ok(Self { response }),
                 Err(_) => Err(MagnusError::new(magnus::exception::runtime_error(), "Failed to create response"))
             }
@@ -141,7 +140,7 @@ impl GrpcResponse {
         trailers.insert("grpc-accept-encoding", "identity".parse().unwrap());
         trailers.insert("accept-encoding", "identity".parse().unwrap());
 
-        Ok(Self { response: builder.body(BodyWithTrailers::new(framed_message, trailers)).unwrap() })
+        Ok(Self { response: builder.body(BodyWithTrailers::new(framed_message, Some(trailers))).unwrap() })
     }
 
     pub fn error(status: Value, message: RString) -> Result<Self, MagnusError> {
@@ -161,7 +160,7 @@ impl GrpcResponse {
             trailers.insert("grpc-message", message_str.parse().unwrap());
         }
 
-        Ok(Self { response: builder.body(BodyWithTrailers::new(Bytes::new(), trailers)).unwrap() })
+        Ok(Self { response: builder.body(BodyWithTrailers::new(Bytes::new(), Some(trailers))).unwrap() })
     }
 
     pub fn status(&self) -> u16 {

test/test_hyper_ruby.rb

@@ -99,7 +99,7 @@ def test_head
 
   def test_blocking
     buffer = String.new(capacity: 1024)
-    with_server(-> (request) { handler_grpc(request, buffer) }) do |client|
+    with_server(-> (request) { handler_accept(request, buffer) }) do |client|
       gets
     end
   end
@@ -146,20 +146,77 @@ def test_grpc_request
       # Create request message
       request = Echo::EchoRequest.new(message: "Hello GRPC")
 
-      puts "\n=== Starting gRPC request ==="
       # Make the gRPC call
       response = stub.echo(request)
-      puts "=== gRPC request complete ===\n"
 
       # Check the response
       assert_instance_of Echo::EchoResponse, response
       assert_equal "Hello GRPC response", response.message
     end
   end
 
+  def test_concurrent_grpc_requests
+    buffer = String.new(capacity: 1024)
+    with_server(-> (request) { handler_grpc(request, buffer) }) do |_client|
+      # Create a gRPC stub using the standard Ruby gRPC client
+      stub = Echo::Echo::Stub.new(
+        "127.0.0.1:3010",
+        :this_channel_is_insecure,
+        channel_args: {
+          'grpc.enable_http_proxy' => 0
+        }
+      )
+      
+      # Create multiple threads to send requests concurrently
+      threads = 5.times.map do |i|
+        Thread.new do
+          request = Echo::EchoRequest.new(message: "Hello GRPC #{i}")
+          response = stub.echo(request)
+          [i, response]
+        end
+      end
+
+      # Collect and verify all responses
+      responses = threads.map(&:value)
+      responses.each do |i, response|
+        assert_instance_of Echo::EchoResponse, response
+        assert_equal "Hello GRPC #{i} response", response.message
+      end
+    end
+  end
+
+  def test_request_type_detection
+    with_server(-> (request) { handler_detect_type(request) }) do |client|
+      # Test regular HTTP request
+      http_response = client.post("/echo", body: "Hello HTTP")
+      assert_equal 200, http_response.status
+      assert_equal "text/plain", http_response.headers["content-type"]
+      assert_equal "HTTP request: Hello HTTP", http_response.body
+
+      # Test gRPC request using the gRPC client
+      stub = Echo::Echo::Stub.new(
+        "127.0.0.1:3010",
+        :this_channel_is_insecure,
+        channel_args: {
+          'grpc.enable_http_proxy' => 0
+        }
+      )
+      
+      request = Echo::EchoRequest.new(message: "Hello gRPC")
+      grpc_response = stub.echo(request)
+      
+      assert_instance_of Echo::EchoResponse, grpc_response
+      assert_equal "gRPC request: Hello gRPC", grpc_response.message
+    end
+  end
+
   def with_server(request_handler, &block)
     server = HyperRuby::Server.new
-    server.configure({ bind_address: "127.0.0.1:3010",tokio_threads: 1 })
+    server.configure({ 
+      bind_address: "127.0.0.1:3010",
+      tokio_threads: 1,
+      #debug: true 
+    })
     server.start
 
     # Create ruby worker threads that process requests;
@@ -185,7 +242,11 @@ def with_server(request_handler, &block)
 
   def with_unix_socket_server(request_handler, &block)
     server = HyperRuby::Server.new
-    server.configure({ bind_address: "unix:/tmp/hyper_ruby_test.sock", tokio_threads: 1 })
+    server.configure({ 
+      bind_address: "unix:/tmp/hyper_ruby_test.sock", 
+      tokio_threads: 1,
+      #debug: true 
+    })
     server.start
 
     # Create ruby worker threads that process requests;
@@ -252,4 +313,24 @@ def handler_grpc(request, buffer)
     # Return gRPC response
     HyperRuby::GrpcResponse.new(0, response_data)
   end
+
+  def handler_detect_type(request)
+    if request.is_a?(HyperRuby::GrpcRequest)
+      # Handle gRPC request
+      buffer = String.new(capacity: 1024)
+      request.fill_body(buffer)
+      echo_request = Echo::EchoRequest.decode(buffer)
+      
+      echo_response = Echo::EchoResponse.new(message: "gRPC request: #{echo_request.message}")
+      response_data = Echo::EchoResponse.encode(echo_response)
+      
+      HyperRuby::GrpcResponse.new(0, response_data)
+    else
+      # Handle regular HTTP request
+      buffer = String.new(capacity: 1024)
+      request.fill_body(buffer)
+      
+      HyperRuby::Response.new(200, { 'Content-Type' => 'text/plain' }, "HTTP request: #{buffer}")
+    end
+  end
 end