Merge branch 'master' into dialyzer

Author: nathany-copia

.github/workflows/release.yml

@@ -16,7 +16,7 @@ jobs:
         elixir-version: '1.16.0'
         otp-version: '26.2.1'
     - name: Restore dependencies cache
-      uses: actions/cache@v3
+      uses: actions/cache@v4
       with:
         path: deps
         key: ${{ runner.os }}-mix-${{ hashFiles('**/mix.lock') }}

.github/workflows/test.yml

@@ -26,7 +26,7 @@ jobs:
         otp-version: ${{ matrix.otp }}
         version-type: strict
     - name: Restore dependencies cache
-      uses: actions/cache@v3
+      uses: actions/cache@v4
       with:
         path: deps
         key: ${{ runner.os }}-mix-${{ hashFiles('**/mix.lock') }}
@@ -64,7 +64,7 @@ jobs:
         otp-version: ${{ matrix.otp }}
         version-type: strict
     - name: Restore dependencies cache
-      uses: actions/cache@v3
+      uses: actions/cache@v4
       with:
         path: deps
         key: ${{ runner.os }}-mix-${{ hashFiles('test/lockfiles/gun1.lock') }}
@@ -93,7 +93,7 @@ jobs:
         otp-version: '24.3'
         version-type: strict
     - name: Restore dependencies cache
-      uses: actions/cache@v3
+      uses: actions/cache@v4
       with:
         path: deps
         key: ${{ runner.os }}-mix-${{ hashFiles('**/mix.lock') }}

README.md

@@ -12,8 +12,8 @@ Tesla is an HTTP client loosely based on [Faraday](https://github.com/lostisland
 It embraces the concept of middleware when processing the request/response cycle.
 
 > Note that this README refers to the `master` branch of Tesla, not the latest
-  released version on Hex. See [the documentation](https://hexdocs.pm/tesla) for
-  the documentation of the version you're using.
+> released version on Hex. See [the documentation](https://hexdocs.pm/tesla) for
+> the documentation of the version you're using.
 
 For the list of changes, checkout the latest [release notes](https://github.com/teamon/tesla/releases).
 
@@ -61,13 +61,13 @@ Add `:tesla` as dependency in `mix.exs`:
 ```elixir
 defp deps do
   [
-    {:tesla, "~> 1.4"},
+    {:tesla, "~> 1.9"},
 
     # optional, but recommended adapter
-    {:hackney, "~> 1.17"},
+    {:hackney, "~> 1.20"},
 
     # optional, required by JSON middleware
-    {:jason, ">= 1.0.0"}
+    {:jason, "~> 1.4"}
   ]
 end
 ```
@@ -83,8 +83,8 @@ config :tesla, adapter: Tesla.Adapter.Hackney
 ```
 
 > The default adapter is erlang's built-in `httpc`, but it is not recommended
-to use it in production environment as it does not validate SSL certificates
-[among other issues](https://github.com/teamon/tesla/issues?utf8=%E2%9C%93&q=is%3Aissue+label%3Ahttpc+).
+> to use it in production environment as it does not validate SSL certificates
+> [among other issues](https://github.com/teamon/tesla/issues?utf8=%E2%9C%93&q=is%3Aissue+label%3Ahttpc+).
 
 ## Documentation
 
@@ -198,8 +198,8 @@ When using adapter other than `:httpc` remember to add it to the dependencies li
 ```elixir
 defp deps do
   [
-    {:tesla, "~> 1.4.0"},
-    {:hackney, "~> 1.10"} # when using hackney adapter
+    {:tesla, "~> 1.9"},
+    {:hackney, "~> 1.20"} # when using hackney adapter
   ]
 end
 ```
@@ -243,7 +243,11 @@ Tesla.get(client, "/", opts: [adapter: [recv_timeout: 30_000]])
 
 ## Streaming
 
-If adapter supports it, you can pass a [Stream](https://hexdocs.pm/elixir/main/Stream.html) as body, e.g.:
+### Streaming Request Body
+
+If adapter supports it, you can pass a
+[Stream](https://hexdocs.pm/elixir/main/Stream.html) as request
+body, e.g.:
 
 ```elixir
 defmodule ElasticSearch do
@@ -259,7 +263,41 @@ defmodule ElasticSearch do
 end
 ```
 
-Each piece of stream will be encoded as JSON and sent as a new line (conforming to JSON stream format).
+Each piece of stream will be encoded as JSON and sent as a new line (conforming
+to JSON stream format).
+
+### Streaming Response Body
+
+If adapter supports it, you can pass a `response: :stream` option to return
+response body as a
+[Stream](https://elixir-lang.org/docs/stable/elixir/Stream.html)
+
+```elixir
+defmodule OpenAI do
+  def new(token) do
+    middleware = [
+      {Tesla.Middleware.BaseUrl, "https://api.openai.com/v1"},
+      {Tesla.Middleware.BearerAuth, token: token},
+      {Tesla.Middleware.JSON, decode_content_types: ["text/event-stream"]},
+      {Tesla.Middleware.SSE, only: :data}
+    ]
+    Tesla.client(middleware, {Tesla.Adapter.Finch, name: MyFinch})
+  end
+
+  def completion(client, prompt) do
+    data = %{
+      model: "gpt-3.5-turbo",
+      messages: [%{role: "user", content: prompt}],
+      stream: true
+    }
+    Tesla.post(client, "/chat/completions", data, opts: [adapter: [response: :stream]])
+  end
+end
+client = OpenAI.new("<token>")
+{:ok, env} = OpenAI.completion(client, "What is the meaning of life?")
+env.body
+|> Stream.each(fn chunk -> IO.inspect(chunk) end)
+```
 
 ## Multipart
 
@@ -476,6 +514,7 @@ use Tesla, except: [:delete, :options]
 ```elixir
 use Tesla, docs: false
 ```
+
 ### Encode only JSON request (do not decode response)
 
 ```elixir

lib/tesla/adapter/finch.ex

@@ -52,37 +52,99 @@ if Code.ensure_loaded?(Finch) do
     @behaviour Tesla.Adapter
     alias Tesla.Multipart
 
+    @defaults [
+      receive_timeout: 15_000
+    ]
+
     @impl Tesla.Adapter
     def call(%Tesla.Env{} = env, opts) do
-      opts = Tesla.Adapter.opts(env, opts)
+      opts = Tesla.Adapter.opts(@defaults, env, opts)
 
       name = Keyword.fetch!(opts, :name)
       url = Tesla.build_url(env.url, env.query)
       req_opts = Keyword.take(opts, [:pool_timeout, :receive_timeout])
+      req = build(env.method, url, env.headers, env.body)
 
-      case request(name, env.method, url, env.headers, env.body, req_opts) do
+      case request(req, name, req_opts, opts) do
         {:ok, %Finch.Response{status: status, headers: headers, body: body}} ->
           {:ok, %Tesla.Env{env | status: status, headers: headers, body: body}}
 
-        {:error, mint_error} ->
-          {:error, Exception.message(mint_error)}
+        {:error, %Mint.TransportError{reason: reason}} ->
+          {:error, reason}
+
+        {:error, reason} ->
+          {:error, reason}
       end
     end
 
-    defp request(name, method, url, headers, %Multipart{} = mp, opts) do
+    defp build(method, url, headers, %Multipart{} = mp) do
       headers = headers ++ Multipart.headers(mp)
       body = Multipart.body(mp) |> Enum.to_list()
 
-      request(name, method, url, headers, body, opts)
+      build(method, url, headers, body)
     end
 
-    defp request(_name, _method, _url, _headers, %Stream{}, _opts) do
-      raise "Streaming is not supported by this adapter!"
+    defp build(method, url, headers, %Stream{} = body_stream) do
+      build(method, url, headers, {:stream, body_stream})
     end
 
-    defp request(name, method, url, headers, body, opts) do
+    defp build(method, url, headers, body_stream_fun) when is_function(body_stream_fun) do
+      build(method, url, headers, {:stream, body_stream_fun})
+    end
+
+    defp build(method, url, headers, body) do
       Finch.build(method, url, headers, body)
-      |> Finch.request(name, opts)
+    end
+
+    defp request(req, name, req_opts, opts) do
+      case opts[:response] do
+        :stream -> stream(req, name, req_opts)
+        nil -> Finch.request(req, name, req_opts)
+        other -> raise "Unknown response option: #{inspect(other)}"
+      end
+    end
+
+    defp stream(req, name, opts) do
+      owner = self()
+      ref = make_ref()
+
+      fun = fn
+        {:status, status}, _acc -> status
+        {:headers, headers}, status -> send(owner, {ref, {:status, status, headers}})
+        {:data, data}, _acc -> send(owner, {ref, {:data, data}})
+      end
+
+      task =
+        Task.async(fn ->
+          case Finch.stream(req, name, nil, fun, opts) do
+            {:ok, _acc} -> send(owner, {ref, :eof})
+            {:error, error} -> send(owner, {ref, {:error, error}})
+          end
+        end)
+
+      receive do
+        {^ref, {:status, status, headers}} ->
+          body =
+            Stream.unfold(nil, fn _ ->
+              receive do
+                {^ref, {:data, data}} ->
+                  {data, nil}
+
+                {^ref, :eof} ->
+                  Task.await(task)
+                  nil
+              after
+                opts[:receive_timeout] ->
+                  Task.shutdown(task, :brutal_kill)
+                  nil
+              end
+            end)
+
+          {:ok, %Finch.Response{status: status, headers: headers, body: body}}
+      after
+        opts[:receive_timeout] ->
+          {:error, :timeout}
+      end
     end
   end
 end

lib/tesla/adapter/httpc.ex

@@ -55,33 +55,33 @@ defmodule Tesla.Adapter.Httpc do
         Enum.map(env.headers, fn {k, v} -> {to_charlist(k), to_charlist(v)} end),
         content_type,
         env.body,
-        Keyword.split(opts, @http_opts)
+        opts
       )
     )
   end
 
   # fix for # see https://github.com/teamon/tesla/issues/147
-  defp request(:delete, url, headers, content_type, nil, {http_opts, opts}) do
-    request(:delete, url, headers, content_type, "", {http_opts, opts})
+  defp request(:delete, url, headers, content_type, nil, opts) do
+    request(:delete, url, headers, content_type, "", opts)
   end
 
-  defp request(method, url, headers, _content_type, nil, {http_opts, opts}) do
-    :httpc.request(method, {url, headers}, http_opts, opts)
+  defp request(method, url, headers, _content_type, nil, opts) do
+    :httpc.request(method, {url, headers}, http_opts(opts), adapter_opts(opts), profile(opts))
   end
 
   # These methods aren't able to contain a content_type and body
-  defp request(method, url, headers, _content_type, _body, {http_opts, opts})
+  defp request(method, url, headers, _content_type, _body, opts)
        when method in [:get, :options, :head, :trace] do
-    :httpc.request(method, {url, headers}, http_opts, opts)
+    :httpc.request(method, {url, headers}, http_opts(opts), adapter_opts(opts), profile(opts))
   end
 
   defp request(method, url, headers, _content_type, %Multipart{} = mp, opts) do
     headers = headers ++ Multipart.headers(mp)
     headers = for {key, value} <- headers, do: {to_charlist(key), to_charlist(value)}
 
     {content_type, headers} =
-      case List.keytake(headers, 'content-type', 0) do
-        nil -> {'text/plain', headers}
+      case List.keytake(headers, ~c"content-type", 0) do
+        nil -> {~c"text/plain", headers}
         {{_, ct}, headers} -> {ct, headers}
       end
 
@@ -100,10 +100,22 @@ defmodule Tesla.Adapter.Httpc do
     request(method, url, headers, content_type, body, opts)
   end
 
-  defp request(method, url, headers, content_type, body, {http_opts, opts}) do
-    :httpc.request(method, {url, headers, content_type, body}, http_opts, opts)
+  defp request(method, url, headers, content_type, body, opts) do
+    :httpc.request(
+      method,
+      {url, headers, content_type, body},
+      http_opts(opts),
+      adapter_opts(opts),
+      profile(opts)
+    )
   end
 
   defp handle({:error, {:failed_connect, _}}), do: {:error, :econnrefused}
   defp handle(response), do: response
+
+  defp http_opts(opts), do: opts |> Keyword.take(@http_opts) |> Keyword.delete(:profile)
+
+  defp adapter_opts(opts), do: opts |> Keyword.drop(@http_opts) |> Keyword.delete(:profile)
+
+  defp profile(opts), do: opts[:profile] || :default
 end

lib/tesla/middleware/json.ex

@@ -113,12 +113,18 @@ defmodule Tesla.Middleware.JSON do
     end
   end
 
+  defp decode_body(body, opts) when is_struct(body, Stream) or is_function(body),
+    do: {:ok, decode_stream(body, opts)}
+
   defp decode_body(body, opts), do: process(body, :decode, opts)
 
   defp decodable?(env, opts), do: decodable_body?(env) && decodable_content_type?(env, opts)
 
   defp decodable_body?(env) do
-    (is_binary(env.body) && env.body != "") || (is_list(env.body) && env.body != [])
+    (is_binary(env.body) && env.body != "") ||
+      (is_list(env.body) && env.body != []) ||
+      is_function(env.body) ||
+      is_struct(env.body, Stream)
   end
 
   defp decodable_content_type?(env, opts) do
@@ -128,6 +134,15 @@ defmodule Tesla.Middleware.JSON do
     end
   end
 
+  defp decode_stream(body, opts) do
+    Stream.map(body, fn chunk ->
+      case decode_body(chunk, opts) do
+        {:ok, item} -> item
+        _ -> chunk
+      end
+    end)
+  end
+
   defp content_types(opts),
     do: @default_content_types ++ Keyword.get(opts, :decode_content_types, [])