<commit_before>public struct Server {
    public let tcpHost: Host
    public let middleware: [Middleware]
    public let responder: Responder
    public let failure: (Error) -> Void

    public let host: String
    public let port: Int
    public let bufferSize: Int

    public init(host: String = "0.0.0.0", port: Int = 8080, middleware: [Middleware] = [], responder: Responder, failure: @escaping (Error) -> Void =  Server.log(error:)) throws {
        let backlog = 128
        let reusePort = false

        let bufferSize = 2048
        let enableLog = true
        let enableSession = true
        let enableContentNegotiation = true

        self.tcpHost = try TCPHost(
            configuration: [
                "host": Map(host),
                "port": Map(port),
                "backlog": Map(backlog),
                "reusePort": Map(reusePort),
                ]
        )

        var chain: [Middleware] = []

        if enableLog {
            chain.append(LogMiddleware())
        }

        if enableSession {
            chain.append(SessionMiddleware())
        }

        if enableContentNegotiation {
            chain.append(ServerContentNegotiationMiddleware(mediaTypes: [JSON.self, URLEncodedForm.self]))
        }

        chain.append(contentsOf: middleware)

        self.host = host
        self.port = port
        self.bufferSize = bufferSize
        self.middleware = chain
        self.responder = responder
        self.failure = failure
    }

    public init(configuration: Map, middleware: [Middleware], responder: Responder, failure: @escaping (Error) -> Void =  Server.log(error:)) throws {
        let host = configuration["tcp", "host"].string ?? "0.0.0.0"
        let port = configuration["tcp", "port"].int ?? 8080
        let backlog = configuration["tcp", "backlog"].int ?? 128
        let reusePort = configuration["tcp", "reusePort"].bool ?? false

        let certificate = configuration["tls", "certificate"].string
        let privateKey = configuration["tls", "privateKey"].string
        let certificateChain = configuration["tls", "certificateChain"].string

        let bufferSize = configuration["bufferSize"].int ?? 2048
        let enableLog = configuration["log"].bool ?? true
        let enableSession = configuration["session"].bool ?? true
        let enableContentNegotiation = configuration["contentNegotiation"].bool ?? true

        if let c = certificate, let pk = privateKey {
            self.tcpHost = try TCPTLSHost(
                configuration: [
                    "host": Map(host),
                    "port": Map(port),
                    "backlog": Map(backlog),
                    "reusePort": Map(reusePort),

                    "certificate": Map(c),
                    "privateKey": Map(pk),
                    "certificateChain": Map(certificateChain),
                ]
            )
        } else {
            self.tcpHost = try TCPHost(
                configuration: [
                    "host": Map(host),
                    "port": Map(port),
                    "backlog": Map(backlog),
                    "reusePort": Map(reusePort),
                ]
            )
        }

        var chain: [Middleware] = []

        if enableLog {
            chain.append(LogMiddleware())
        }

        if enableSession {
            chain.append(SessionMiddleware())
        }

        if enableContentNegotiation {
            chain.append(ServerContentNegotiationMiddleware(mediaTypes: [JSON.self, URLEncodedForm.self]))
        }

        chain.append(contentsOf: middleware)

        self.host = host
        self.port = port
        self.bufferSize = bufferSize
        self.middleware = chain
        self.responder = responder
        self.failure = failure
    }

    public init(configuration: Map, middleware: [Middleware] = [], responder representable: ResponderRepresentable, failure: @escaping (Error) -> Void = Server.log(error:)) throws {
        try self.init(
            configuration: configuration,
            middleware: middleware,
            responder: representable.responder,
            failure: failure
        )
    }
}

func retry(times: Int, waiting duration: Double, work: (Void) throws -> Void) throws {
    var failCount = 0
    var lastError: Error!
    while failCount < times {
        do {
            try work()
        } catch {
            failCount += 1
            lastError = error
            print("Error: \(error)")
            print("Retrying in \(duration) seconds.")
            nap(for: duration)
            print("Retrying.")
        }
    }
    throw lastError
}

extension Server {
    public func start() throws {
        printHeader()
        try retry(times: 10, waiting: 5.seconds) {
            while true {
                let stream = try tcpHost.accept(deadline: .never)
                co { do { try self.process(stream: stream) } catch { self.failure(error) } }
            }
        }
    }

    public func startInBackground() {
        co { do { try self.start() } catch { self.failure(error) } }
    }

    public func process(stream: Stream) throws {
        let buffer = UnsafeMutableBufferPointer<Byte>(capacity: bufferSize)
        defer { buffer.deallocate(capacity: bufferSize) }

        let parser = MessageParser(mode: .request)
        let serializer = ResponseSerializer(stream: stream, bufferSize: bufferSize)

        while !stream.closed {
            do {
                // TODO: Add timeout parameter
                let bytesRead = try stream.read(into: buffer, deadline: 30.seconds.fromNow())
                
                for message in try parser.parse(bytesRead) {
                    let request = message as! Request
                    let response = try middleware.chain(to: responder).respond(to: request)
                    // TODO: Add timeout parameter
                    try serializer.serialize(response, deadline: 5.minutes.fromNow())
                    
                    if let upgrade = response.upgradeConnection {
                        try upgrade(request, stream)
                        stream.close()
                    }
                    
                    if !request.isKeepAlive {
                        stream.close()
                    }
                }
            } catch SystemError.brokenPipe {
                break
            } catch {
                if stream.closed {
                    break
                }
                
                let (response, unrecoveredError) = Server.recover(error: error)
                try serializer.serialize(response, deadline: .never)

                if let error = unrecoveredError {
                    throw error
                }
            }
        }
    }

    private static func recover(error: Error) -> (Response, Error?) {
        guard let representable = error as? ResponseRepresentable else {
            let body = Buffer(String(describing: error))
            return (Response(status: .internalServerError, body: body), error)
        }
        return (representable.response, nil)
    }

    public static func log(error: Error) -> Void {
        print("Error: \(error)")
    }

    public func printHeader() {
        var header = "\n"
        header += "\n"
        header += "\n"
        header += "                             _____\n"
        header += "     ,.-``-._.-``-.,        /__  /  ___ _      ______\n"
        header += "    |`-._,.-`-.,_.-`|         / /  / _ \\ | /| / / __ \\\n"
        header += "    |   |ˆ-. .-`|   |        / /__/  __/ |/ |/ / /_/ /\n"
        header += "    `-.,|   |   |,.-`       /____/\\___/|__/|__/\\____/ (c)\n"
        header += "        `-.,|,.-`           -----------------------------\n"
        header += "\n"
        header += "================================================================================\n"
        header += "Started HTTP server at \(host), listening on port \(port)."
        print(header)
    }
}
<commit_msg>Improve content negotiation middleware (#172)<commit_after>public struct Server {
    public let tcpHost: Host
    public let middleware: [Middleware]
    public let responder: Responder
    public let failure: (Error) -> Void

    public let host: String
    public let port: Int
    public let bufferSize: Int

    public init(host: String = "0.0.0.0", port: Int = 8080, backlog: Int = 128, reusePort: Bool = false, bufferSize: Int = 4096, middleware: [Middleware] = [], responder: ResponderRepresentable, failure: @escaping (Error) -> Void =  Server.log(error:)) throws {
        self.tcpHost = try TCPHost(
            host: host,
            port: port,
            backlog: backlog,
            reusePort: reusePort
        )
        self.host = host
        self.port = port
        self.bufferSize = bufferSize
        self.middleware = middleware
        self.responder = responder.responder
        self.failure = failure
    }

    public init(host: String = "0.0.0.0", port: Int = 8080, backlog: Int = 128, reusePort: Bool = false, bufferSize: Int = 4096, certificatePath: String, privateKeyPath: String, certificateChainPath: String? = nil, middleware: [Middleware] = [], responder: ResponderRepresentable, failure: @escaping (Error) -> Void =  Server.log(error:)) throws {
        self.tcpHost = try TCPTLSHost(
            host: host,
            port: port,
            backlog: backlog,
            reusePort: reusePort,
            certificatePath: certificatePath,
            privateKeyPath: privateKeyPath,
            certificateChainPath: certificateChainPath
        )
        self.host = host
        self.port = port
        self.bufferSize = bufferSize
        self.middleware = middleware
        self.responder = responder.responder
        self.failure = failure
    }
}

func retry(times: Int, waiting duration: Double, work: (Void) throws -> Void) throws {
    var failCount = 0
    var lastError: Error!
    while failCount < times {
        do {
            try work()
        } catch {
            failCount += 1
            lastError = error
            print("Error: \(error)")
            print("Retrying in \(duration) seconds.")
            nap(for: duration)
            print("Retrying.")
        }
    }
    throw lastError
}

extension Server {
    public func start() throws {
        printHeader()
        try retry(times: 10, waiting: 5.seconds) {
            while true {
                let stream = try tcpHost.accept(deadline: .never)
                co { do { try self.process(stream: stream) } catch { self.failure(error) } }
            }
        }
    }

    public func startInBackground() {
        co { do { try self.start() } catch { self.failure(error) } }
    }

    public func process(stream: Stream) throws {
        let buffer = UnsafeMutableBufferPointer<Byte>(capacity: bufferSize)
        defer { buffer.deallocate(capacity: bufferSize) }

        let parser = MessageParser(mode: .request)
        let serializer = ResponseSerializer(stream: stream, bufferSize: bufferSize)

        while !stream.closed {
            do {
                // TODO: Add timeout parameter
                let bytesRead = try stream.read(into: buffer, deadline: 30.seconds.fromNow())
                
                for message in try parser.parse(bytesRead) {
                    let request = message as! Request
                    let response = try middleware.chain(to: responder).respond(to: request)
                    // TODO: Add timeout parameter
                    try serializer.serialize(response, deadline: 5.minutes.fromNow())
                    
                    if let upgrade = response.upgradeConnection {
                        try upgrade(request, stream)
                        stream.close()
                    }
                    
                    if !request.isKeepAlive {
                        stream.close()
                    }
                }
            } catch SystemError.brokenPipe {
                break
            } catch {
                if stream.closed {
                    break
                }
                
                let (response, unrecoveredError) = Server.recover(error: error)
                try serializer.serialize(response, deadline: .never)

                if let error = unrecoveredError {
                    throw error
                }
            }
        }
    }

    private static func recover(error: Error) -> (Response, Error?) {
        guard let representable = error as? ResponseRepresentable else {
            let body = Buffer(String(describing: error))
            return (Response(status: .internalServerError, body: body), error)
        }
        return (representable.response, nil)
    }

    public static func log(error: Error) -> Void {
        print("Error: \(error)")
    }

    public func printHeader() {
        var header = "\n"
        header += "\n"
        header += "\n"
        header += "                             _____\n"
        header += "     ,.-``-._.-``-.,        /__  /  ___ _      ______\n"
        header += "    |`-._,.-`-.,_.-`|         / /  / _ \\ | /| / / __ \\\n"
        header += "    |   |ˆ-. .-`|   |        / /__/  __/ |/ |/ / /_/ /\n"
        header += "    `-.,|   |   |,.-`       /____/\\___/|__/|__/\\____/ (c)\n"
        header += "        `-.,|,.-`           -----------------------------\n"
        header += "\n"
        header += "================================================================================\n"
        header += "Started HTTP server at \(host), listening on port \(port)."
        print(header)
    }
}

{'fraction_non_alphanumeric': 0.12595419847328243, 'fraction_numerical': 0.005798590722254845, 'mean_word_length': 1.5659133709981168, 'pattern_counts': {'":': 15, '<': 7, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5201258', 'n_tokens_mistral': 3673, 'n_tokens_neox': 3459, 'n_words': 885}