input-state: use a type to indicate Reqd input state

lib/reqd.ml

@@ -41,6 +41,12 @@ module Response_state = struct
     | Streaming of Response.t * [`write] Body.t
 end
 
+module Input_state = struct
+  type t =
+    | Ready
+    | Complete
+end
+
 type error_handler =
   ?request:Request.t -> error -> (Headers.t -> [`write] Body.t) -> unit
 
@@ -224,8 +230,11 @@ let on_more_output_available t f =
 let persistent_connection t =
   t.persistent
 
-let requires_input { request_body; _ } =
-  not (Body.is_closed request_body)
+let input_state t : Input_state.t =
+  if Body.is_closed t.request_body
+  then Complete
+  else Ready
+;;
 
 let requires_output { response_state; _ } =
   match response_state with
@@ -236,7 +245,10 @@ let requires_output { response_state; _ } =
   | Waiting _ -> true
 
 let is_complete t =
-  not (requires_input t || requires_output t)
+  match input_state t with
+  | Complete -> not (requires_output t)
+  | Ready    -> false
+;;
 
 let flush_request_body t =
   let request_body = request_body t in

lib/server_connection.ml

@@ -215,24 +215,30 @@ let advance_request_queue_if_necessary t =
       wakeup_writer t;
       if Reqd.is_complete reqd
       then shutdown t
-      else if not (Reqd.requires_input reqd)
-      then shutdown_reader t
+      else
+        match Reqd.input_state reqd with
+        | Ready -> ()
+        | Complete -> shutdown_reader t
     end
   end else if Reader.is_closed t.reader
   then shutdown t
 
 let _next_read_operation t =
   advance_request_queue_if_necessary t;
-  if is_active t then begin
+  if is_active t
+  then (
     let reqd = current_reqd_exn t in
-    if      Reqd.requires_input        reqd then Reader.next t.reader
-    else if Reqd.persistent_connection reqd then `Yield
-    else begin
-      shutdown_reader t;
-      Reader.next t.reader
-    end
-  end else
-    Reader.next t.reader
+    match Reqd.input_state reqd with
+    | Ready    -> Reader.next t.reader
+    | Complete ->
+      if Reqd.persistent_connection reqd
+      then `Yield
+      else (
+        shutdown_reader t;
+        Reader.next t.reader)
+  )
+  else Reader.next t.reader
+;;
 
 let next_read_operation t =
   match _next_read_operation t with