No OneTemporary
Actions

Size

20 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/tesla/adapter/gun.ex b/lib/tesla/adapter/gun.ex
	index da26927..ca819b3 100644
	--- a/lib/tesla/adapter/gun.ex
	+++ b/lib/tesla/adapter/gun.ex
	@@ -1,363 +1,363 @@
	if Code.ensure_loaded?(:gun) do
	defmodule Tesla.Adapter.Gun do
	@moduledoc """
	Adapter for [gun] https://github.com/ninenines/gun

	Remember to add `{:gun, "~> 1.3"}` to dependencies.
	In version 1.3 gun sends `host` header with port. Fixed in master branch.
	Also, you need to recompile tesla after adding `:gun` dependency:
	```
	mix deps.clean tesla
	mix deps.compile tesla
	```
	### Example usage
	```
	# set globally in config/config.exs
	config :tesla, :adapter, Tesla.Adapter.Gun
	# set per module
	defmodule MyClient do
	use Tesla
	adapter Tesla.Adapter.Gun
	end
	```

	### Adapter specific options:

	* `timeout` - Time, while process, will wait for gun messages.
	* `body_as` - What will be returned in `%Tesla.Env{}` body key. Possible values - `:plain`, `:stream`, `:chunks`. Defaults to `:plain`.
	* `:plain` - as binary.
	* `:stream` - as stream. If you don't want to close connection (because you want to reuse it later) pass `close_conn: false` in adapter opts.
	* `:chunks` - as chunks. You can get response body in chunks using `Tesla.Adapter.Gun.read_chunk/3` function.
	Processing of the chunks and checking body size must be done by yourself. Example of processing function
	is in `test/tesla/adapter/gun_test.exs` - `read_body/3`. If you don't need connection later don't forget
	to close it with `Tesla.Adapter.Gun.close/1`.
	* `max_body` - Max response body size in bytes. Works only with `body_as: :plain`, with other settings you need to check response
	body size by yourself.
	* `conn` - Opened connection pid with gun. Is used for reusing gun connections.
	* `original` - Original host with port, for which reused connection was open. Needed for `Tesla.Middleware.FollowRedirects`. Otherwise
	adapter will use connection for another open host.
	* `close_conn` - Close connection or not after receiving full response body. Is used for reusing gun connections.
	Defaults to `true`.

	### Gun options https://ninenines.eu/docs/en/gun/1.3/manual/gun/:

	* `connect_timeout` - Connection timeout.
	* `http_opts` - Options specific to the HTTP protocol.
	* `http2_opts` - Options specific to the HTTP/2 protocol.
	* `protocols` - Ordered list of preferred protocols. Defaults: [http2, http] - for :tls, [http] - for :tcp.
	* `trace` - Whether to enable dbg tracing of the connection process. Should only be used during debugging. Default: false.
	* `transport` - Whether to use TLS or plain TCP. The default varies depending on the port used. Port 443 defaults to tls.
	All other ports default to tcp.
	* `transport_opts` - Transport options. They are TCP options or TLS options depending on the selected transport. Default: [].
	Gun version: 1.3
	* `tls_opts` - TLS transport options. Default: []. Gun from master branch.
	* `tcp_opts` - TCP trasnport options. Default: []. Gun from master branch.
	* `ws_opts` - Options specific to the Websocket protocol. Default: %{}.
	* `compress` - Whether to enable permessage-deflate compression. This does not guarantee that compression will
	be used as it is the server that ultimately decides. Defaults to false.
	* `protocols` - A non-empty list enables Websocket protocol negotiation. The list of protocols will be sent
	in the sec-websocket-protocol request header.
	The handler module interface is currently undocumented and must be set to `gun_ws_h`.
	}

	"""
	@behaviour Tesla.Adapter
	alias Tesla.Multipart

	@gun_keys [
	:connect_timeout,
	:http_opts,
	:http2_opts,
	:protocols,
	:retry,
	:retry_timeout,
	:trace,
	:transport,
	:ws_opts
	]

	@adapter_default_timeout 1_000

	@impl true
	@doc false
	def call(env, opts) do
	with {:ok, status, headers, body} <- request(env, opts) do
	{:ok, %{env \| status: status, headers: format_headers(headers), body: body}}
	end
	end

	defp format_headers(headers) do
	for {key, value} <- headers do
	{String.downcase(to_string(key)), to_string(value)}
	end
	end

	defp format_method(method), do: String.upcase(to_string(method))

	- defp format_url(nil, nil), do: ""
	- defp format_url(nil, query), do: "?" <> query
	+ defp format_url(nil, nil), do: "/"
	+ defp format_url(nil, query), do: "/?" <> query
	defp format_url(path, nil), do: path
	defp format_url(path, query), do: path <> "?" <> query

	defp request(env, opts) do
	request(
	format_method(env.method),
	Tesla.build_url(env.url, env.query),
	env.headers,
	env.body \|\| "",
	Tesla.Adapter.opts(
	[close_conn: true, body_as: :plain, send_body: :at_once, receive: true],
	env,
	opts
	)
	\|> Enum.into(%{})
	)
	end

	defp request(method, url, headers, %Stream{} = body, opts),
	do: request_stream(method, url, headers, body, Map.put(opts, :send_body, :stream))

	defp request(method, url, headers, body, opts) when is_function(body),
	do: request_stream(method, url, headers, body, Map.put(opts, :send_body, :stream))

	defp request(method, url, headers, %Multipart{} = mp, opts) do
	headers = headers ++ Multipart.headers(mp)
	body = Multipart.body(mp)

	request(method, url, headers, body, opts)
	end

	defp request(method, url, headers, body, opts),
	do: do_request(method, url, headers, body, opts)

	defp request_stream(method, url, headers, body, opts),
	do: do_request(method, url, headers, body, opts)

	defp do_request(method, url, headers, body, opts) do
	with uri <- URI.parse(url),
	path_with_query <- format_url(uri.path, uri.query),
	{pid, opts} <- open_conn(uri, opts),
	stream <- open_stream(pid, method, path_with_query, headers, body, opts[:send_body]) do
	read_response(pid, stream, opts)
	end
	end

	defp open_conn(uri, %{conn: conn, original: original} = opts) do
	if original == "#{uri.host}:#{uri.port}" do
	{conn, Map.put(opts, :receive, false)}
	else
	# current url is different from the original, maybe there were redirects
	# so we can't use transferred connection
	open_conn(uri, Map.delete(opts, :conn))
	end
	end

	defp open_conn(uri, opts) do
	opts =
	if uri.scheme == "https" and uri.port != 443 do
	Map.put(opts, :transport, :tls)
	else
	opts
	end

	# We need to add `server_name_indication` option, because gun connects through ip in master branch.
	# [SNI] - http://erlang.org/doc/man/ssl.html#type-sni
	# Support for gun master branch where transport_opts, were splitted to tls_opts and tcp_opts
	# https://github.com/ninenines/gun/blob/491ddf58c0e14824a741852fdc522b390b306ae2/doc/src/manual/gun.asciidoc#changelog

	tls_opts =
	Map.get(opts, :tls_opts, [])
	\|> Keyword.merge(Map.get(opts, :transport_opts, []))

	tls_opts =
	if uri.scheme == "https" do
	host = uri.host \|> to_charlist()
	Keyword.put(tls_opts, :server_name_indication, host)
	else
	tls_opts
	end

	gun_opts = Map.take(opts, @gun_keys)

	with {:ok, pid} <- do_open_conn(uri, opts, gun_opts, tls_opts) do
	# If there were redirects, and passed `closed_conn: false`, we need to close opened connections to these intermediate hosts.
	{pid, Map.put(opts, :close_conn, true)}
	end
	end

	defp do_open_conn(uri, %{proxy: {proxy_host, proxy_port}}, gun_opts, tls_opts) do
	connect_opts = %{host: to_charlist(uri.host), port: uri.port}

	connect_opts =
	if uri.scheme == "https" do
	Map.put(connect_opts, :protocols, [:http2])
	\|> Map.put(:transport, :tls)
	\|> Map.put(:tls_opts, tls_opts)
	else
	connect_opts
	end

	with {:ok, pid} <- :gun.open(proxy_host, proxy_port, gun_opts),
	{:ok, _} <- :gun.await_up(pid),
	stream <- :gun.connect(pid, connect_opts),
	{:response, :fin, 200, _} <- :gun.await(pid, stream) do
	{:ok, pid}
	end
	end

	defp do_open_conn(uri, opts, gun_opts, tls_opts) do
	tcp_opts = Map.get(opts, :tcp_opts, [])

	# if gun used from master
	opts_with_master_keys =
	Map.put(gun_opts, :tls_opts, tls_opts)
	\|> Map.put(:tcp_opts, tcp_opts)

	host = to_charlist(uri.host)

	with {:ok, pid} <- :gun.open(host, uri.port, opts_with_master_keys) do
	{:ok, pid}
	else
	{:error, {:options, {key, _}}} when key in [:tcp_opts, :tls_opts] ->
	:gun.open(
	host,
	uri.port,
	Map.put(gun_opts, :transport_opts, tls_opts)
	)
	end
	end

	defp open_stream(pid, method, url, headers, body, :stream) do
	stream = :gun.request(pid, method, url, headers, "")
	for data <- body, do: :ok = :gun.data(pid, stream, :nofin, data)
	:gun.data(pid, stream, :fin, "")
	stream
	end

	defp open_stream(pid, method, url, headers, body, :at_once),
	do: :gun.request(pid, method, url, headers, body)

	defp read_response(pid, stream, opts) do
	receive? = opts[:receive]

	receive do
	{:gun_response, ^pid, ^stream, :fin, status, headers} ->
	if opts[:close_conn], do: close(pid)
	{:ok, status, headers, ""}

	{:gun_response, ^pid, ^stream, :nofin, status, headers} ->
	format_response(pid, stream, opts, status, headers, opts[:body_as])

	{:error, error} ->
	if opts[:close_conn], do: close(pid)
	{:error, error}

	{:gun_up, ^pid, _protocol} when receive? ->
	read_response(pid, stream, opts)

	{:gun_error, ^pid, reason} ->
	if opts[:close_conn], do: close(pid)
	{:error, reason}

	{:gun_down, ^pid, _, _, _, _} when receive? ->
	read_response(pid, stream, opts)

	{:DOWN, _, _, _, reason} ->
	if opts[:close_conn], do: close(pid)
	{:error, reason}
	after
	opts[:timeout] \|\| @adapter_default_timeout ->
	if opts[:close_conn], do: :ok = close(pid)
	{:error, :timeout}
	end
	end

	defp format_response(pid, stream, opts, status, headers, :plain) do
	case read_body(pid, stream, opts) do
	{:ok, body} ->
	if opts[:close_conn], do: close(pid)
	{:ok, status, headers, body}

	{:error, error} ->
	if opts[:close_conn], do: close(pid)
	{:error, error}
	end
	end

	defp format_response(pid, stream, opts, status, headers, :stream) do
	stream_body =
	Stream.resource(
	fn -> %{pid: pid, stream: stream} end,
	fn
	%{pid: pid, stream: stream} ->
	case read_chunk(pid, stream, opts) do
	{:nofin, part} -> {[part], %{pid: pid, stream: stream}}
	{:fin, body} -> {[body], %{pid: pid, final: :fin}}
	end

	%{pid: pid, final: :fin} ->
	{:halt, %{pid: pid}}
	end,
	fn %{pid: pid} ->
	if opts[:close_conn], do: close(pid)
	end
	)

	{:ok, status, headers, stream_body}
	end

	defp format_response(pid, stream, opts, status, headers, :chunks) do
	{:ok, status, headers, %{pid: pid, stream: stream, opts: Enum.into(opts, [])}}
	end

	@spec read_chunk(pid(), reference(), keyword() \| map()) ::
	{:fin, binary()} \| {:nofin, binary()} \| {:error, :timeout}
	def read_chunk(pid, stream, opts) do
	receive do
	{:gun_data, ^pid, ^stream, :fin, body} ->
	{:fin, body}

	{:gun_data, ^pid, ^stream, :nofin, part} ->
	{:nofin, part}
	after
	opts[:timeout] \|\| @adapter_default_timeout ->
	{:error, :timeout}
	end
	end

	@spec close(pid()) :: :ok
	def close(pid) do
	:gun.close(pid)
	end

	defp read_body(pid, stream, opts, acc \\ "") do
	limit = opts[:max_body]

	receive do
	{:gun_data, ^pid, ^stream, :fin, body} ->
	check_body_size(acc, body, limit)

	{:gun_data, ^pid, ^stream, :nofin, part} ->
	case check_body_size(acc, part, limit) do
	{:ok, acc} -> read_body(pid, stream, opts, acc)
	{:error, error} -> {:error, error}
	end
	after
	opts[:timeout] \|\| @adapter_default_timeout ->
	{:error, :timeout}
	end
	end

	defp check_body_size(acc, part, nil), do: {:ok, acc <> part}

	defp check_body_size(acc, part, limit) do
	body = acc <> part

	if limit - byte_size(body) >= 0 do
	{:ok, body}
	else
	{:error, :body_too_large}
	end
	end
	end
	end
	diff --git a/test/tesla/adapter/gun_test.exs b/test/tesla/adapter/gun_test.exs
	index a98d6e5..acfc29c 100644
	--- a/test/tesla/adapter/gun_test.exs
	+++ b/test/tesla/adapter/gun_test.exs
	@@ -1,253 +1,266 @@
	defmodule Tesla.Adapter.GunTest do
	use ExUnit.Case

	use Tesla.AdapterCase, adapter: Tesla.Adapter.Gun
	use Tesla.AdapterCase.Basic
	use Tesla.AdapterCase.Multipart
	use Tesla.AdapterCase.StreamRequestBody
	use Tesla.AdapterCase.SSL
	alias Tesla.Adapter.Gun

	test "fallback adapter timeout option" do
	request = %Env{
	method: :get,
	url: "#{@http}/delay/2"
	}

	assert {:error, :timeout} = call(request, timeout: 1_000)
	end

	test "max_body option" do
	request = %Env{
	method: :get,
	url: "#{@http}/stream-bytes/100",
	query: [
	message: "Hello world!"
	]
	}

	assert {:error, :body_too_large} = call(request, max_body: 5)
	end

	test "without slash" do
	request = %Env{
	method: :get,
	url: "#{@http}"
	}

	assert {:ok, %Env{} = response} = call(request)
	assert response.status == 400
	end

	test "response stream" do
	request = %Env{
	method: :get,
	url: "#{@http}/stream-bytes/10"
	}

	assert {:ok, %Env{} = response} = call(request)
	assert response.status == 200
	assert byte_size(response.body) == 16
	end

	test "read response body in chunks" do
	request = %Env{
	method: :get,
	url: "#{@http}/stream-bytes/10"
	}

	assert {:ok, %Env{} = response} = call(request, body_as: :chunks)
	assert response.status == 200
	%{pid: pid, stream: stream, opts: opts} = response.body
	assert opts[:body_as] == :chunks
	assert is_pid(pid)
	assert is_reference(stream)

	assert read_body(pid, stream) \|> byte_size() == 16
	refute Process.alive?(pid)
	end

	test "with body_as :plain reusinng connection" do
	uri = URI.parse(@http)
	{:ok, conn} = :gun.open(to_charlist(uri.host), uri.port)

	original = "#{uri.host}:#{uri.port}"

	request = %Env{
	method: :get,
	url: "#{@http}/ip"
	}

	assert {:ok, %Env{} = response} =
	call(request, conn: conn, close_conn: false, original: original)

	assert response.status == 200
	assert Process.alive?(conn)

	assert {:ok, %Env{} = response} =
	call(request, conn: conn, close_conn: false, original: original)

	assert response.status == 200
	assert Process.alive?(conn)
	:ok = Gun.close(conn)
	refute Process.alive?(conn)
	end

	test "read response body in chunks with reused connection and closing it" do
	uri = URI.parse(@http)
	{:ok, conn} = :gun.open(to_charlist(uri.host), uri.port)

	original = "#{uri.host}:#{uri.port}"

	request = %Env{
	method: :get,
	url: "#{@http}/stream-bytes/10"
	}

	assert {:ok, %Env{} = response} =
	call(request, body_as: :chunks, conn: conn, original: original)

	assert response.status == 200
	%{pid: pid, stream: stream, opts: opts} = response.body
	assert opts[:body_as] == :chunks
	assert is_pid(pid)
	assert is_reference(stream)
	assert conn == pid

	assert read_body(pid, stream, "", false) \|> byte_size() == 16
	assert Process.alive?(pid)

	# reusing connection
	assert {:ok, %Env{} = response} =
	call(request, body_as: :chunks, conn: conn, original: original)

	assert response.status == 200
	%{pid: pid, stream: stream, opts: opts} = response.body
	assert opts[:body_as] == :chunks
	assert is_pid(pid)
	assert is_reference(stream)
	assert conn == pid

	assert read_body(pid, stream, "", false) \|> byte_size() == 16
	assert Process.alive?(pid)

	:ok = Gun.close(pid)
	refute Process.alive?(pid)
	end

	test "don't reuse connection if original does not match" do
	uri = URI.parse(@http)
	{:ok, conn} = :gun.open(to_charlist(uri.host), uri.port)

	request = %Env{
	method: :get,
	url: "#{@http}/stream-bytes/10"
	}

	assert {:ok, %Env{} = response} =
	call(request, body_as: :chunks, conn: conn, original: "example.com:80")

	assert response.status == 200
	%{pid: pid, stream: stream, opts: opts} = response.body
	assert opts[:body_as] == :chunks
	assert is_pid(pid)
	assert is_reference(stream)

	assert read_body(pid, stream, "", false) \|> byte_size() == 16
	assert Process.alive?(pid)
	refute conn == pid
	end

	test "read response body in stream" do
	request = %Env{
	method: :get,
	url: "#{@http}/stream-bytes/10"
	}

	assert {:ok, %Env{} = response} = call(request, body_as: :stream)
	assert response.status == 200
	assert is_function(response.body)
	assert Enum.to_list(response.body) \|> List.to_string() \|> byte_size() == 16
	end

	test "read response body in stream with opened connection without closing connection" do
	uri = URI.parse(@http)
	{:ok, conn} = :gun.open(to_charlist(uri.host), uri.port)
	original = "#{uri.host}:#{uri.port}"

	request = %Env{
	method: :get,
	url: "#{@http}/stream-bytes/10"
	}

	assert {:ok, %Env{} = response} =
	call(request, body_as: :stream, conn: conn, close_conn: false, original: original)

	assert response.status == 200
	assert is_function(response.body)
	assert Enum.to_list(response.body) \|> List.to_string() \|> byte_size() == 16

	assert Process.alive?(conn)

	:ok = Gun.close(conn)
	refute Process.alive?(conn)
	end

	test "read response body in stream with opened connection with closing connection" do
	uri = URI.parse(@http)
	{:ok, conn} = :gun.open(to_charlist(uri.host), uri.port)

	original = "#{uri.host}:#{uri.port}"

	request = %Env{
	method: :get,
	url: "#{@http}/stream-bytes/10"
	}

	assert {:ok, %Env{} = response} =
	call(request, body_as: :stream, conn: conn, original: original)

	assert response.status == 200
	assert is_function(response.body)
	assert Enum.to_list(response.body) \|> List.to_string() \|> byte_size() == 16

	refute Process.alive?(conn)
	end

	test "error response" do
	request = %Env{
	method: :get,
	url: "#{@http}/status/500"
	}

	assert {:ok, %Env{} = response} = call(request, timeout: 1_000)
	assert response.status == 500
	end

	test "query without path" do
	request = %Env{
	method: :get,
	url: "#{@http}",
	query: [
	param: "value"
	]
	}

	assert {:ok, %Env{} = response} = call(request, timeout: 1_000)
	- assert response.status == 400
	+ assert response.status == 200
	+ end
	+
	+ test "query without path with query" do
	+ request = %Env{
	+ method: :get,
	+ url: "#{@http}",
	+ query: [
	+ param: "value"
	+ ]
	+ }
	+
	+ assert {:ok, %Env{} = response} = call(request, timeout: 1_000)
	+ assert response.status == 200
	end

	defp read_body(pid, stream, acc \\ "", close_conn \\ true) do
	case Gun.read_chunk(pid, stream, timeout: 1_000) do
	{:fin, body} ->
	if close_conn do
	:ok = Gun.close(pid)
	end

	acc <> body

	{:nofin, part} ->
	read_body(pid, stream, acc <> part, close_conn)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Tue, Nov 26, 5:25 PM (1 d, 9 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 40441
Default Alt Text: (20 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions