Packages

ch

0.2.6
0.8.3 0.8.2 0.8.1 0.8.0 0.7.3 0.7.2 0.7.1 0.7.0 0.6.4 0.6.3 0.6.2 0.6.1 0.6.0 0.5.9 0.5.8 0.5.7 0.5.6 0.5.5 0.5.4 retired 0.5.3 retired 0.5.2 retired 0.5.1 retired 0.5.0 retired 0.4.3 0.4.2 0.4.1 0.4.0 0.3.6 0.3.5 0.3.4 0.3.3 0.3.2 0.3.1 0.3.0 0.2.10 0.2.9 0.2.8 0.2.7 0.2.6 0.2.5 0.2.5-rc.0 0.2.4 0.2.3 0.2.2 0.2.2-rc.3 0.2.2-rc.2 0.2.2-rc.1 0.2.2-rc.0 0.2.1 0.2.0 0.1.15 0.1.14 0.1.13 0.1.12 0.1.11 0.1.10 0.1.9 0.1.8 0.1.7 retired 0.1.6 retired 0.1.5 retired 0.1.4 retired 0.1.3 retired 0.1.2 retired 0.1.1 0.1.0

HTTP ClickHouse driver for Elixir

HexDocs

Current section

Files

Jump to
Readme
66 Versions
5 Dependencies
Dependants
Files
Activity
ch lib ch stream.ex
Raw

lib/ch/stream.ex 

defmodule Ch.Stream do

  @moduledoc false



  @derive {Inspect, only: []}

  defstruct [:conn, :ref, :query, :params, :opts]



  @type t :: %__MODULE__{

          conn: DBConnection.conn(),

          ref: Mint.Types.request_ref() | nil,

          query: Ch.Query.t(),

          params: term,

          opts: [Ch.query_option()]

        }



  defimpl Enumerable do

    def reduce(stream, acc, fun) do

      %Ch.Stream{conn: conn, query: query, params: params, opts: opts} = stream

      stream = %DBConnection.Stream{conn: conn, query: query, params: params, opts: opts}

      DBConnection.reduce(stream, acc, fun)

    end



    def member?(_, _), do: {:error, __MODULE__}

    def count(_), do: {:error, __MODULE__}

    def slice(_), do: {:error, __MODULE__}

  end



  defimpl Collectable do

    def into(stream) do

      %Ch.Stream{conn: conn, query: query, params: params, opts: opts} = stream

      ref = DBConnection.execute!(conn, query, {:stream, params}, opts)

      {%{stream | ref: ref}, &collect/2}

    end



    defp collect(%{conn: conn, query: query, ref: ref} = stream, {:cont, data}) do

      ^ref = DBConnection.execute!(conn, query, {:stream, ref, data})

      stream

    end



    defp collect(%{conn: conn, query: query, ref: ref}, eof) when eof in [:halt, :done] do

      DBConnection.execute!(conn, query, {:stream, ref, :eof})

    end

  end

end