kadabra

defmodule Kadabra.Stream do
  @moduledoc false

  defstruct [:id, :uri, :connection, :encoder, :decoder, :settings,
             :socket, :flow_control, headers: [], body: "", scheme: :https]

  alias Kadabra.{Connection, Encodable, Hpack, Http2, Stream}
  alias Kadabra.Frame.{Continuation, Data, Headers, PushPromise, RstStream}

  @type t :: %__MODULE__{
    id: pos_integer,
    uri: charlist,
    connection: pid,
    encoder: pid,
    decoder: pid,
    settings: pid,
    socket: pid,
    flow_control: pid,
    headers: [...],
    body: binary,
    scheme: :https
  }

  @data 0x0
  # @headers 0x1

  @closed :closed
  @half_closed_local :half_closed_local
  @half_closed_remote :half_closed_remote
  @idle :idle
  @open :open
  # @reserved_local :reserved_local
  @reserved_remote :reserved_remote

  def new(%Connection{} = conn) do
    %__MODULE__{
      id: conn.stream_id,
      uri: conn.uri,
      connection: self(),
      socket: conn.socket,
      settings: conn.settings,
      encoder: conn.encoder_state,
      decoder: conn.decoder_state,
      flow_control: conn.flow_control
    }
  end
  def new(%Connection{} = conn, stream_id) do
    conn
    |> new()
    |> Map.put(:id, stream_id)
  end

  def start_link(stream) do
    :gen_statem.start_link(__MODULE__, stream, [])
  end

  def cast_recv(pid, frame) do
    :gen_statem.cast(pid, {:recv, frame})
  end

  def cast_send(pid, frame) do
    :gen_statem.cast(pid, {:send, frame})
  end

  def recv(%Data{end_stream: end_stream?, data: data}, _state, stream) do
    stream = %Stream{stream | body: stream.body <> data}

    cond do
      end_stream? -> {:next_state, @half_closed_remote, stream}
      true -> {:keep_state, stream}
    end
  end

  def recv(%Headers{header_block_fragment: fragment} = frame, _state, stream) do
    {:ok, headers} = Hpack.decode(stream.decoder, fragment)
    stream = %Stream{stream | headers: stream.headers ++ headers}

    if frame.end_stream do
      {:next_state, @half_closed_remote, stream}
    else
      {:keep_state, stream}
    end
  end

  def recv(%PushPromise{header_block_fragment: fragment}, state, stream)
    when state in [@idle] do

    {:ok, headers} = Hpack.decode(stream.decoder, fragment)
    stream = %Stream{stream | headers: stream.headers ++ headers}

    send(stream.connection, {:push_promise, Stream.Response.new(stream)})
    {:next_state, @reserved_remote, stream}
  end

  def recv(%RstStream{}, state, stream)
    when state in [@open, @half_closed_local, @half_closed_remote, @closed] do
    {:next_state, :closed, stream}
  end

  def recv(%Continuation{header_block_fragment: fragment}, state, stream)
    when state in [@idle] do

    {:ok, headers} = Hpack.decode(stream.decoder, fragment)
    stream = %Stream{stream | headers: stream.headers ++ headers}

    {:keep_state, stream}
  end

  def recv(frame, state, stream) do
    IO.puts("Unknown RECV: #{inspect(frame)}, #{inspect(state)},
    #{inspect(stream)}")
    {:keep_state, stream}
  end

  # Enter Events

  def handle_event(:enter, _old, @half_closed_remote, stream) do
    bin = stream.id |> RstStream.new |> Encodable.to_bin
    :ssl.send(stream.socket, bin)

    :gen_statem.cast(self(), :close)
    {:keep_state, stream}
  end
  def handle_event(:enter, _old, @closed, stream) do
    send(stream.connection, {:finished, Stream.Response.new(stream)})
    {:stop, :normal}
  end
  def handle_event(:enter, _old, _new, stream), do: {:keep_state, stream}

  # Casts

  def handle_event(:cast, :close, _state, stream) do
    {:next_state, @closed, stream}
  end

  def handle_event(:cast, {:recv, frame}, state, stream) do
    recv(frame, state, stream)
  end

  def handle_event(:cast, {:send_headers, _h, _p}, _state, stream) do
    # headers = add_headers(headers, stream)

    # {:ok, encoded} = Hpack.encode(stream.encoder, headers)
    # headers_payload = :erlang.iolist_to_binary(encoded)

    # h = Http2.build_frame(@headers, 0x4, stream.id, headers_payload)
    # :ssl.send(stream.socket, h)
    # IO.puts("Sending, Stream ID: #{stream.id}")

    # #send(stream.connection, {:registered, stream.id, self()})

    # if payload do
    #   {:ok, settings} = Kadabra.ConnectionSettings.fetch(stream.settings)
    #   chunks = chunk(settings.max_frame_size, payload)
    #   send_chunks(stream.socket, stream.id, chunks)
    # end

    {:next_state, @open, stream}
  end

  def handle_event(:cast, msg, state, stream) do
    IO.inspect("""
    === Unknown cast ===
    #{inspect(msg)}
    State: #{inspect(state)}
    Stream: #{inspect(stream)}
    """)
    {:keep_state, stream}
  end

  def send_chunks(_socket, _stream_id, []), do: :ok
  def send_chunks(socket, stream_id, [chunk | []]) do
    h_p = Http2.build_frame(@data, 0x1, stream_id, chunk)
    :ssl.send(socket, h_p)
  end
  def send_chunks(socket, stream_id, [chunk | rest]) do
    h_p = Http2.build_frame(@data, 0x0, stream_id, chunk)
    :ssl.send(socket, h_p)

    send_chunks(socket, stream_id, rest)
  end

  def chunk(size, bin) when byte_size(bin) >= size do
    {chunk, rest} = :erlang.split_binary(bin, size)
    [chunk | chunk(size, rest)]
  end
  def chunk(_size, <<>>), do: []
  def chunk(_size, bin), do: [bin]

  def add_headers(headers, stream) do
    h = headers ++
    [
      {":scheme", Atom.to_string(stream.scheme)},
      {":authority", List.to_string(stream.uri)}
    ]
    # sorting headers to have pseudo headers first.
    Enum.sort(h, fn({a, _b}, {c, _d}) -> a < c end)
  end

  def init(stream), do: {:ok, @idle, stream}

  def callback_mode, do: [:handle_event_function, :state_enter]

  def terminate(_reason, _state, _data), do: :void

  def code_change(_vsn, state, data, _extra), do: {:ok, state, data}
end