Fixed HTTPServer read

Author: colemancda

Sources/HTTP/Server.swift

@@ -16,7 +16,7 @@ public final class HTTPServer {
 
     internal let response: (IPAddress, HTTPRequest) async -> (HTTPResponse)
 
-    internal let log: (String) -> ()
+    internal let log: ((String) -> ())?
 
     internal let socket: Socket
 
@@ -35,19 +35,19 @@ public final class HTTPServer {
         log: ((String) -> ())? = nil,
         response: ((IPAddress, HTTPRequest) async -> HTTPResponse)? = nil
     ) async throws {
+        #if DEBUG
         let log = log ?? {
-            #if DEBUG
-            print("HTTPServer: \($0)")
-            #endif
+            NSLog("HTTPServer: \($0)")
         }
+        #endif
         self.configuration = configuration
+        self.log = log
         self.response = response ?? { (address, request) in
             #if DEBUG
             log("[\(address)] \(request.method) \(request.uri)")
             #endif
             return .init(code: .ok)
         }
-        self.log = log
         // create listening socket
         self.socket = try await Socket(.tcp, bind: IPv4SocketAddress(address: .any, port: configuration.port))
         try socket.fileDescriptor.listen(backlog: configuration.backlog)
@@ -61,24 +61,23 @@ public final class HTTPServer {
         assert(task == nil)
         // listening run loop
         self.task = Task.detached(priority: .high) { [weak self] in
-            self?.log("Started HTTP Server")
+            self?.log?("Started HTTP Server")
             do {
                 while let socket = self?.socket {
                     // wait for incoming sockets
                     let newSocket = try await Socket(fileDescriptor: socket.fileDescriptor.accept())
                     // read remote address
                     let address = try newSocket.fileDescriptor.peerAddress(IPv4SocketAddress.self).address // TODO: Support IPv6
                     if let self = self {
-                        // create connection actor
+                        self.log?("[\(address)] New connection")
                         let connection = await Connection(address: .v4(address), socket: newSocket, server: self)
                         await self.storage.newConnection(connection)
-                        self.log("[\(address)] New connection")
                     }
                 }
             }
             catch _ as CancellationError { }
             catch {
-                self?.log("Error waiting for new connection: \(error)")
+                self?.log?("Error waiting for new connection: \(error)")
             }
         }
     }
@@ -89,7 +88,7 @@ public final class HTTPServer {
         let storage = self.storage
         self.task?.cancel()
         self.task = nil
-        self.log("Stopped GATT Server")
+        self.log?("Stopped GATT Server")
         Task {
             await storage.removeAllConnections()
             await socket.close()
@@ -103,7 +102,7 @@ internal extension HTTPServer {
         // remove connection cache
         await storage.removeConnection(address)
         // log
-        log("[\(address)]: " + "Did disconnect. \(error?.localizedDescription ?? "")")
+        log?("[\(address)]: " + "Did disconnect. \(error?.localizedDescription ?? "")")
     }
 }
 
@@ -125,7 +124,7 @@ public extension HTTPServer {
             port: UInt16 = 8080,
             backlog: Int = 10_000,
             headerMaxSize: Int = 4096,
-            bodyMaxSize: Int = 1024 * 1024 * 5
+            bodyMaxSize: Int = 1024 * 1024 * 10
         ) {
             self.port = port
             self.backlog = backlog
@@ -223,14 +222,22 @@ internal extension HTTPServer {
         }
 
         private func read(_ length: Int) async throws {
-            let data = try await socket.read(length)
-            self.server.log("[\(address)] Read \(data.count) bytes")
-            self.readData.append(data)
+            let chunkSize = 536 // The default TCP Maximum Segment Size is 536
+            var readLength = 0
+            var readMore = true
+            while readMore {
+                let chunk = try await socket.read(chunkSize)
+                readLength += chunk.count
+                self.readData.append(chunk)
+                readMore = readLength < length && chunk.count == chunkSize // need more data and read max
+            }
+            self.server.log?("[\(address)] Read \(readLength) bytes")
         }
 
         private func respond(_ response: HTTPResponse) async throws {
+            let chunkSize = 536 // The default TCP Maximum Segment Size is 536
             let data = response.data
-            self.server.log("[\(address)] Response \(response.code.rawValue) \(response.status) \(response.body.count) bytes")
+            self.server.log?("[\(address)] Response \(response.code.rawValue) \(response.status) \(response.body.count) bytes")
             try await socket.write(data)
             await socket.close()
         }
@@ -276,12 +283,12 @@ internal extension HTTPServer {
                 } else {
                     request.body = Data()
                 }
-                self.server.log("[\(address)] \(request.method) \(request.uri) \(request.body.count) bytes")
+                self.server.log?("[\(address)] \(request.method) \(request.uri) \(request.body.count) bytes")
                 // respond
                 let response = await self.server.response(address, request)
                 try await respond(response)
             } catch {
-                self.server.log("[\(address)] \(error.localizedDescription)")
+                self.server.log?("[\(address)] \(error.localizedDescription)")
                 await self.socket.close()
             }
         }

Tests/HTTPTests/HTTPTests.swift

@@ -146,6 +146,7 @@ final class HTTPTests: XCTestCase {
     func testServerGetBlob() async throws {
         let port = UInt16.random(in: 8080 ..< 9000)
         let blob = Data(repeating: .random(in: .min ... .max), count: 1024 * .random(in: 1 ... 10))
+        print("Blob:", blob.count)
         var server: HTTPServer? = try await HTTPServer(configuration: .init(port: port), response: { (address, request) in
             if request.uri == "/blob", request.method == .get {
                 return request.body.isEmpty ? .init(code: .ok, body: blob) : .init(code: .badRequest)
@@ -169,6 +170,61 @@ final class HTTPTests: XCTestCase {
     func testServerPostBlob() async throws {
         let port = UInt16.random(in: 8080 ..< 9000)
         let blob = Data(repeating: .random(in: .min ... .max), count: 1024 * .random(in: 1 ... 10))
+        print("Blob:", blob.count)
+        var server: HTTPServer? = try await HTTPServer(configuration: .init(port: port), response: { (address, request) in
+            if request.uri == "/blob", request.method == .post {
+                XCTAssertEqual(request.headers[.contentLength], blob.count.description)
+                XCTAssertEqual(request.body.count, blob.count)
+                XCTAssertEqual(request.body, blob)
+                return .init(code: request.body == blob ? .ok : .badRequest)
+            }  else {
+                return .init(code: 404)
+            }
+        })
+        assert(server != nil)
+        let client = URLSession(configuration: .ephemeral)
+        let serverURL  = URL(string: "http://localhost:\(port)")!
+        
+        var request = URLRequest(url: serverURL.appendingPathComponent("blob"))
+        request.httpMethod = HTTPMethod.post.rawValue
+        request.httpBody = blob
+        let (data, urlResponse) = try await client.data(for: request)
+        XCTAssertEqual((urlResponse as! HTTPURLResponse).statusCode, 200)
+        XCTAssert(data.isEmpty)
+        
+        //
+        server = nil
+        try await Task.sleep(nanoseconds: 100_000)
+    }
+    
+    func testServerGetLargeBlob() async throws {
+        let port = UInt16.random(in: 8080 ..< 9000)
+        let blob = Data(repeating: .random(in: .min ... .max), count: 1024 * 1024 * .random(in: 2 ... 4))
+        print("Blob:", blob.count)
+        var server: HTTPServer? = try await HTTPServer(configuration: .init(port: port), response: { (address, request) in
+            if request.uri == "/blob", request.method == .get {
+                return request.body.isEmpty ? .init(code: .ok, body: blob) : .init(code: .badRequest)
+            } else {
+                return .init(code: 404)
+            }
+        })
+        assert(server != nil)
+        let client = URLSession(configuration: .ephemeral)
+        let serverURL  = URL(string: "http://localhost:\(port)")!
+        
+        let (data, urlResponse) = try await client.data(for: URLRequest(url: serverURL.appendingPathComponent("blob")))
+        XCTAssertEqual((urlResponse as! HTTPURLResponse).statusCode, 200)
+        XCTAssertEqual(data, blob)
+        
+        //
+        server = nil
+        try await Task.sleep(nanoseconds: 100_000)
+    }
+    
+    func testServerPostLargeBlob() async throws {
+        let port = UInt16.random(in: 8080 ..< 9000)
+        let blob = Data(repeating: .random(in: .min ... .max), count: 1024 * 1024 * .random(in: 2 ... 4))
+        print("Blob:", blob.count)
         var server: HTTPServer? = try await HTTPServer(configuration: .init(port: port), response: { (address, request) in
             if request.uri == "/blob", request.method == .post {
                 XCTAssertEqual(request.headers[.contentLength], blob.count.description)