Pony HTTP/1.1 client

client.pony

use "net"
use "net/ssl"
use "collections"

actor HttpClient
  let _env: Env
  let _ssl_ctx: SSLContext
  embed _connections: Map[String, _HttpConnection] = _connections.create()

  new create(env: Env) =>
    _env = env
    _ssl_ctx = recover SSLContext
      .>set_client_verify(false)
      .>set_server_verify(false)
    end

  fun tag request(req: HttpRequest iso, retries: USize = 5): HttpPromise ? =>
    let uri = req.get_uri()?
    let callback = HttpPromise
    _request(_HttpSession(uri, consume req, callback, retries))
    callback

  be _start(hostname: String, conn: TCPConnection) =>
    try _connections(hostname)?.reset(conn) end

  be _redirect(hostname: String, response: HttpResponse val) =>
    try _connections(hostname)?.redirect(response.headers("location")?)? end

  be _finish(hostname: String, response: HttpResponse val) =>
    try
      if not _connections(hostname)?.finish(response) then
        _connections.remove(hostname)?
      end
    end

  be _request(session: _HttpSession iso) =>
    let hostname = session.hostname()
    let http_conn = try _connections(hostname)? else None end
    try
      if http_conn isnt None then
        (http_conn as _HttpConnection).enqueue(consume session)
      else
        _connections(hostname) = _create_http_connection(hostname, consume session)?
      end
    end

  be _reconnect(hostname: String) =>
    try
      let http_conn = _connections(hostname)?
      while http_conn.retry() do
        try
          (let is_ssl, let host, let port) = http_conn.connection_info()
          http_conn.reset(_create_tcp_connection(hostname, is_ssl, host, port)?)
          return
        end
      end
      _connections.remove(hostname)?
    end

  fun _create_tcp_connection(hostname: String, is_ssl: Bool, host: String, port: String): TCPConnection ? =>
    let default_handler = _HttpConnectionHandler(_env, hostname, this)
    let handler = 
      if not is_ssl then consume default_handler
      else SSLConnection(consume default_handler, _ssl_ctx.client()?) end
    TCPConnection(_env.root as AmbientAuth, consume handler, host, port)

  fun _create_http_connection(hostname: String, session: _HttpSession iso): _HttpConnection ? =>
    (let is_ssl, let host, let port) = session.connection_info()
    let tcp_conn = try
      _create_tcp_connection(hostname, is_ssl, consume host, consume port)?
    else None end

    if tcp_conn isnt None then
      _HttpConnection(_env, (tcp_conn as TCPConnection), consume session)
    else
      session.callback.reject()
      error
    end

class _HttpSession
  var _uri: URI val
  var retries: USize = 0
  let callback: HttpPromise
  let request: HttpRequest val

  new iso create(uri: URI val, req: HttpRequest iso, resp: HttpPromise, num_retries: USize) =>
    _uri = uri
    retries = num_retries
    request = consume req
    callback = consume resp

  fun ref redirect(url: String) ? =>
    _uri = recover val URI(consume url)? end

  fun hostname(): String =>
    request.hostname(_uri)

  fun string(): String =>
    recover val request.string(_uri) end

  fun connection_info(): (Bool, String, String) =>
    (_uri.is_ssl, _uri.host.string(), _uri.port.string())

class _HttpConnection
  let _env: Env
  var _tcp_conn: TCPConnection
  var _session: _HttpSession iso
  embed _queue: Deque[_HttpSession iso] = _queue.create()

  new create(env: Env, conn: TCPConnection, session: _HttpSession iso) =>
    _env = env
    _tcp_conn = conn
    _session = consume session

  fun ref connection_info(): (Bool, String, String) =>
    _session.connection_info()

  fun ref redirect(url: String) ? =>
    _session.redirect(consume url)?
    reset(_tcp_conn)

  fun ref enqueue(session: _HttpSession iso) =>
    _queue.push(consume session)

  fun ref _next_session(): Bool =>
    try _session = _queue.pop()?; true
    else false end

  fun ref reset(new_tcp_conn: TCPConnection) =>
    _tcp_conn = new_tcp_conn
    _tcp_conn.write(_session.string())

  fun ref finish(response: HttpResponse val): Bool =>
    _session.callback(response)
    if _next_session() then _tcp_conn.write(_session.string()); true
    else false end

  fun ref retry(): Bool =>
    if _session.retries isnt 0 then
      _session.retries = _session.retries - 1
      true
    else
      _next_session()
    end
    
class _HttpConnectionHandler is TCPConnectionNotify
  let _env: Env
  let _hostname: String
  let _client: HttpClient
  embed _parser: HttpParser

  new iso create(env: Env, hostname: String, client: HttpClient) =>
    _env = env
    _client = client
    _hostname = hostname
    _parser = HttpParser(env)

  fun ref connected(conn: TCPConnection ref) =>
    _client._start(_hostname, conn)

  fun ref closed(conn: TCPConnection ref) =>
    _client._reconnect(_hostname)

  fun ref connect_failed(conn: TCPConnection ref) => 
    closed(consume conn)

  fun ref received(conn: TCPConnection ref, data: Array[U8] iso, times: USize): Bool =>
    try
      let response = _parser.feed(consume data)?
      if (response.status is 301) and response.headers.contains("location") then
        _client._redirect(_hostname, consume response)
      else
        _client._finish(_hostname, consume response)
      end
    end
    true

data.pony

use "json"
use "promises"
use "collections"

type Json is JsonDoc
type HttpHeaders is Map[String, String]
type HttpContent is (Array[U8] | JsonDoc)
type HttpPromise is Promise[HttpResponse val]

class HttpResponse
  let status: U16
  let reason: String
  let headers: HttpHeaders iso
  let content: Maybe[HttpContent val]

  new create(s: U16, r: String, h: HttpHeaders iso, c: Maybe[HttpContent val]) =>
    status = s
    reason = consume r
    headers = consume h
    content = consume c

class HttpRequest
  let _method: String
  var _uri: Maybe[URI val] = None
  var _content: Maybe[HttpContent] = None
  embed _headers: HttpHeaders = _headers.create()

  new iso create(method: String, url: String) ? =>
    _method = consume method
    _uri = recover val URI(consume url)? end

  fun ref get_uri(): URI val^ ? =>
    (_uri = None) as URI val^

  fun apply(header: String): this->String ? =>
    _headers(header)?

  fun ref update(header: String, value: String) =>
    _headers(header) = value

  fun ref content(data: HttpContent) =>
    _content = consume data

  fun hostname(uri: URI val): String =>
    let port = uri.port.string()
    let host_size = uri.host.size() + port.size() + 1
    recover String(host_size)
      .>append(uri.host).>push(':').>append(consume port)
    end

  fun string(uri: URI val): String ref^ =>
    let host = hostname(uri)
    let size = _method.size() + host.size() + 16
              + uri.path.size() + uri.query.size()

    var output = String(size)
      .>append(_method).>push(' ')
      .>append(uri.path).>append(uri.query)
      .>append(" HTTP/1.1\r\nHost: ").>append(host).>append("\r\n")

    for (key, value) in _headers.pairs() do
      if key isnt "Host" then
        output.>append(key).>append(": ").>append(value).>append("\r\n")
      end
    end

    output.append("\r\n")
    match _content
    | let data: this->Array[U8] => output.append(data)
    | let json: this->JsonDoc => output.append(json.string())
    end

    consume output
    

    

parser.pony

use "buffered"

type _ParserState is (_WantLine | _WantHeaders | _WantContent | _WantChunkHead | _WantChunkData)

primitive _WantLine
primitive _WantHeaders
primitive _WantContent
primitive _WantChunkHead
primitive _WantChunkData

class HttpParser
  let _env: Env
  var _status: U16 = 0
  var _reason: String = ""
  var _block_size: USize = 0
  var _state: _ParserState = _WantLine
  var _headers: Maybe[HttpHeaders iso] = None
  embed _buffer: Reader = _buffer.create()
  embed _content: Reader = _content.create()

  new create(env: Env) =>
    _env = env

  fun ref feed(data: Array[U8] iso): HttpResponse val^ ? =>
    _buffer.append(consume data)
    match _state
    | _WantLine => _parse_line()?
    | _WantHeaders => _parse_headers()?
    | _WantContent => _parse_content()?
    | _WantChunkHead => _parse_chunk_head()?
    | _WantChunkData => _parse_chunk_data()?
    end

  fun ref _parse_line(): HttpResponse val^ ? =>
    let line = _buffer.line()?
    let delimiter = line.find(" ", 9)?
    _reason = line.substring(delimiter + 1)
    _status = line.substring(9, delimiter).u16()?
    _headers = recover HttpHeaders end
    _state = _WantHeaders
    _parse_headers()?

  fun ref _parse_headers(): HttpResponse val^ ? =>
    let line = _buffer.line()?
    if line.size() > 0 then
      let delimiter = line.find(":")?
      let key = recover val line.substring(0, delimiter).>lower_in_place() end
      let value = recover val line.substring(delimiter + 2) end
      (_headers as HttpHeaders iso)(key) = consume value
      _parse_headers()?
    else
      _state = _next_header_state()
      match _state
      | _WantLine => _finalize()?
      | _WantContent => _parse_content()?
      | _WantChunkHead => _parse_chunk_head()?
      else
        error
      end
    end

  fun ref _parse_content(): HttpResponse val^ ? =>
    _content.append(_buffer.block(_block_size)?)
    _finalize()?

  fun ref _parse_chunk_head(): HttpResponse val^ ? =>
    _block_size = _buffer.line()?.u16(16)?.usize()
    _state = _WantChunkData
    _parse_chunk_data()?

  fun ref _parse_chunk_data(): HttpResponse val^ ? =>
    if _block_size is 0 then
      _buffer.block(2)?
      _finalize()?
    else
      _content.append(_buffer.block(_block_size + 2)?.>truncate(_block_size))
      _state = _WantChunkHead
      _parse_chunk_head()?
    end

  fun ref _next_header_state(): _ParserState =>
    try
      _block_size = (_headers as HttpHeaders iso)("content-length")?.usize()?
      _WantContent
    else
      try
        if (_headers as HttpHeaders iso)("transfer-encoding")?.contains("chunked")
        then _WantChunkHead else error end
      else
        _WantLine
      end
    end

  fun ref _finalize(): HttpResponse val^ ? =>
    _state = _WantLine
    let data = _content.block(_content.size())?
    let is_json = try 
      (_headers as HttpHeaders iso)("content-type")?.contains("application/json")
    else false end

    let reason = (_reason = "")
    let headers = (_headers = None) as HttpHeaders iso^
    let content = recover val
      if data.size() is 0 then None
      elseif is_json then Json.>parse(String.from_iso_array(consume data))?
      else consume data end
    end

    recover val
      HttpResponse(_status, consume reason, consume headers, consume content)
    end

uri.pony

class URI
  let proto: String
  let host: String
  let port: U16
  let path: String
  let query: String
  let is_ssl: Bool

  new create(url: String) ? =>
    let u_end   = url.size().isize()
    let p_end   = url.find("://")?
    let h_start = p_end + 3
    let p_start = try url.find("/", h_start)? else u_end end
    let q_start = try url.find("?", p_start)? else u_end end
    let h_end   = try url.find(":", h_start)? else p_start end

    proto = url.substring(0, p_end)
    host = url.substring(h_start, h_end)
    path = url.substring(p_start, q_start)
    query = url.substring(q_start, u_end)
    is_ssl = proto(proto.size() - 1)? == 's'

    port =
      if h_end == u_end
      then url.substring(h_end, p_start).u16()?
      else if is_ssl then 443 else 80 end end

utils.pony

type Maybe[T] is (T | None)

class _DequeNode[T]
  var value: Maybe[T] 
  var next: Maybe[_DequeNode[T]] = None
  
  new create(item: T) =>
    value = consume item
  
  fun ref pop(): T^ ? =>
    (value = None) as T^

class Deque[T]
  var _head: Maybe[_DequeNode[T]] = None
  var _tail: Maybe[_DequeNode[T]] = None

  fun ref pop(): T^ ? =>
    try
      let head = _head as _DequeNode[T]
      _head = head.next
      head.pop()?
    else
      error
    end

  fun ref push(item: T) =>
    let node = _DequeNode[T](consume item)
    try
      if _head is None then
        _head = consume node
      elseif _tail is None then
        _tail = consume node
        (_head as _DequeNode[T]).next = _tail
      else
        (_tail as _DequeNode[T]).next = consume node
        _tail = (_tail as _DequeNode[T]).next
      end
    end