Merge branch 'master' into improve-mocking

.github/workflows/release.yml

@@ -16,7 +16,7 @@ jobs:
         elixir-version: '1.13'
         otp-version: '24.3'
     - 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

@@ -23,7 +23,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') }}
@@ -38,6 +38,44 @@ jobs:
     - name: Run Tests
       run: mix test --trace
 
+  Test-gun1:
+    runs-on: ubuntu-latest
+    name: OTP ${{matrix.otp}} / Elixir ${{matrix.elixir}} - Gun1
+    strategy:
+      matrix:
+        elixir:
+          - 1.14
+          - 1.13
+        otp:
+          - 25.3
+          - 24.3
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Elixir
+      uses: erlef/setup-beam@v1
+      with:
+        elixir-version: ${{ matrix.elixir }}
+        otp-version: ${{ matrix.otp }}
+        version-type: strict
+    - name: Restore dependencies cache
+      uses: actions/cache@v4
+      with:
+        path: deps
+        key: ${{ runner.os }}-mix-${{ hashFiles('test/lockfiles/gun1.lock') }}
+        restore-keys: ${{ runner.os }}-mix-
+    - name: Install Dependencies
+      env:
+        MIX_ENV: test
+        LOCKFILE: gun1
+      run: |
+        mix local.rebar --force
+        mix local.hex --force
+        mix deps.get
+    - name: Run Tests
+      env:
+        LOCKFILE: gun1
+      run: mix test test/tesla/adapter/gun_test.exs --trace
+
   Linting:
     runs-on: ubuntu-latest
     steps:
@@ -49,7 +87,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://elixir-lang.org/docs/stable/elixir/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.ex

@@ -115,7 +115,7 @@ defmodule Tesla do
 
   defp prepare(module, %{pre: pre, post: post} = client, options) do
     env = struct(Env, options ++ [__module__: module, __client__: client])
-    stack = pre ++ module.__middleware__ ++ post ++ [effective_adapter(module, client)]
+    stack = pre ++ module.__middleware__() ++ post ++ [effective_adapter(module, client)]
     {env, stack}
   end
 

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/gun.ex

@@ -55,7 +55,7 @@ if Code.ensure_loaded?(:gun) do
         [ssl_verify_fun.erl](https://github.com/deadtrickster/ssl_verify_fun.erl)
 
     - `:proxy` - Proxy for requests.
-        **Socks proxy are supported only for gun master branch**.
+        **Socks proxy are supported from gun >= 2.0**.
         Examples: `{'localhost', 1234}`, `{{127, 0, 0, 1}, 1234}`, `{:socks5, 'localhost', 1234}`.
 
       **NOTE:** By default GUN uses TLS as transport if the specified port is 443,
@@ -435,7 +435,7 @@ if Code.ensure_loaded?(:gun) do
     end
 
     defp open_stream(pid, method, path, headers, body, req_opts, :stream) do
-      stream = :gun.request(pid, method, path, headers, "", req_opts)
+      stream = perform_stream_request(pid, method, path, headers, req_opts)
       for data <- body, do: :ok = :gun.data(pid, stream, :nofin, data)
       :gun.data(pid, stream, :fin, "")
       stream
@@ -553,5 +553,16 @@ if Code.ensure_loaded?(:gun) do
           ip
       end
     end
+
+    # Backwards compatibility with gun < 2.0. See https://ninenines.eu/docs/en/gun/2.0/manual/gun.headers/
+    if Application.spec(:gun, :vsn) |> List.to_string() |> Version.match?("~> 2.0") do
+      defp perform_stream_request(pid, method, path, headers, req_opts) do
+        :gun.headers(pid, method, path, headers, req_opts)
+      end
+    else
+      defp perform_stream_request(pid, method, path, headers, req_opts) do
+        :gun.request(pid, method, path, headers, "", req_opts)
+      end
+    end
   end
 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, [])