Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 21 additions & 1 deletion lib/ex_aws/bedrock/event_stream.ex
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,10 @@ defmodule ExAws.Bedrock.EventStream do
encoded_data
)

hackney_opts = hackney_options(config)

request_fun = fn [] ->
{:ok, ref} = :hackney.post(url, full_headers, encoded_data, @hackney_options)
{:ok, ref} = :hackney.post(url, full_headers, encoded_data, hackney_opts)

receive do
{:hackney_response, ^ref, {:status, 200, _reason}} ->
Expand Down Expand Up @@ -104,6 +106,24 @@ defmodule ExAws.Bedrock.EventStream do
Stream.flat_map(stream, &decode_chunk/1)
end

@doc """
Builds the hackney options for the streaming request.

Merges caller-provided options from the ExAws config `:http_opts` (e.g.
`recv_timeout`, `connect_timeout`, `pool`) on top of the async-streaming
defaults. Without this, the stream would always use hackney's built-in
`recv_timeout` (5s) and drop slow responses regardless of the timeout the
caller configured.

The streaming defaults win on conflicting keys, so the async-streaming mode
(`async: :once`) can't be accidentally disabled by caller options.
"""
def hackney_options(config) do
config
|> Map.get(:http_opts, [])
|> Keyword.merge(@hackney_options)
end

defp verify_event_stream!(headers) do
verify_header!(headers, "Content-Type", @content_type)
end
Expand Down
25 changes: 25 additions & 0 deletions test/ex_aws/bedrock/event_stream_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,31 @@ defmodule ExAws.Bedrock.EventStreamTest do
end
end

describe "hackney_options/1" do
test "defaults to async streaming options when no http_opts are given" do
assert EventStream.hackney_options(%{}) == [async: :once]
end

test "honors caller recv_timeout / connect_timeout / pool from :http_opts" do
opts =
EventStream.hackney_options(%{
http_opts: [recv_timeout: 600_000, connect_timeout: 10_000, pool: :ex_aws]
})

assert Keyword.get(opts, :async) == :once
assert Keyword.get(opts, :recv_timeout) == 600_000
assert Keyword.get(opts, :connect_timeout) == 10_000
assert Keyword.get(opts, :pool) == :ex_aws
end

test "streaming defaults win so async: :once cannot be disabled by caller opts" do
opts = EventStream.hackney_options(%{http_opts: [async: false, recv_timeout: 1_000]})

assert Keyword.get(opts, :async) == :once
assert Keyword.get(opts, :recv_timeout) == 1_000
end
end

setup_all do
# Claude 3.5 Sonnet Multi-Chunk response
multipart_chunk =
Expand Down