feat: big refactor

Author: algebraic-dev

DNS.lean

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

src/Std/Internal/Http.lean

@@ -41,7 +41,7 @@ public structure Connection (α : Type) where
   /--
   The processing machine for HTTP 1.1
   -/
-  machine : Protocol.H1.Machine .response
+  machine : Protocol.H1.Machine .sending
 
 /--
 A request packet to be sent through the persistent connection channel
@@ -124,187 +124,10 @@ end PersistentConnection
 
 namespace Connection
 
-private inductive Recv
-  | bytes (x : Option ByteArray)
-  | timeout
-
-private def receiveWithTimeout [Transport α] (socket : α) (expect : UInt64)
-    (timeoutMs : Millisecond.Offset) :
-  Async Recv := do
-    Selectable.one #[
-      .case (Transport.recvSelector socket expect) (fun x => pure <| .bytes x),
-      .case (← Selector.sleep timeoutMs) (fun _ => pure <| .timeout)]
-
-private def processNeedMoreData
-    [Transport α] (config : Client.Config) (socket : α) (expect : Option Nat) :
-    Async (Except Protocol.H1.Machine.Error (Option ByteArray)) := do
-  try
-    let expect := expect
-      |>.getD config.defaultRequestBufferSize
-      |>.min config.maxRecvChunkSize
-
-    let data ← receiveWithTimeout socket expect.toUInt64 config.readTimeout
-
-    match data with
-    | .bytes (some bytes) => pure (.ok <| some bytes)
-    | .bytes none => pure (.ok <| none)
-    | .timeout => pure (.error Protocol.H1.Machine.Error.timeout)
-
-  catch _ =>
-    pure (.error Protocol.H1.Machine.Error.timeout)
-
-private def handle [Transport α] (connection : Connection α) (config : Client.Config)
-    (requestChannel : CloseableChannel RequestPacket) : Async Unit := do
-
+private def handle [Transport α] (connection : Connection α) (config : Client.Config) (requestChannel : CloseableChannel RequestPacket) : Async Unit := do
   let mut machine := connection.machine
-  let mut running := true
   let socket := connection.socket
 
-  let mut responseStream ← Body.ByteStream.emptyWithCapacity
-  let mut requestTimer ← Interval.mk config.requestTimeout.val config.requestTimeout.property
-  let mut connectionTimer ← Sleep.mk config.keepAliveTimeout
-
-  let mut currentRequest : Option RequestPacket := none
-  let mut receivedResponse := false
-  let mut reqStream := none
-  let mut waitingForRequest := true
-
-  -- Wait for the first tick that is immediate
-  requestTimer.tick
-
-  let mut requestTimerTask ← async requestTimer.tick
-  let connectionTimerTask ← async connectionTimer.wait
-
-  while running do
-
-    if waitingForRequest then
-      match ← await (← requestChannel.recv) with
-      | some packet =>
-        currentRequest := some packet
-        waitingForRequest := false
-
-        machine := machine.sendMessage packet.request.head
-
-        match packet.request.body with
-        | .bytes data => machine := machine.writeUserData #[Chunk.mk data #[]] |>.closeWriter
-        | .zero => machine := machine.closeWriter
-        | .stream stream => do
-          if let some size ← stream.getKnownSize then
-            machine := machine.setKnownSize size
-          reqStream := some stream
-
-        requestTimer.reset
-
-      | none =>
-        running := false
-        continue
-
-    machine := machine.processRead.processWrite
-
-    let (newMachine, events) := machine.takeEvents
-    machine := newMachine
-
-    -- Sends the output of the machine to the socket in a vectored write.
-    if let some (newMachine, data) := machine.takeOutput then
-      machine := newMachine
-
-      if data.size > 0 then
-        try
-          Transport.sendAll socket data.data
-        catch e =>
-          if let some packet := currentRequest then
-            packet.onError e
-          running := false
-
-    for event in events do
-      match event with
-      | .needMoreData expect => do
-        try
-          match ← processNeedMoreData config socket expect with
-          | .ok (some bs) =>
-            machine := machine.feed bs
-          | .ok none =>
-            machine := machine.setFailure .connectionClosed
-          | .error _ => do
-            machine := machine.setFailure .timeout
-        catch e =>
-          if let some packet := currentRequest then
-            packet.onError e
-          running := false
-
-      | .endHeaders head => do
-        if let some (.fixed n) := Protocol.H1.Machine.getMessageSize head then
-          responseStream.setKnownSize (some n)
-
-        receivedResponse := true
-
-        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))
-        pure ()
-
-      | .close =>
-        running := false
-
-      | .next =>
-        -- Request/response cycle complete, ready for next request
-        requestTimer.reset
-        responseStream ← Body.ByteStream.emptyWithCapacity
-        reqStream := none
-        receivedResponse := false
-        currentRequest := none
-        waitingForRequest := true
-
-    -- Sends data from the request body.
-    if let some stream := reqStream then
-      if machine.isWriterOpened then
-        if machine.isReaderComplete ∧ events.isEmpty then
-          if let some data ← stream.recv then
-            machine := machine.writeUserData data
-          else
-            machine := machine.closeWriter
-        else
-          if ← stream.isClosed then
-            pure ()
-          else
-            match ← stream.tryRecv with
-            | some res => machine := machine.writeUserData res
-            | none => machine := machine.closeWriter
-
-    -- Checks for things that can close the connection.
-    if ¬ waitingForRequest then
-      if (← requestTimerTask.isFinished) ∨ (← connectionTimerTask.isFinished) then
-        machine := machine.setFailure .timeout
-        if let some packet := currentRequest then
-          packet.onError (.userError "timeout")
-        running := false
-
-  -- End of the connection
-  connectionTimer.stop
-  requestTimer.stop
-
-  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/Client/Connection.lean

@@ -23,8 +23,6 @@ open Std Internal IO Async
 
 set_option linter.all true
 
-set_option linter.all true
-
 /--
 Type that represents the body of a request or response with streams of bytearrays or bytearrays of fixed
 size.
@@ -91,7 +89,6 @@ protected partial def forIn
 instance : ForIn Async Body Chunk where
   forIn := Body.forIn
 
-
 /--
 Collect all data from the body into a single `ByteArray`. This reads the entire body content into memory,
 so use with caution for large bodies as it may consume significant memory.

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

@@ -194,7 +194,8 @@ partial def write (stream : ByteStream) (data : ByteArray) (extensions : Array (
 
   match result with
   | .inr (success, remaining, exts) =>
-    if remaining.isEmpty then
+    if remaining.isEmpty ∨ ¬ success
+     then
       return success
     else
       write stream remaining exts

src/Std/Internal/Http/Data/Body/ByteStream.lean

@@ -60,7 +60,7 @@ as: ;name=value;name=value Plus 2 bytes for \r\n at the end.
 -/
 def size (chunk : Chunk) : Nat :=
   let extensionsSize := chunk.extensions.foldl (fun acc (name, value) => acc + name.length + (value.get!).length + 2) 0
-  chunk.data.size + extensionsSize + 2
+  chunk.data.size + extensionsSize + (if extensionsSize > 0 then 2 else 0)
 
 instance : Encode .v11 Chunk where
   encode buffer chunk :=

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

@@ -155,7 +155,7 @@ Tries to retrieve the `HeaderValue` for the given key.
 Returns `none` if the header is absent.
 -/
 @[inline]
-def get? (headers : Headers) (name : String) : Option HeaderValue :=
+def get (headers : Headers) (name : String) : Option HeaderValue :=
   headers.data.get? name.toLower
   |>.map (.joinCommaSep ∘ Prod.snd)
 
@@ -184,14 +184,14 @@ Like `get?`, but returns an empty HashSet if absent.
 -/
 @[inline]
 def getD (headers : Headers) (name : String) (d : HeaderValue) : HeaderValue :=
-  headers.get? name |>.getD d
+  headers.get name |>.getD d
 
 /--
 Like `get?`, but panics if absent.
 -/
 @[inline]
 def get! (headers : Headers) (name : String) : HeaderValue :=
-  headers.get? name |>.get!
+  headers.get name |>.get!
 
 /--
 Inserts a new key-value pair into the headers.