Packages

tesla

1.14.0
1.20.0 1.18.3 1.18.2 1.18.1 1.18.0 1.17.0 1.16.0 1.15.3 1.15.2 1.15.1 1.15.0 1.14.3 1.14.2 1.14.1 1.14.0 1.13.2 1.13.1 1.13.0 1.12.3 1.12.2 1.12.1 1.12.0 1.11.2 1.11.1 1.11.0 1.10.3 1.10.2 1.10.1 1.10.0 1.9.0 1.8.1 retired 1.8.0 1.7.0 1.6.1 1.6.0 1.5.1 1.5.0 1.4.4 1.4.3 1.4.2 1.4.1 1.4.0 1.3.3 1.3.2 1.3.1 1.3.0 1.2.1 1.2.0 1.1.0 1.0.0 1.0.0-beta.1 0.10.0 0.9.0 0.8.0 0.7.2 0.7.1 0.7.0 0.6.0 0.5.2 0.5.1 0.5.0 0.3.6 0.3.5 0.3.4 0.3.3 0.3.2 0.3.1 0.2.2 0.2.1 0.1.5 0.1.4 0.1.3 0.1.2 0.1.1 0.1.0

HTTP client library, with support for middleware and multiple adapters.

Security advisory: This version has known vulnerabilities. View advisories
HexDocs

Current section

Files

Jump to
Readme
75 Versions
14 Dependencies
512 Dependants
5 Advisories
Files
Activity
tesla lib tesla middleware sse.ex
Raw

lib/tesla/middleware/sse.ex 

defmodule Tesla.Middleware.SSE do

  @moduledoc """

  Decode Server Sent Events.



  This middleware is mostly useful when streaming response body.



  ## Examples



  ```elixir

  defmodule MyClient do

    def client do

      Tesla.client([Tesla.Middleware.SSE, only: :data])

    end

  end

  ```



  ## Options



  - `:only` - keep only specified keys in event (necessary for using with `JSON` middleware)

  - `:decode_content_types` - list of additional decodable content-types

  """



  @behaviour Tesla.Middleware



  @default_content_types ["text/event-stream"]



  @impl Tesla.Middleware

  def call(env, next, opts) do

    opts = opts || []



    with {:ok, env} <- Tesla.run(env, next) do

      decode(env, opts)

    end

  end



  def decode(env, opts) do

    if decodable_content_type?(env, opts) do

      {:ok, %{env | body: decode_body(env.body, opts)}}

    else

      {:ok, env}

    end

  end



  defp decode_body(body, opts) when is_struct(body, Stream) or is_function(body) do

    body

    |> Stream.chunk_while(

      "",

      fn elem, acc ->

        {lines, [rest]} = (acc <> elem) |> String.split("\n\n") |> Enum.split(-1)

        {:cont, lines, rest}

      end,

      fn

        "" -> {:cont, ""}

        acc -> {:cont, acc, ""}

      end

    )

    |> Stream.flat_map(& &1)

    |> Stream.map(&decode_message/1)

    |> Stream.flat_map(&only(&1, opts[:only]))

  end



  defp decode_body(binary, opts) when is_binary(binary) do

    binary

    |> String.split("\n\n")

    |> Enum.map(&decode_message/1)

    |> Enum.flat_map(&only(&1, opts[:only]))

  end



  defp decode_message(message) do

    message

    |> String.split("\n")

    |> Enum.map(&decode_body/1)

    |> Enum.reduce(%{}, fn

      :empty, acc -> acc

      {:data, data}, acc -> Map.update(acc, :data, data, &(&1 <> "\n" <> data))

      {key, value}, acc -> Map.put_new(acc, key, value)

    end)

  end



  defp decode_body(": " <> comment), do: {:comment, comment}

  defp decode_body("data: " <> data), do: {:data, data}

  defp decode_body("event: " <> event), do: {:event, event}

  defp decode_body("id: " <> id), do: {:id, id}

  defp decode_body("retry: " <> retry), do: {:retry, retry}

  defp decode_body(""), do: :empty



  defp decodable_content_type?(env, opts) do

    case Tesla.get_header(env, "content-type") do

      nil -> false

      content_type -> Enum.any?(content_types(opts), &String.starts_with?(content_type, &1))

    end

  end



  defp content_types(opts),

    do: @default_content_types ++ Keyword.get(opts, :decode_content_types, [])



  defp only(message, nil), do: [message]



  defp only(message, key) do

    case Map.get(message, key) do

      nil -> []

      val -> [val]

    end

  end

end