Support setting the number of tokio threads.

Author: alistairjevans

ext/hyper_ruby/src/lib.rs

@@ -34,12 +34,14 @@ use http_body_util::Full;
 #[derive(Clone)]
 struct ServerConfig {
     bind_address: String,
+    tokio_threads: Option<usize>,
 }
 
 impl ServerConfig {
     fn new() -> Self {
         Self {
             bind_address: String::from("127.0.0.1:3000"),
+            tokio_threads: None,
         }
     }
 }
@@ -78,6 +80,11 @@ impl Server {
         if let Some(bind_address) = config.get(magnus::Symbol::new("bind_address")) {
             server_config.bind_address = String::try_convert(bind_address)?;
         }
+
+        if let Some(tokio_threads) = config.get(magnus::Symbol::new("tokio_threads")) {
+            server_config.tokio_threads = Some(usize::try_convert(tokio_threads)?);
+        }
+
         Ok(())
     }
 
@@ -163,9 +170,16 @@ impl Server {
             .as_ref()
             .ok_or_else(|| MagnusError::new(magnus::exception::runtime_error(), "Work channel not initialized"))?
             .clone();
- 
-        let rt = Arc::new(tokio::runtime::Builder::new_multi_thread()
-            .enable_all()
+
+        let mut rt_builder = tokio::runtime::Builder::new_multi_thread();
+            
+        rt_builder.enable_all();
+
+        if let Some(tokio_threads) = config.tokio_threads {
+            rt_builder.worker_threads(tokio_threads);
+        }
+
+        let rt = Arc::new(rt_builder
             .build()
             .map_err(|e| MagnusError::new(magnus::exception::runtime_error(), e.to_string()))?);
 
@@ -311,6 +325,7 @@ fn init(ruby: &Ruby) -> Result<(), MagnusError> {
     request_class.define_method("http_method", method!(Request::method, 0))?;
     request_class.define_method("path", method!(Request::path, 0))?;
     request_class.define_method("header", method!(Request::header, 1))?;
+    request_class.define_method("body", method!(Request::body, 0))?;
     request_class.define_method("fill_body", method!(Request::fill_body, 1))?;
     request_class.define_method("body_size", method!(Request::body_size, 0))?;
     request_class.define_method("inspect", method!(Request::inspect, 0))?;

ext/hyper_ruby/src/request.rs

@@ -36,6 +36,12 @@ impl Request {
         self.request.body().len()
     }
 
+    pub fn body(&self) -> RString {        
+        let buffer = RString::buf_new(self.body_size());
+        self.fill_body(buffer);
+        buffer
+    }
+
     pub fn fill_body(&self, buffer: RString) -> usize {
         let body = self.request.body();
         let body_len = body.len();

test/test_hyper_ruby.rb

@@ -70,7 +70,7 @@ def test_blocking
 
   def with_server(request_handler, &block)
     server = HyperRuby::Server.new
-    server.configure({ bind_address: "127.0.0.1:3010" })
+    server.configure({ bind_address: "127.0.0.1:3010",tokio_threads: 1 })
     server.start
 
     # Create ruby worker threads that process requests;
@@ -94,7 +94,7 @@ 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" })
+    server.configure({ bind_address: "unix:/tmp/hyper_ruby_test.sock", tokio_threads: 1 })
     server.start
 
     # Create ruby worker threads that process requests;