refactor: part of the connection API

Author: algebraic-dev

src/Std/Internal/Http.lean

@@ -7,6 +7,7 @@ module
 
 prelude
 public import Std.Internal.Http.Server
+public import Std.Internal.Http.Client
 
 public section
 

src/Std/Internal/Http/Client/Config.lean

@@ -86,7 +86,7 @@ namespace Config
 /--
 Convert this client config into an HTTP/1.1 protocol configuration.
 -/
-def toH1Config (config : Config) : Protocol.H1.Machine.Config :=
+def toH1Config (config : Config) : Protocol.H1.Config :=
   { maxMessages := config.maxRequestsPerConnection
     maxHeaders := config.maxResponseHeaders
     maxHeaderSize := config.maxHeaderValueSize

src/Std/Internal/Http/Client/Connection.lean

@@ -124,10 +124,148 @@ end PersistentConnection
 
 namespace Connection
 
+private inductive Recv
+  | bytes (x : Option ByteArray)
+  | timeout
+
+private def receiveWithTimeout [Transport α] (socket : α) (expect : UInt64)
+    (timeoutMs : Millisecond.Offset) (req : Timestamp) (all : Timestamp) :
+  Async Recv := do
+    Selectable.one #[
+      .case (Transport.recvSelector socket expect) (fun x => pure <| .bytes x),
+      .case (← Selector.sleep timeoutMs) (fun _ => pure <| .timeout),
+      .case (← Selector.sleep (req - (← Timestamp.now) |>.toMilliseconds)) (fun _ => pure <| .timeout),
+      .case (← Selector.sleep (all - (← Timestamp.now) |>.toMilliseconds)) (fun _ => pure <| .timeout)]
+
+private def processNeedMoreData
+    [Transport α] (config : Config) (socket : α) (expect : Option Nat)
+    (req : Timestamp) (all : Timestamp):
+    Async (Except Protocol.H1.Error (Option ByteArray)) := do
+  try
+    let expect := expect
+      |>.getD config.defaultRequestBufferSize
+      |>.min config.maxRecvChunkSize
+
+    let data ← receiveWithTimeout socket expect.toUInt64 config.readTimeout req all
+
+    match data with
+    | .bytes (some bytes) => pure (.ok <| some bytes)
+    | .bytes none => pure (.ok <| none)
+    | .timeout => pure (.error .timeout)
+
+  catch _ =>
+    pure (.error .timeout)
+
 private def handle [Transport α] (connection : Connection α) (config : Client.Config) (requestChannel : CloseableChannel RequestPacket) : Async Unit := do
   let mut machine := connection.machine
   let socket := connection.socket
 
+  let mut responseStream ← Body.ByteStream.emptyWithCapacity
+  let mut waitingForRequest := true
+  let mut reqStream := none
+  let mut requestTimer := (← Timestamp.now) + config.requestTimeout.val
+  let mut connectionTimer := (← Timestamp.now) + config.keepAliveTimeout.val
+
+  let mut currentRequest := none
+
+  while ¬machine.halted do
+    if waitingForRequest then
+      match ← await (← requestChannel.recv) with
+      | some packet =>
+        currentRequest := some packet
+        waitingForRequest := false
+
+        machine := machine.send packet.request.head
+
+        match packet.request.body with
+        | .bytes data => machine := machine.sendData #[Chunk.mk data #[]] |>.userClosedBody
+        | .zero => machine := machine.userClosedBody
+        | .stream stream =>
+          if let some size ← stream.getKnownSize then
+            machine := machine.setKnownSize size
+
+          reqStream := some stream
+
+      | none =>
+        break
+
+    let (newMachine, step) := machine.step
+    machine := newMachine
+
+    if step.output.size > 0 then
+      try
+        Transport.sendAll socket step.output.data
+      catch e =>
+        if let some packet := currentRequest then
+          packet.onError e
+
+    for event in step.events do
+      match event with
+      | .needMoreData expect => do
+        try
+          match ← processNeedMoreData config socket expect requestTimer connectionTimer with
+          | .ok (some bs) => machine := machine.feed bs
+          | .ok none => machine := machine.noMoreInput
+          | .error _ => machine := machine.closeWriter.closeReader.userClosedBody
+
+        catch e =>
+          if let some packet := currentRequest then
+            packet.onError e
+
+      | .endHeaders head => do
+        if let some (.fixed n) := Protocol.H1.Machine.getMessageSize head then
+          responseStream.setKnownSize (some n)
+
+        if let some packet := currentRequest then
+          let response := { head := machine.reader.messageHead, body := some (.stream responseStream) }
+          packet.onResponse response
+
+      | .gotData final data =>
+        discard <| responseStream.write data.toByteArray
+
+        if final then
+          responseStream.close
+
+      | .chunkExt _ =>
+        pure ()
+
+      | .failed e =>
+        if let some packet := currentRequest then
+          packet.onError (.userError (toString e))
+
+      | .close =>
+        pure ()
+
+      | .next =>
+        requestTimer := (← Timestamp.now) + config.requestTimeout.val
+        responseStream ← Body.ByteStream.emptyWithCapacity
+        reqStream := none
+        currentRequest := none
+        waitingForRequest := true
+
+    if let some stream := reqStream then
+      if ¬machine.writer.isClosed then
+        if machine.isReaderComplete then
+          if let some data ← stream.recv then
+            machine := machine.sendData data
+          else
+            machine := machine.userClosedBody
+        else
+          if ← stream.isClosed then
+            pure ()
+          else
+            match ← stream.tryRecv with
+            | some res => machine := machine.sendData res
+            | none => machine := machine.userClosedBody
+
+  responseStream.close
+  requestChannel.close
+
+  while true do
+    if let some x ← requestChannel.tryRecv then
+      x.onError (.userError "connection closed, cannot send more requests")
+    else
+      break
 
 end Connection
 

src/Std/Internal/Http/Data/Headers.lean

@@ -159,6 +159,15 @@ def get (headers : Headers) (name : String) : Option HeaderValue :=
   headers.data.get? name.toLower
   |>.map (.joinCommaSep ∘ Prod.snd)
 
+/--
+Tries to retrieve the `HeaderValue` for the given key.
+Returns `none` if the header is absent.
+-/
+@[inline]
+def getAll? (headers : Headers) (name : String) : Option (Array HeaderValue) :=
+  headers.data.get? name.toLower
+  |>.map Prod.snd
+
 /--
 Tries to check if the entry is present in the `Headers`
 -/
@@ -265,8 +274,9 @@ def merge (headers1 headers2 : Headers) : Headers :=
 
 instance : ToString Headers where
   toString headers :=
-    let pairs := headers.data.toList.map (fun (_, (k, vs)) => s!"{k}: {HeaderValue.joinCommaSep vs |>.value}")
-    String.intercalate "\r\n" pairs
+    let pairs := headers.data.toArray.flatMap (fun (_, (k, vs)) => vs.map (k, ·))
+    let pairs := pairs.map (fun (k, vs) => s!"{k}: {vs.value}")
+    String.intercalate "\r\n" pairs.toList
 
 instance : Encode .v11 Headers where
   encode buffer := buffer.writeString ∘ toString

src/Std/Internal/Http/Data/URI/Basic.lean

@@ -100,6 +100,11 @@ structure Path where
   absolute : Bool := false
 deriving Inhabited, Repr
 
+instance : ToString Path where
+  toString path :=
+    let result := String.intercalate "/" path.segments.toList
+    if path.absolute then "/" ++ result else result
+
 /--
 Query string represented as an array of key–value pairs.
 -/

src/Std/Internal/Http/Protocol/H1.lean

@@ -220,7 +220,6 @@ private def processHeaders (machine : Machine dir) : Machine dir :=
       | .fixed n => .needFixedBody n
       | .chunked => .needChunkedSize
 
-
 def setHeaders (messageHead : Message.Head dir.swap) (machine : Machine dir) : Machine dir :=
   let hasClose := hasConnectionClose messageHead.headers
 
@@ -285,7 +284,8 @@ def shouldFlush (machine : Machine dir) : Bool :=
   machine.failed ∨
   machine.reader.state == .closed ∨
   machine.writer.userData.size ≥ machine.config.highMark ∨
-  machine.writer.isReadyToSend
+  machine.writer.isReadyToSend ∨
+  machine.writer.knownSize.isSome
 
 @[inline]
 def isReaderComplete (machine : Machine dir) : Bool :=

src/Std/Internal/Http/Server/Connection.lean

@@ -88,7 +88,7 @@ private def handle
   let socket := connection.socket
 
   let mut requestStream ← Body.ByteStream.emptyWithCapacity
-  let mut requestTimer := (← Timestamp.now ) + config.requestTimeout.val
+  let mut requestTimer := (← Timestamp.now) + config.requestTimeout.val
   let mut connectionTimer := (← Timestamp.now) + config.keepAliveTimeout.val
 
   let mut response ← IO.Promise.new