<commit_before>import Foundation
import NIO
import NIOHTTP1
import Optics
import Prelude

public func run(
  _ middleware: @escaping Middleware<StatusLineOpen, ResponseEnded, Prelude.Unit, Data>,
  on port: Int = 8080,
  gzip: Bool = false,
  baseUrl: URL
  ) {

  do {
    let reuseAddrOpt = ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR)
    let group = MultiThreadedEventLoopGroup(numberOfThreads: System.coreCount)
    let bootstrap = ServerBootstrap(group: group)
      .serverChannelOption(ChannelOptions.backlog, value: 256)
      .serverChannelOption(reuseAddrOpt, value: 1)
      .childChannelInitializer { channel in
        channel.pipeline.configureHTTPServerPipeline().then {
          let handlers: [ChannelHandler] = gzip
            ? [HTTPResponseCompressor(), Handler(baseUrl: baseUrl, middleware: middleware)]
            : [Handler(baseUrl: baseUrl, middleware: middleware)]
          return channel.pipeline.addHandlers(handlers, first: false)
        }
      }
      .childChannelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
      .childChannelOption(reuseAddrOpt, value: 1)
      .childChannelOption(ChannelOptions.maxMessagesPerRead, value: 1)

    let host = "0.0.0.0"
    let serverChannel = try bootstrap.bind(host: host, port: port).wait()
    print("Listening on \(host):\(port)...")
    try serverChannel.closeFuture.wait()
    try group.syncShutdownGracefully()
  } catch {
    fatalError(error.localizedDescription)
  }
}

private final class Handler: ChannelInboundHandler {
  typealias InboundIn = HTTPServerRequestPart

  let baseUrl: URL
  var request: URLRequest?
  let middleware: Middleware<StatusLineOpen, ResponseEnded, Prelude.Unit, Data>

  init(baseUrl: URL, middleware: @escaping Middleware<StatusLineOpen, ResponseEnded, Prelude.Unit, Data>) {
    self.baseUrl = baseUrl
    self.middleware = middleware
  }

  func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
    let reqPart = self.unwrapInboundIn(data)

    switch reqPart {
    case let .head(header):
      self.request = URL(string: header.uri).map {
        var req = URLRequest(url: $0)
          |> \.httpMethod .~ method(from: header.method)
          |> \.allHTTPHeaderFields .~ header.headers.reduce(into: [:]) { $0[$1.name] = $1.value }
        let proto = req.value(forHTTPHeaderField: "X-Forwarded-Proto") ?? "http"
        req.url = req.value(forHTTPHeaderField: "Host").flatMap {
          URL(string: proto + "://" + $0 + header.uri)
        }
        return req
      }
    case var .body(bodyPart):
      self.request = self.request |> map <<< \.httpBody %~ {
        var data = $0 ?? .init()
        bodyPart.readBytes(length: bodyPart.readableBytes).do { data.append(Data($0)) }
        return data
      }
    case .end:
      guard let req = self.request else {
        _ = ctx.channel.write(HTTPServerResponsePart.head(HTTPResponseHead(
          version: .init(major: 1, minor: 1),
          status: .init(statusCode: 307),
          headers: .init([("location", self.baseUrl.absoluteString)])
        )))
        _ = ctx.channel.writeAndFlush(HTTPServerResponsePart.end(nil)).then {
          ctx.channel.close()
        }
        return
      }

      let promise = ctx.eventLoop.newPromise(of: Conn<ResponseEnded, Data>.self)
      self.middleware(connection(from: req)).parallel.run(promise.succeed(result:))
      _ = promise.futureResult.then { conn -> EventLoopFuture<Void> in
        let res = conn.response

        let head = HTTPResponseHead(
          version: .init(major: 1, minor: 1),
          status: .init(statusCode: res.status.rawValue),
          headers: .init(res.headers.map { ($0.name, $0.value) })
        )
        _ = ctx.channel.write(HTTPServerResponsePart.head(head))

        var buffer = ctx.channel.allocator.buffer(capacity: res.body.count)
        buffer.write(bytes: res.body)
        _ = ctx.channel.write(HTTPServerResponsePart.body(.byteBuffer(buffer)))

        return ctx.channel.writeAndFlush(HTTPServerResponsePart.end(nil)).then {
          ctx.channel.close()
        }
      }
    }
  }
}

private func method(from method: HTTPMethod) -> String {
  switch method {
  case .GET: return "GET"
  case .PUT: return "PUT"
  case .ACL: return "ACL"
  case .HEAD: return "HEAD"
  case .POST: return "POST"
  case .COPY: return "COPY"
  case .LOCK: return "LOCK"
  case .MOVE: return "MOVE"
  case .BIND: return "BIND"
  case .LINK: return "LINK"
  case .PATCH: return "PATCH"
  case .TRACE: return "TRACE"
  case .MKCOL: return "MKCOL"
  case .MERGE: return "MERGE"<commit_msg>Close Channel on Error (#183)<commit_after>import Foundation
import NIO
import NIOHTTP1
import Optics
import Prelude

public func run(
  _ middleware: @escaping Middleware<StatusLineOpen, ResponseEnded, Prelude.Unit, Data>,
  on port: Int = 8080,
  gzip: Bool = false,
  baseUrl: URL
  ) {

  do {
    let reuseAddrOpt = ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR)
    let group = MultiThreadedEventLoopGroup(numberOfThreads: System.coreCount)
    let bootstrap = ServerBootstrap(group: group)
      .serverChannelOption(ChannelOptions.backlog, value: 256)
      .serverChannelOption(reuseAddrOpt, value: 1)
      .childChannelInitializer { channel in
        channel.pipeline.configureHTTPServerPipeline().then {
          let handlers: [ChannelHandler] = gzip
            ? [HTTPResponseCompressor(), Handler(baseUrl: baseUrl, middleware: middleware)]
            : [Handler(baseUrl: baseUrl, middleware: middleware)]
          return channel.pipeline.addHandlers(handlers, first: false)
        }
      }
      .childChannelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
      .childChannelOption(reuseAddrOpt, value: 1)
      .childChannelOption(ChannelOptions.maxMessagesPerRead, value: 1)

    let host = "0.0.0.0"
    let serverChannel = try bootstrap.bind(host: host, port: port).wait()
    print("Listening on \(host):\(port)...")
    try serverChannel.closeFuture.wait()
    try group.syncShutdownGracefully()
  } catch {
    fatalError(error.localizedDescription)
  }
}

private final class Handler: ChannelInboundHandler {
  typealias InboundIn = HTTPServerRequestPart

  let baseUrl: URL
  var request: URLRequest?
  let middleware: Middleware<StatusLineOpen, ResponseEnded, Prelude.Unit, Data>

  init(baseUrl: URL, middleware: @escaping Middleware<StatusLineOpen, ResponseEnded, Prelude.Unit, Data>) {
    self.baseUrl = baseUrl
    self.middleware = middleware
  }

  func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
    let reqPart = self.unwrapInboundIn(data)

    switch reqPart {
    case let .head(header):
      self.request = URL(string: header.uri).map {
        var req = URLRequest(url: $0)
          |> \.httpMethod .~ method(from: header.method)
          |> \.allHTTPHeaderFields .~ header.headers.reduce(into: [:]) { $0[$1.name] = $1.value }
        let proto = req.value(forHTTPHeaderField: "X-Forwarded-Proto") ?? "http"
        req.url = req.value(forHTTPHeaderField: "Host").flatMap {
          URL(string: proto + "://" + $0 + header.uri)
        }
        return req
      }
    case var .body(bodyPart):
      self.request = self.request |> map <<< \.httpBody %~ {
        var data = $0 ?? .init()
        bodyPart.readBytes(length: bodyPart.readableBytes).do { data.append(Data($0)) }
        return data
      }
    case .end:
      guard let req = self.request else {
        ctx.channel.write(
          HTTPServerResponsePart.head(
            HTTPResponseHead(
              version: .init(major: 1, minor: 1),
              status: .init(statusCode: 307),
              headers: .init([("location", self.baseUrl.absoluteString)])
            )
          ),
          promise: nil
        )
        _ = ctx.channel.writeAndFlush(HTTPServerResponsePart.end(nil)).then {
          ctx.channel.close()
        }
        return
      }

      let promise = ctx.eventLoop.newPromise(of: Conn<ResponseEnded, Data>.self)
      self.middleware(connection(from: req)).parallel.run(promise.succeed(result:))
      _ = promise.futureResult.then { conn -> EventLoopFuture<Void> in
        let res = conn.response

        let head = HTTPResponseHead(
          version: .init(major: 1, minor: 1),
          status: .init(statusCode: res.status.rawValue),
          headers: .init(res.headers.map { ($0.name, $0.value) })
        )
        ctx.channel.write(HTTPServerResponsePart.head(head), promise: nil)

        var buffer = ctx.channel.allocator.buffer(capacity: res.body.count)
        buffer.write(bytes: res.body)
        ctx.channel.write(HTTPServerResponsePart.body(.byteBuffer(buffer)), promise: nil)

        return ctx.channel.writeAndFlush(HTTPServerResponsePart.end(nil)).then {
          ctx.channel.close()
        }
      }
    }
  }

  func errorCaught(ctx: ChannelHandlerContext, error: Error) {
    ctx.close(promise: nil)
  }
}

private func method(from method: HTTPMethod) -> String {
  switch method {
  case .GET: return "GET"
  case .PUT: return "PUT"
  case .ACL: return "ACL"
  case .HEAD: return "HEAD"
  case .POST: return "POST"
  case .COPY: return "COPY"
  case .LOCK: return "LOCK"
  case .MOVE: return "MOVE"
  case .BIND: return "BIND"
  case .LINK: return "LINK"
  case .PATCH: return "PATCH"
  case .TRACE: return "TRACE"
  case .MKCOL: return "MKCOL"
  case .MERGE: return "MERGE"
{'fraction_non_alphanumeric': 0.12167340855865447, 'fraction_numerical': 0.0071322120502448376, 'mean_word_length': 2.9607925801011805, 'pattern_counts': {'":': 2, '<': 19, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '792198', 'n_tokens_mistral': 2990, 'n_tokens_neox': 2907, 'n_words': 642}