From fbae48d562bebd97292481328fc0971f24106de0 Mon Sep 17 00:00:00 2001 From: Milad Rastian Date: Wed, 12 Aug 2026 06:50:36 +0200 Subject: [PATCH 1/7] Add low-level Req SQS request client --- config/config.exs | 3 + config/dev.exs | 1 + config/test.exs | 3 + lib/broadway_sqs/req_client/request.ex | 112 ++++++++++++++++++ mix.exs | 2 + mix.lock | 20 +++- test/broadway_sqs/req_client/request_test.exs | 84 +++++++++++++ 7 files changed, 220 insertions(+), 5 deletions(-) create mode 100644 config/config.exs create mode 100644 config/dev.exs create mode 100644 config/test.exs create mode 100644 lib/broadway_sqs/req_client/request.ex create mode 100644 test/broadway_sqs/req_client/request_test.exs diff --git a/config/config.exs b/config/config.exs new file mode 100644 index 0000000..d1186fe --- /dev/null +++ b/config/config.exs @@ -0,0 +1,3 @@ +import Config + +import_config "#{config_env()}.exs" diff --git a/config/dev.exs b/config/dev.exs new file mode 100644 index 0000000..becde76 --- /dev/null +++ b/config/dev.exs @@ -0,0 +1 @@ +import Config diff --git a/config/test.exs b/config/test.exs new file mode 100644 index 0000000..739552e --- /dev/null +++ b/config/test.exs @@ -0,0 +1,3 @@ +import Config + +config :aws_credentials, credential_providers: [] diff --git a/lib/broadway_sqs/req_client/request.ex b/lib/broadway_sqs/req_client/request.ex new file mode 100644 index 0000000..3ccd9c5 --- /dev/null +++ b/lib/broadway_sqs/req_client/request.ex @@ -0,0 +1,112 @@ +defmodule BroadwaySQS.ReqClient.Request do + @moduledoc false + + @content_type "application/x-amz-json-1.0" + + @type options :: [ + credentials: map() | keyword(), + endpoint: String.t(), + region: String.t(), + queue_url: String.t() + ] + + @doc """ + Sends one JSON SQS API request. + + The request is signed with AWS Signature Version 4. Credentials can be + supplied in `opts` or are loaded from `aws_credentials`. + """ + @spec call(String.t(), map(), options()) :: {:ok, map()} | {:error, term()} + def call(action, payload, opts \\ []) when is_binary(action) and is_map(payload) do + with {:ok, queue_url} <- queue_url(payload, opts), + {:ok, credentials} <- credentials(opts), + {:ok, region} <- region(opts, credentials), + req = request(opts, credentials, region, queue_url), + {:ok, response} <- + Req.post(req, + headers: [{"x-amz-target", action}], + json: payload, + decode_body: false + ), + {:ok, body} <- decode_body(response) do + if response.status >= 200 and response.status < 300 do + {:ok, body} + else + {:error, {:http_error, response.status, body}} + end + end + end + + defp request(opts, credentials, region, queue_url) do + headers = [ + {"content-type", @content_type} + ] + + headers = + case credentials[:token] do + nil -> headers + token -> [{"x-amz-security-token", token} | headers] + end + + Req.new( + url: Keyword.get(opts, :endpoint, queue_url), + headers: headers, + aws_sigv4: [ + access_key_id: credentials[:access_key_id], + secret_access_key: credentials[:secret_access_key], + region: region, + service: :sqs + ] + ) + end + + defp credentials(opts) do + credentials = + case Keyword.fetch(opts, :credentials) do + {:ok, credentials} -> credentials + :error -> aws_credentials() + end + + credentials = normalize_credentials(credentials) + + if credentials[:access_key_id] && credentials[:secret_access_key] do + {:ok, credentials} + else + {:error, :aws_credentials_not_found} + end + end + + defp aws_credentials do + case :aws_credentials.get_credentials() do + credentials when is_map(credentials) -> credentials + _ -> %{} + end + end + + defp normalize_credentials(credentials) when is_list(credentials), do: Map.new(credentials) + defp normalize_credentials(credentials) when is_map(credentials), do: credentials + defp normalize_credentials(_credentials), do: %{} + + defp queue_url(payload, opts) do + case Keyword.get(opts, :queue_url, payload["QueueUrl"]) do + queue_url when is_binary(queue_url) and queue_url != "" -> {:ok, queue_url} + _ -> {:error, :queue_url_not_found} + end + end + + defp region(opts, credentials) do + case Keyword.get(opts, :region) || credentials[:region] do + region when is_binary(region) and region != "" -> {:ok, region} + _ -> {:error, :aws_region_not_found} + end + end + + defp decode_body(%{body: body}) when is_binary(body) do + case Jason.decode(body) do + {:ok, decoded} -> {:ok, decoded} + {:error, reason} -> {:error, {:invalid_json, reason, body}} + end + end + + defp decode_body(%{body: body}), do: {:error, {:invalid_body, body}} +end diff --git a/mix.exs b/mix.exs index 54dbdff..ec2ee18 100644 --- a/mix.exs +++ b/mix.exs @@ -27,6 +27,8 @@ defmodule BroadwaySqs.MixProject do defp deps do [ {:broadway, "~> 1.0"}, + {:req, "~> 0.7.2"}, + {:aws_credentials, "~> 1.0"}, {:ex_aws_sqs, "~> 3.2.1 or ~> 3.3"}, {:nimble_options, "~> 0.3.7 or ~> 0.4 or ~> 1.0"}, {:telemetry, "~> 0.4.3 or ~> 1.0"}, diff --git a/mix.lock b/mix.lock index b73e7e1..c113d20 100644 --- a/mix.lock +++ b/mix.lock @@ -1,4 +1,5 @@ %{ + "aws_credentials": {:hex, :aws_credentials, "1.1.1", "4a28d7d2c01956dd9a2cc52a7edd5cb1e58343766715753d4b8d0d043771b95c", [:rebar3], [{:eini, "~> 2.2.5", [hex: :eini_beam, repo: "hexpm", optional: false]}, {:iso8601, "~> 1.3.4", [hex: :iso8601, repo: "hexpm", optional: false]}, {:jsx, "~> 3.1.0", [hex: :jsx, repo: "hexpm", optional: false]}], "hexpm", "8655e0e3c82c5ad659729ea8717ae1f0a087ea268e15aefa674fc3d781c1e9e2"}, "broadway": {:hex, :broadway, "1.0.7", "7808f9e3eb6f53ca6d060f0f9d61012dd8feb0d7a82e62d087dd517b9b66fa53", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.3.7 or ~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "e76cfb0a7d64176c387b8b1ddbfb023e2ee8a63e92f43664d78e6d5d0b1177c6"}, "bypass": {:hex, :bypass, "2.1.0", "909782781bf8e20ee86a9cabde36b259d44af8b9f38756173e8f5e2e1fabb9b1", [:mix], [{:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.0", [hex: :plug_cowboy, repo: "hexpm", optional: false]}, {:ranch, "~> 1.3", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "d9b5df8fa5b7a6efa08384e9bbecfe4ce61c77d28a4282f79e02f1ef78d96b80"}, "certifi": {:hex, :certifi, "2.9.0", "6f2a475689dd47f19fb74334859d460a2dc4e3252a3324bd2111b8f0429e7e21", [:rebar3], [], "hexpm", "266da46bdb06d6c6d35fde799bcb28d36d985d424ad7c08b5bb48f5b5cdd4641"}, @@ -6,27 +7,36 @@ "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.4.0", "f239f68b588efa7707abce16a84d0d2acf3a0f50571f8bb7f56a15865aae820c", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7d98bac1ee4565d31b62d59f8823dfd8356a169e7fcbb83831b8a5397404c9de"}, "cowlib": {:hex, :cowlib, "2.12.1", "a9fa9a625f1d2025fe6b462cb865881329b5caff8f1854d1cbc9f9533f00e1e1", [:make, :rebar3], [], "hexpm", "163b73f6367a7341b33c794c4e88e7dbfe6498ac42dcd69ef44c5bc5507c8db0"}, "earmark_parser": {:hex, :earmark_parser, "1.4.39", "424642f8335b05bb9eb611aa1564c148a8ee35c9c8a8bba6e129d51a3e3c6769", [:mix], [], "hexpm", "06553a88d1f1846da9ef066b87b57c6f605552cfbe40d20bd8d59cc6bde41944"}, + "eini": {:hex, :eini_beam, "2.2.5", "28fa0a4eb7ff885cc388877b73fbd56fcca3b2eae4d81e47c47f317b900da1da", [:rebar3], [], "hexpm", "511e9207649f3becb5d945f1813615987899cf78fa130f92be07193a6a74ecb8"}, "ex_aws": {:hex, :ex_aws, "2.4.3", "6c6d88ba7b9c07e3b0f4b70406d5fccb9f5358f5ef18138f7bd396f7863e8255", [:mix], [{:configparser_ex, "~> 4.0", [hex: :configparser_ex, repo: "hexpm", optional: true]}, {:hackney, "~> 1.16", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:jsx, "~> 2.8 or ~> 3.0", [hex: :jsx, repo: "hexpm", optional: true]}, {:mime, "~> 1.2 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:sweet_xml, "~> 0.7", [hex: :sweet_xml, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "67f61f8b6aec740150d483a21f551fabce26a481d9917305ed2bb47717007519"}, "ex_aws_sqs": {:hex, :ex_aws_sqs, "3.4.0", "f7c4d0177c1c954776363d3dc05e5dfd37ddf0e2c65ec3f047e5c9c7dd1b71ac", [:mix], [{:ex_aws, "~> 2.1", [hex: :ex_aws, repo: "hexpm", optional: false]}, {:hackney, "~> 1.9", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:saxy, "~> 1.1", [hex: :saxy, repo: "hexpm", optional: true]}, {:sweet_xml, ">= 0.0.0", [hex: :sweet_xml, repo: "hexpm", optional: true]}], "hexpm", "b504482206ccaf767b714888e9d41a1cfcdcb241577985517114191c812f155a"}, "ex_doc": {:hex, :ex_doc, "0.34.1", "9751a0419bc15bc7580c73fde506b17b07f6402a1e5243be9e0f05a68c723368", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "d441f1a86a235f59088978eff870de2e815e290e44a8bd976fe5d64470a4c9d2"}, + "finch": {:hex, :finch, "0.23.0", "e3f9287ac25a8832f848b144c2b57346aac65b205e2e0629a52adfe6507fd837", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.8", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "80e58d3f936f57e3fdf404f83a3642897ae6d9fb642934e46da4d8fe761b99d5"}, "gen_stage": {:hex, :gen_stage, "1.2.1", "19d8b5e9a5996d813b8245338a28246307fd8b9c99d1237de199d21efc4c76a1", [:mix], [], "hexpm", "83e8be657fa05b992ffa6ac1e3af6d57aa50aace8f691fcf696ff02f8335b001"}, "hackney": {:hex, :hackney, "1.18.1", "f48bf88f521f2a229fc7bae88cf4f85adc9cd9bcf23b5dc8eb6a1788c662c4f6", [:rebar3], [{:certifi, "~> 2.9.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~> 6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~> 1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.3.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "a4ecdaff44297e9b5894ae499e9a070ea1888c84afdd1fd9b7b2bc384950128e"}, + "hpax": {:hex, :hpax, "1.0.4", "777de5d433b0fbdc7c418159c8055910faa8047ffdb3d6b31098d2a46cd7685c", [:mix], [], "hexpm", "afc7cb142ebcc2d01ce7816190b98ce5dd49e799111b24249f3443d730f377ca"}, "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"}, + "iso8601": {:hex, :iso8601, "1.3.4", "7b1f095f86f6cf65e1e5a77872e8e8bf69bd58d4c3a415b3f77d9cc9423ecbb9", [:rebar3], [], "hexpm", "a334469c07f1c219326bc891a95f5eec8eb12dd8071a3fff56a7843cb20fae34"}, + "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"}, + "jsx": {:hex, :jsx, "3.1.0", "d12516baa0bb23a59bb35dccaf02a1bd08243fcbb9efe24f2d9d056ccff71268", [:rebar3], [], "hexpm", "0c5cc8fdc11b53cc25cf65ac6705ad39e54ecc56d1c22e4adb8f5a53fb9427f3"}, "makeup": {:hex, :makeup, "1.1.2", "9ba8837913bdf757787e71c1581c21f9d2455f4dd04cfca785c70bbfff1a76a3", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cce1566b81fbcbd21eca8ffe808f33b221f9eee2cbc7a1706fc3da9ff18e6cac"}, "makeup_elixir": {:hex, :makeup_elixir, "0.16.2", "627e84b8e8bf22e60a2579dad15067c755531fea049ae26ef1020cad58fe9578", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "41193978704763f6bbe6cc2758b84909e62984c7752b3784bd3c218bb341706b"}, "makeup_erlang": {:hex, :makeup_erlang, "1.0.0", "6f0eff9c9c489f26b69b61440bf1b238d95badae49adac77973cbacae87e3c2e", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "ea7a9307de9d1548d2a72d299058d1fd2339e3d398560a0e46c27dab4891e4d2"}, "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"}, - "mime": {:hex, :mime, "2.0.5", "dc34c8efd439abe6ae0343edbb8556f4d63f178594894720607772a041b04b02", [:mix], [], "hexpm", "da0d64a365c45bc9935cc5c8a7fc5e49a0e0f9932a761c55d6c52b142780a05c"}, + "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"}, "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm", "f278585650aa581986264638ebf698f8bb19df297f66ad91b18910dfc6e19323"}, - "nimble_options": {:hex, :nimble_options, "1.0.2", "92098a74df0072ff37d0c12ace58574d26880e522c22801437151a159392270e", [:mix], [], "hexpm", "fd12a8db2021036ce12a309f26f564ec367373265b53e25403f0ee697380f1b8"}, + "mint": {:hex, :mint, "1.9.3", "3337184d69179695c7a9f1714d92c11e629d36c8c037a21cf490131d3d150554", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "5f7c9342480c069dbbc4eeac3490303c9e01870ff01a7f1d29b6107054fc1e74"}, + "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"}, "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"}, + "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, "parse_trans": {:hex, :parse_trans, "3.3.1", "16328ab840cc09919bd10dab29e431da3af9e9e7e7e6f0089dd5a2d2820011d8", [:rebar3], [], "hexpm", "07cd9577885f56362d414e8c4c4e6bdf10d43a8767abb92d24cbe8b24c54888b"}, - "plug": {:hex, :plug, "1.14.2", "cff7d4ec45b4ae176a227acd94a7ab536d9b37b942c8e8fa6dfc0fff98ff4d80", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "842fc50187e13cf4ac3b253d47d9474ed6c296a8732752835ce4a86acdf68d13"}, + "plug": {:hex, :plug, "1.20.3", "56c480c633ec2ce10140e236e15233bf576e1d323887d7c96711bd02ab5160db", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "be266aee1b8536ef6409d58cf39a3121319f0ec47cfa1b24024485aa0e76ad76"}, "plug_cowboy": {:hex, :plug_cowboy, "2.6.1", "9a3bbfceeb65eff5f39dab529e5cd79137ac36e913c02067dba3963a26efe9b2", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "de36e1a21f451a18b790f37765db198075c25875c64834bcc82d90b309eb6613"}, - "plug_crypto": {:hex, :plug_crypto, "1.2.5", "918772575e48e81e455818229bf719d4ab4181fcbf7f85b68a35620f78d89ced", [:mix], [], "hexpm", "26549a1d6345e2172eb1c233866756ae44a9609bd33ee6f99147ab3fd87fd842"}, + "plug_crypto": {:hex, :plug_crypto, "2.2.0", "144014737daaf485407f5ed77daeaad74d651b216a28c87543f8cc7043f8efc8", [:mix], [], "hexpm", "83a95744ab1c75876542b6fab135fcc176280e0f301a111c1f757fddcec95d2c"}, "ranch": {:hex, :ranch, "1.8.0", "8c7a100a139fd57f17327b6413e4167ac559fbc04ca7448e9be9057311597a1d", [:make, :rebar3], [], "hexpm", "49fbcfd3682fab1f5d109351b61257676da1a2fdbe295904176d5e521a2ddfe5"}, + "req": {:hex, :req, "0.7.2", "364eae2e5f5c984f2dac6d71c07f8c8c89ce0bc49c4d746dacb7a306823020de", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:finch, "~> 0.21", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "c9cdfa276b05d8db2a27fda5d233e6858b764d47189d76cbb186e130a871ae0b"}, "saxy": {:hex, :saxy, "1.5.0", "0141127f2d042856f135fb2d94e0beecda7a2306f47546dbc6411fc5b07e28bf", [:mix], [], "hexpm", "ea7bb6328fbd1f2aceffa3ec6090bfb18c85aadf0f8e5030905e84235861cf89"}, "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"}, - "telemetry": {:hex, :telemetry, "1.2.1", "68fdfe8d8f05a8428483a97d7aab2f268aaff24b49e0f599faa091f1d4e7f61c", [:rebar3], [], "hexpm", "dad9ce9d8effc621708f99eac538ef1cbe05d6a874dd741de2e689c47feafed5"}, + "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.0", "bc84380c9ab48177092f43ac89e4dfa2c6d62b40b8bd132b1059ecc7232f9a78", [:rebar3], [], "hexpm", "25eee6d67df61960cf6a794239566599b09e17e668d3700247bc498638152521"}, } diff --git a/test/broadway_sqs/req_client/request_test.exs b/test/broadway_sqs/req_client/request_test.exs new file mode 100644 index 0000000..db9ee03 --- /dev/null +++ b/test/broadway_sqs/req_client/request_test.exs @@ -0,0 +1,84 @@ +defmodule BroadwaySQS.ReqClient.RequestTest do + use ExUnit.Case, async: true + + alias BroadwaySQS.ReqClient.Request + + @credentials [access_key_id: "access-key", secret_access_key: "secret-key"] + @request_opts [region: "eu-west-1", credentials: @credentials] + + test "sends a signed JSON SQS request and decodes the response" do + bypass = Bypass.open() + + Bypass.expect_once(bypass, "POST", "/", fn conn -> + {:ok, body, conn} = Plug.Conn.read_body(conn) + payload = Jason.decode!(body) + + assert Plug.Conn.get_req_header(conn, "content-type") == ["application/x-amz-json-1.0"] + assert Plug.Conn.get_req_header(conn, "x-amz-target") == ["AmazonSQS.ReceiveMessage"] + assert [authorization] = Plug.Conn.get_req_header(conn, "authorization") + assert String.starts_with?(authorization, "AWS4-HMAC-SHA256 ") + assert payload == %{"QueueUrl" => "http://localhost/queue"} + + Plug.Conn.resp(conn, 200, Jason.encode!(%{"Messages" => []})) + end) + + assert {:ok, %{"Messages" => []}} = + Request.call( + "AmazonSQS.ReceiveMessage", + %{"QueueUrl" => "http://localhost/queue"}, + Keyword.merge(@request_opts, endpoint: "http://localhost:#{bypass.port}") + ) + end + + test "sends a session token" do + bypass = Bypass.open() + + Bypass.expect_once(bypass, "POST", "/", fn conn -> + assert Plug.Conn.get_req_header(conn, "x-amz-security-token") == ["session-token"] + Plug.Conn.resp(conn, 200, Jason.encode!(%{})) + end) + + assert {:ok, %{}} = + Request.call( + "AmazonSQS.DeleteMessageBatch", + %{"QueueUrl" => "http://localhost/queue"}, + Keyword.merge(@request_opts, + endpoint: "http://localhost:#{bypass.port}", + credentials: Keyword.put(@credentials, :token, "session-token") + ) + ) + end + + test "returns decoded AWS errors for non-success responses" do + bypass = Bypass.open() + + Bypass.expect_once(bypass, "POST", "/", fn conn -> + Plug.Conn.resp(conn, 400, Jason.encode!(%{"__type" => "InvalidParameterValue"})) + end) + + assert {:error, {:http_error, 400, %{"__type" => "InvalidParameterValue"}}} = + Request.call( + "AmazonSQS.ChangeMessageVisibilityBatch", + %{"QueueUrl" => "http://localhost/queue"}, + Keyword.merge(@request_opts, endpoint: "http://localhost:#{bypass.port}") + ) + end + + test "returns an error when credentials are unavailable" do + assert {:error, :aws_credentials_not_found} = + Request.call( + "AmazonSQS.ReceiveMessage", + %{"QueueUrl" => "http://localhost/queue"}, + credentials: [] + ) + end + + test "returns an error when the AWS region is unavailable" do + assert {:error, :aws_region_not_found} = + Request.call( + "AmazonSQS.ReceiveMessage", + %{"QueueUrl" => "http://localhost/queue"}, + credentials: [access_key_id: "access-key", secret_access_key: "secret-key"] + ) + end +end From bd79cb48c486549b8e006fc8cd9503e1a457061f Mon Sep 17 00:00:00 2001 From: Milad Rastian Date: Wed, 12 Aug 2026 07:41:13 +0200 Subject: [PATCH 2/7] Add Req-based SQS client implementation --- lib/broadway_sqs/req_client.ex | 164 ++++++++++ lib/broadway_sqs/req_client/request.ex | 4 +- lib/broadway_sqs/req_client/sqs.ex | 129 ++++++++ test/broadway_sqs/req_client/sqs_test.exs | 161 ++++++++++ test/broadway_sqs/req_client_test.exs | 373 ++++++++++++++++++++++ 5 files changed, 830 insertions(+), 1 deletion(-) create mode 100644 lib/broadway_sqs/req_client.ex create mode 100644 lib/broadway_sqs/req_client/sqs.ex create mode 100644 test/broadway_sqs/req_client/sqs_test.exs create mode 100644 test/broadway_sqs/req_client_test.exs diff --git a/lib/broadway_sqs/req_client.ex b/lib/broadway_sqs/req_client.ex new file mode 100644 index 0000000..a79d599 --- /dev/null +++ b/lib/broadway_sqs/req_client.ex @@ -0,0 +1,164 @@ +defmodule BroadwaySQS.ReqClient do + @moduledoc """ + SQS client backed by `BroadwaySQS.ReqClient.Request`. + + This module adapts the SQS JSON API to the Broadway producer and + acknowledger behaviours. + """ + + alias Broadway.{Acknowledger, Message} + alias BroadwaySQS.ReqClient.SQS + require Logger + + @behaviour BroadwaySQS.SQSClient + @behaviour Acknowledger + + @max_num_messages_allowed_by_aws 10 + + @impl true + def init(opts) do + {:ok, Map.put(Map.new(opts), :ack_ref, opts[:broadway][:name])} + end + + @impl true + def receive_messages(demand, opts) do + receive_options = %{opts | max_number_of_messages: min(demand, opts.max_number_of_messages)} + + case SQS.receive_message(opts.queue_url, receive_options, request_options(opts)) do + {:ok, %{"Messages" => messages}} -> + wrap_received_messages(messages, opts.ack_ref) + + {:ok, _response} -> + [] + + {:error, reason} -> + Logger.error( + "Unable to fetch events from AWS queue #{opts.queue_url}. Reason: #{inspect(reason)}" + ) + + [] + end + end + + @impl Acknowledger + def ack(ack_ref, successful, failed) do + ack_options = :persistent_term.get(ack_ref) + + messages_to_delete = + Enum.filter(successful, &ack?(&1, ack_options, :on_success)) ++ + Enum.filter(failed, &ack?(&1, ack_options, :on_failure)) + + messages_to_nack_with_timeout = + Enum.flat_map(successful, &nack(&1, ack_options, :on_success)) ++ + Enum.flat_map(failed, &nack(&1, ack_options, :on_failure)) + + messages_to_delete + |> Enum.chunk_every(@max_num_messages_allowed_by_aws) + |> Enum.each(&delete_messages(&1, ack_options)) + + messages_to_nack_with_timeout + |> Enum.chunk_every(@max_num_messages_allowed_by_aws) + |> Enum.each(&change_message_visibilities(&1, ack_options)) + end + + @impl Acknowledger + def configure(_ack_ref, ack_data, options) do + {:ok, Map.merge(ack_data, Map.new(options))} + end + + defp ack?(message, ack_options, option) do + {_, _, message_ack_options} = message.acknowledger + (message_ack_options[option] || Map.fetch!(ack_options, option)) == :ack + end + + defp nack(message, ack_options, option) do + {_, _, message_ack_options} = message.acknowledger + + case message_ack_options[option] || Map.fetch!(ack_options, option) do + {:nack, timeout} -> [{message, timeout}] + _ -> [] + end + end + + defp delete_messages(messages, opts) do + entries = Enum.map(messages, &delete_entry/1) + request!(:delete_message_batch, [opts.queue_url, entries], opts) + end + + defp change_message_visibilities(messages_with_timeouts, opts) do + entries = + Enum.map(messages_with_timeouts, fn {message, timeout} -> + message + |> receipt() + |> Map.put("VisibilityTimeout", timeout) + end) + + request!(:change_message_visibility_batch, [opts.queue_url, entries], opts) + end + + defp request!(function, arguments, opts) do + case apply(SQS, function, arguments ++ [request_options(opts)]) do + {:ok, response} -> response + {:error, reason} -> raise "SQS request #{function} failed: #{inspect(reason)}" + end + end + + defp delete_entry(message) do + receipt = receipt(message) + %{"Id" => receipt["Id"], "ReceiptHandle" => receipt["ReceiptHandle"]} + end + + defp receipt(message) do + {_, _, %{receipt: receipt}} = message.acknowledger + %{"Id" => receipt.id, "ReceiptHandle" => receipt.receipt_handle} + end + + defp wrap_received_messages(messages, ack_ref) do + Enum.map(messages, fn message -> + message = SQS.normalize_message(message) + + %Message{ + data: message.data, + metadata: message.metadata, + acknowledger: build_acknowledger(message, ack_ref) + } + end) + end + + defp build_acknowledger(message, ack_ref) do + {__MODULE__, ack_ref, %{receipt: message.receipt}} + end + + defp request_options(opts) do + config = option_get(opts, :config, []) + credentials = credentials_from_config(config) + + [] + |> put_option(:region, option_get(config, :region)) + |> put_option(:credentials, credentials) + |> put_option(:endpoint, option_get(config, :endpoint)) + |> put_option(:plug, option_get(config, :plug)) + |> Enum.reject(fn {_key, value} -> is_nil(value) end) + end + + defp credentials_from_config(config) do + credentials = [ + access_key_id: option_get(config, :access_key_id), + secret_access_key: option_get(config, :secret_access_key), + token: option_get(config, :token) + ] + + if credentials[:access_key_id] && credentials[:secret_access_key], do: credentials + end + + defp put_option(options, _key, nil), do: options + defp put_option(options, key, value), do: Keyword.put(options, key, value) + + defp option_get(options, key, default \\ nil) + + defp option_get(options, key, default) when is_list(options), + do: Keyword.get(options, key, default) + + defp option_get(options, key, default) when is_map(options), + do: Map.get(options, key, default) +end diff --git a/lib/broadway_sqs/req_client/request.ex b/lib/broadway_sqs/req_client/request.ex index 3ccd9c5..7421c3a 100644 --- a/lib/broadway_sqs/req_client/request.ex +++ b/lib/broadway_sqs/req_client/request.ex @@ -6,6 +6,7 @@ defmodule BroadwaySQS.ReqClient.Request do @type options :: [ credentials: map() | keyword(), endpoint: String.t(), + plug: term(), region: String.t(), queue_url: String.t() ] @@ -29,7 +30,7 @@ defmodule BroadwaySQS.ReqClient.Request do decode_body: false ), {:ok, body} <- decode_body(response) do - if response.status >= 200 and response.status < 300 do + if response.status in 200..299 do {:ok, body} else {:error, {:http_error, response.status, body}} @@ -50,6 +51,7 @@ defmodule BroadwaySQS.ReqClient.Request do Req.new( url: Keyword.get(opts, :endpoint, queue_url), + plug: Keyword.get(opts, :plug), headers: headers, aws_sigv4: [ access_key_id: credentials[:access_key_id], diff --git a/lib/broadway_sqs/req_client/sqs.ex b/lib/broadway_sqs/req_client/sqs.ex new file mode 100644 index 0000000..345dd7f --- /dev/null +++ b/lib/broadway_sqs/req_client/sqs.ex @@ -0,0 +1,129 @@ +defmodule BroadwaySQS.ReqClient.SQS do + @moduledoc false + + alias BroadwaySQS.ReqClient.Request + + def receive_message(queue_url, options, request_options) do + payload = + %{ + "QueueUrl" => queue_url, + "MaxNumberOfMessages" => options.max_number_of_messages + } + |> put_if_present("WaitTimeSeconds", options[:wait_time_seconds]) + |> put_if_present("VisibilityTimeout", options[:visibility_timeout]) + |> put_if_present("AttributeNames", attribute_names(options[:attribute_names])) + |> put_if_present("MessageAttributeNames", options[:message_attribute_names]) + + Request.call("AmazonSQS.ReceiveMessage", payload, request_options) + end + + def delete_message_batch(queue_url, entries, request_options) do + Request.call( + "AmazonSQS.DeleteMessageBatch", + %{"QueueUrl" => queue_url, "Entries" => entries}, + request_options + ) + end + + def change_message_visibility_batch(queue_url, entries, request_options) do + Request.call( + "AmazonSQS.ChangeMessageVisibilityBatch", + %{"QueueUrl" => queue_url, "Entries" => entries}, + request_options + ) + end + + def normalize_message(message) do + %{ + data: message["Body"], + metadata: message_metadata(message), + receipt: %{ + id: message["MessageId"], + receipt_handle: message["ReceiptHandle"] + } + } + end + + defp put_if_present(payload, _key, nil), do: payload + defp put_if_present(payload, key, value), do: Map.put(payload, key, value) + + defp attribute_names(nil), do: nil + defp attribute_names(:all), do: ["All"] + defp attribute_names(names), do: Enum.map(names, &attribute_name/1) + + defp attribute_name(:sender_id), do: "SenderId" + defp attribute_name(:sent_timestamp), do: "SentTimestamp" + defp attribute_name(:approximate_receive_count), do: "ApproximateReceiveCount" + + defp attribute_name(:approximate_first_receive_timestamp), + do: "ApproximateFirstReceiveTimestamp" + + defp attribute_name(:sequence_number), do: "SequenceNumber" + defp attribute_name(:message_deduplication_id), do: "MessageDeduplicationId" + defp attribute_name(:message_group_id), do: "MessageGroupId" + defp attribute_name(:aws_trace_header), do: "AWSTraceHeader" + defp attribute_name(name) when is_binary(name), do: name + + defp message_metadata(message) do + %{ + message_id: message["MessageId"], + receipt_handle: message["ReceiptHandle"], + md5_of_body: message["MD5OfBody"], + attributes: convert_attributes(message["Attributes"]), + message_attributes: convert_message_attributes(message["MessageAttributes"]) + } + |> Enum.reject(fn {_key, value} -> is_nil(value) end) + |> Map.new() + end + + defp convert_attributes(nil), do: [] + + defp convert_attributes(attributes) when is_map(attributes) do + Map.new(attributes, fn {name, value} -> + {metadata_attribute_name(name), parse_integer(value)} + end) + end + + defp convert_attributes(attributes), do: attributes + + defp convert_message_attributes(nil), do: [] + + defp convert_message_attributes(attributes) when is_map(attributes) do + Map.new(attributes, fn {name, attribute} -> + value = attribute["StringValue"] || attribute["BinaryValue"] || "" + + {name, + %{ + name: name, + data_type: attribute["DataType"], + string_value: attribute["StringValue"] || "", + binary_value: attribute["BinaryValue"] || "", + value: value + }} + end) + end + + defp convert_message_attributes(attributes), do: attributes + + defp metadata_attribute_name("SenderId"), do: "sender_id" + defp metadata_attribute_name("SentTimestamp"), do: "sent_timestamp" + defp metadata_attribute_name("ApproximateReceiveCount"), do: "approximate_receive_count" + + defp metadata_attribute_name("ApproximateFirstReceiveTimestamp"), + do: "approximate_first_receive_timestamp" + + defp metadata_attribute_name("SequenceNumber"), do: "sequence_number" + defp metadata_attribute_name("MessageDeduplicationId"), do: "message_deduplication_id" + defp metadata_attribute_name("MessageGroupId"), do: "message_group_id" + defp metadata_attribute_name("AWSTraceHeader"), do: "aws_trace_header" + defp metadata_attribute_name(name), do: name + + defp parse_integer(value) when is_binary(value) do + case Integer.parse(value) do + {integer, ""} -> integer + _ -> value + end + end + + defp parse_integer(value), do: value +end diff --git a/test/broadway_sqs/req_client/sqs_test.exs b/test/broadway_sqs/req_client/sqs_test.exs new file mode 100644 index 0000000..e156a89 --- /dev/null +++ b/test/broadway_sqs/req_client/sqs_test.exs @@ -0,0 +1,161 @@ +defmodule BroadwaySQS.ReqClient.SQSTest do + use ExUnit.Case, async: true + + alias BroadwaySQS.ReqClient.SQS + + @request_options [ + region: "eu-west-1", + credentials: [access_key_id: "access-key", secret_access_key: "secret-key"] + ] + + test "receive_message builds the SQS JSON payload" do + bypass = Bypass.open() + queue_url = "http://localhost:#{bypass.port}/queue" + + Bypass.expect_once(bypass, "POST", "/queue", fn conn -> + {:ok, body, conn} = Plug.Conn.read_body(conn) + + assert Jason.decode!(body) == %{ + "QueueUrl" => queue_url, + "MaxNumberOfMessages" => 2, + "WaitTimeSeconds" => 10, + "VisibilityTimeout" => 30, + "AttributeNames" => ["ApproximateReceiveCount", "SenderId"], + "MessageAttributeNames" => ["TestAttribute"] + } + + assert Plug.Conn.get_req_header(conn, "x-amz-target") == ["AmazonSQS.ReceiveMessage"] + Plug.Conn.resp(conn, 200, Jason.encode!(%{"Messages" => []})) + end) + + options = %{ + max_number_of_messages: 2, + wait_time_seconds: 10, + visibility_timeout: 30, + attribute_names: [:approximate_receive_count, :sender_id], + message_attribute_names: ["TestAttribute"] + } + + assert {:ok, %{"Messages" => []}} = + SQS.receive_message( + queue_url, + options, + Keyword.put(@request_options, :endpoint, queue_url) + ) + end + + test "receive_message supports requesting all attributes" do + bypass = Bypass.open() + queue_url = "http://localhost:#{bypass.port}/queue" + + Bypass.expect_once(bypass, "POST", "/queue", fn conn -> + {:ok, body, conn} = Plug.Conn.read_body(conn) + assert Jason.decode!(body)["AttributeNames"] == ["All"] + Plug.Conn.resp(conn, 200, Jason.encode!(%{"Messages" => []})) + end) + + options = %{max_number_of_messages: 1, attribute_names: :all} + + assert {:ok, %{"Messages" => []}} = + SQS.receive_message( + queue_url, + options, + Keyword.put(@request_options, :endpoint, queue_url) + ) + end + + test "delete_message_batch builds the delete payload" do + bypass = Bypass.open() + queue_url = "http://localhost:#{bypass.port}/queue" + entries = [%{"Id" => "1", "ReceiptHandle" => "receipt-1"}] + + Bypass.expect_once(bypass, "POST", "/queue", fn conn -> + {:ok, body, conn} = Plug.Conn.read_body(conn) + + assert Jason.decode!(body) == %{"QueueUrl" => queue_url, "Entries" => entries} + assert Plug.Conn.get_req_header(conn, "x-amz-target") == ["AmazonSQS.DeleteMessageBatch"] + Plug.Conn.resp(conn, 200, Jason.encode!(%{"Successful" => [%{"Id" => "1"}]})) + end) + + assert {:ok, %{"Successful" => [%{"Id" => "1"}]}} = + SQS.delete_message_batch( + queue_url, + entries, + Keyword.put(@request_options, :endpoint, queue_url) + ) + end + + test "change_message_visibility_batch builds the visibility payload" do + bypass = Bypass.open() + queue_url = "http://localhost:#{bypass.port}/queue" + + entries = [ + %{ + "Id" => "1", + "ReceiptHandle" => "receipt-1", + "VisibilityTimeout" => 12 + } + ] + + Bypass.expect_once(bypass, "POST", "/queue", fn conn -> + {:ok, body, conn} = Plug.Conn.read_body(conn) + + assert Jason.decode!(body) == %{"QueueUrl" => queue_url, "Entries" => entries} + + assert Plug.Conn.get_req_header(conn, "x-amz-target") == [ + "AmazonSQS.ChangeMessageVisibilityBatch" + ] + + Plug.Conn.resp(conn, 200, Jason.encode!(%{"Successful" => [%{"Id" => "1"}]})) + end) + + assert {:ok, %{"Successful" => [%{"Id" => "1"}]}} = + SQS.change_message_visibility_batch( + queue_url, + entries, + Keyword.put(@request_options, :endpoint, queue_url) + ) + end + + test "normalize_message converts SQS messages to the client format" do + message = %{ + "MessageId" => "message-id", + "ReceiptHandle" => "receipt-handle", + "MD5OfBody" => "body-md5", + "Body" => "hello", + "Attributes" => %{ + "ApproximateReceiveCount" => "5", + "SenderId" => "sender" + }, + "MessageAttributes" => %{ + "TestAttribute" => %{ + "StringValue" => "test", + "DataType" => "String" + } + } + } + + assert SQS.normalize_message(message) == %{ + data: "hello", + metadata: %{ + message_id: "message-id", + receipt_handle: "receipt-handle", + md5_of_body: "body-md5", + attributes: %{ + "approximate_receive_count" => 5, + "sender_id" => "sender" + }, + message_attributes: %{ + "TestAttribute" => %{ + name: "TestAttribute", + data_type: "String", + string_value: "test", + binary_value: "", + value: "test" + } + } + }, + receipt: %{id: "message-id", receipt_handle: "receipt-handle"} + } + end +end diff --git a/test/broadway_sqs/req_client_test.exs b/test/broadway_sqs/req_client_test.exs new file mode 100644 index 0000000..8d74749 --- /dev/null +++ b/test/broadway_sqs/req_client_test.exs @@ -0,0 +1,373 @@ +defmodule BroadwaySQS.ReqClientTest do + use ExUnit.Case, async: true + + import ExUnit.CaptureLog + + alias Broadway.Message + alias BroadwaySQS.ReqClient + + @config [ + access_key_id: "access-key", + secret_access_key: "secret-key", + region: "eu-west-1" + ] + + setup do + Req.Test.set_req_test_from_context(__MODULE__) + Req.Test.stub(__MODULE__, fn conn -> Req.Test.json(conn, %{}) end) + :ok + end + + test "receives messages and preserves Broadway metadata" do + queue_url = queue_url() + + Req.Test.expect(__MODULE__, fn conn -> + payload = Jason.decode!(Req.Test.raw_body(conn)) + + assert payload == %{ + "QueueUrl" => queue_url, + "MaxNumberOfMessages" => 2, + "WaitTimeSeconds" => 10, + "VisibilityTimeout" => 30, + "AttributeNames" => ["ApproximateReceiveCount"], + "MessageAttributeNames" => ["TestAttribute"] + } + + response = %{ + "Messages" => [ + %{ + "MessageId" => "message-id", + "ReceiptHandle" => "receipt-handle", + "MD5OfBody" => "body-md5", + "Body" => "hello", + "Attributes" => %{"ApproximateReceiveCount" => "5"}, + "MessageAttributes" => %{ + "TestAttribute" => %{ + "StringValue" => "test", + "DataType" => "String" + } + } + } + ] + } + + Req.Test.json(conn, response) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url)) + [message] = ReqClient.receive_messages(2, opts) + + assert %Message{data: "hello"} = message + assert message.metadata.message_id == "message-id" + assert message.metadata.receipt_handle == "receipt-handle" + assert message.metadata.md5_of_body == "body-md5" + assert message.metadata.attributes == %{"approximate_receive_count" => 5} + + assert message.metadata.message_attributes == %{ + "TestAttribute" => %{ + name: "TestAttribute", + data_type: "String", + string_value: "test", + binary_value: "", + value: "test" + } + } + end + + test "logs receive errors" do + queue_url = queue_url() + + Req.Test.expect(__MODULE__, fn conn -> + conn + |> Plug.Conn.put_status(500) + |> Req.Test.json(%{"message" => "failure"}) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url)) + + log = capture_log(fn -> assert ReqClient.receive_messages(1, opts) == [] end) + assert log =~ "Unable to fetch events from AWS queue #{queue_url}" + end + + test "acknowledges messages in delete batches" do + queue_url = queue_url() + test_pid = self() + + Req.Test.expect(__MODULE__, fn conn -> + body = Req.Test.raw_body(conn) + assert Plug.Conn.get_req_header(conn, "x-amz-target") == ["AmazonSQS.DeleteMessageBatch"] + + send(test_pid, {:delete_payload, Jason.decode!(body)}) + Req.Test.json(conn, %{"Successful" => []}) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url, on_success: :ack)) + put_ack_options(opts) + + message = message(opts.ack_ref, "1", "receipt-1") + ReqClient.ack(opts.ack_ref, [message], []) + + assert_received {:delete_payload, + %{ + "QueueUrl" => ^queue_url, + "Entries" => [%{"Id" => "1", "ReceiptHandle" => "receipt-1"}] + }} + end + + test "nacks messages by changing their visibility" do + queue_url = queue_url() + test_pid = self() + + Req.Test.expect(__MODULE__, fn conn -> + body = Req.Test.raw_body(conn) + + assert Plug.Conn.get_req_header(conn, "x-amz-target") == [ + "AmazonSQS.ChangeMessageVisibilityBatch" + ] + + send(test_pid, {:visibility_payload, Jason.decode!(body)}) + Req.Test.json(conn, %{"Successful" => []}) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url, on_failure: {:nack, 12})) + put_ack_options(opts) + + message = message(opts.ack_ref, "1", "receipt-1") + ReqClient.ack(opts.ack_ref, [], [message]) + + assert_received {:visibility_payload, + %{ + "QueueUrl" => ^queue_url, + "Entries" => [ + %{ + "Id" => "1", + "ReceiptHandle" => "receipt-1", + "VisibilityTimeout" => 12 + } + ] + }} + end + + test "per-message acknowledgement options override producer defaults" do + queue_url = queue_url() + test_pid = self() + + Req.Test.expect(__MODULE__, 2, fn conn -> + body = Req.Test.raw_body(conn) + send(test_pid, {Plug.Conn.get_req_header(conn, "x-amz-target"), Jason.decode!(body)}) + Req.Test.json(conn, %{"Successful" => []}) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url, on_success: :noop, on_failure: :noop)) + put_ack_options(opts) + + success = message(opts.ack_ref, "success", "success-receipt") + failure = message(opts.ack_ref, "failure", "failure-receipt") + + success = Message.configure_ack(success, on_success: :ack) + failure = Message.configure_ack(failure, on_failure: {:nack, 7}) + + ReqClient.ack(opts.ack_ref, [success], [failure]) + + assert_received { + ["AmazonSQS.DeleteMessageBatch"], + %{"Entries" => [%{"Id" => "success"}]} + } + + assert_received { + ["AmazonSQS.ChangeMessageVisibilityBatch"], + %{"Entries" => [%{"Id" => "failure", "VisibilityTimeout" => 7}]} + } + end + + test "converts all supported SQS attributes" do + queue_url = queue_url() + + Req.Test.expect(__MODULE__, fn conn -> + response = %{ + "Messages" => [ + %{ + "MessageId" => "id", + "ReceiptHandle" => "receipt", + "Body" => "body", + "Attributes" => %{ + "SenderId" => "sender", + "SentTimestamp" => "123", + "ApproximateReceiveCount" => "5", + "ApproximateFirstReceiveTimestamp" => "456", + "SequenceNumber" => "sequence", + "MessageDeduplicationId" => "deduplication", + "MessageGroupId" => "group", + "AWSTraceHeader" => "trace" + } + } + ] + } + + Req.Test.json(conn, response) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url, attribute_names: :all)) + [message] = ReqClient.receive_messages(1, opts) + + assert message.metadata.attributes == %{ + "sender_id" => "sender", + "sent_timestamp" => 123, + "approximate_receive_count" => 5, + "approximate_first_receive_timestamp" => 456, + "sequence_number" => "sequence", + "message_deduplication_id" => "deduplication", + "message_group_id" => "group", + "aws_trace_header" => "trace" + } + end + + test "handles empty and missing Messages responses" do + queue_url = queue_url() + + Req.Test.expect(__MODULE__, 2, fn conn -> + response = if Req.Test.raw_body(conn) == "missing", do: %{}, else: %{"Messages" => []} + Req.Test.json(conn, response) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url)) + assert ReqClient.receive_messages(1, opts) == [] + + assert ReqClient.receive_messages(1, opts) == [] + end + + test "caps receive demand at ten messages" do + queue_url = queue_url() + + Req.Test.expect(__MODULE__, fn conn -> + body = Req.Test.raw_body(conn) + assert Jason.decode!(body)["MaxNumberOfMessages"] == 10 + Req.Test.json(conn, %{"Messages" => []}) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url)) + assert ReqClient.receive_messages(100, opts) == [] + end + + test "omits unset receive options" do + queue_url = queue_url() + + Req.Test.expect(__MODULE__, fn conn -> + body = Req.Test.raw_body(conn) + + assert Jason.decode!(body) == %{ + "QueueUrl" => queue_url, + "MaxNumberOfMessages" => 1 + } + + Req.Test.json(conn, %{"Messages" => []}) + end) + + {:ok, opts} = + ReqClient.init( + opts(queue_url, + wait_time_seconds: nil, + visibility_timeout: nil, + attribute_names: nil, + message_attribute_names: nil + ) + ) + + assert ReqClient.receive_messages(1, opts) == [] + end + + test "splits delete acknowledgements into batches of ten" do + queue_url = queue_url() + test_pid = self() + + Req.Test.expect(__MODULE__, 2, fn conn -> + body = Req.Test.raw_body(conn) + send(test_pid, {:delete_batch, Jason.decode!(body)["Entries"]}) + Req.Test.json(conn, %{"Successful" => []}) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url)) + put_ack_options(opts) + + messages = Enum.map(1..11, &message(opts.ack_ref, to_string(&1), "receipt-#{&1}")) + ReqClient.ack(opts.ack_ref, messages, []) + + assert_received {:delete_batch, first_batch} + assert_received {:delete_batch, second_batch} + assert length(first_batch) == 10 + assert length(second_batch) == 1 + end + + test "splits visibility changes into batches of ten" do + queue_url = queue_url() + test_pid = self() + + Req.Test.expect(__MODULE__, 2, fn conn -> + body = Req.Test.raw_body(conn) + send(test_pid, {:visibility_batch, Jason.decode!(body)["Entries"]}) + Req.Test.json(conn, %{"Successful" => []}) + end) + + {:ok, opts} = ReqClient.init(opts(queue_url, on_failure: {:nack, 3})) + put_ack_options(opts) + + messages = Enum.map(1..11, &message(opts.ack_ref, to_string(&1), "receipt-#{&1}")) + ReqClient.ack(opts.ack_ref, [], messages) + + assert_received {:visibility_batch, first_batch} + assert_received {:visibility_batch, second_batch} + assert length(first_batch) == 10 + assert length(second_batch) == 1 + end + + test "configure merges acknowledgement options" do + ack_data = %{receipt: %{id: "id", receipt_handle: "receipt"}, on_success: :noop} + + assert {:ok, configured} = + ReqClient.configure(:ack_ref, ack_data, on_success: :ack, on_failure: :noop) + + assert configured == %{ + receipt: %{id: "id", receipt_handle: "receipt"}, + on_success: :ack, + on_failure: :noop + } + end + + defp opts(queue_url, overrides \\ []) do + Keyword.merge( + [ + broadway: [name: unique_ack_ref()], + queue_url: queue_url, + config: Keyword.put(@config, :plug, {Req.Test, __MODULE__}), + on_success: :ack, + on_failure: :noop, + max_number_of_messages: 10, + wait_time_seconds: 10, + visibility_timeout: 30, + attribute_names: [:approximate_receive_count], + message_attribute_names: ["TestAttribute"] + ], + overrides + ) + end + + defp message(ack_ref, id, receipt_handle) do + %Message{ + data: "data", + acknowledger: {ReqClient, ack_ref, %{receipt: %{id: id, receipt_handle: receipt_handle}}} + } + end + + defp put_ack_options(opts) do + :persistent_term.put(opts.ack_ref, %{ + queue_url: opts.queue_url, + config: opts.config, + on_success: opts.on_success, + on_failure: opts.on_failure + }) + end + + defp unique_ack_ref, do: {__MODULE__, make_ref()} + + defp queue_url, do: "https://sqs.eu-west-1.amazonaws.com/123456789012/test-queue" +end From 50e2da6b40165cd4a0d1aac6e56bbe4bcae6e5a0 Mon Sep 17 00:00:00 2001 From: Milad Rastian Date: Wed, 12 Aug 2026 09:28:12 +0200 Subject: [PATCH 3/7] refactor test to use Req.Test --- lib/broadway_sqs/req_client.ex | 20 +-- test/broadway_sqs/integration_test.exs | 116 ++++++------------ test/broadway_sqs/req_client/request_test.exs | 84 ------------- test/broadway_sqs/req_client/sqs_test.exs | 51 ++++---- test/broadway_sqs/req_client_test.exs | 60 ++++++++- 5 files changed, 134 insertions(+), 197 deletions(-) delete mode 100644 test/broadway_sqs/req_client/request_test.exs diff --git a/lib/broadway_sqs/req_client.ex b/lib/broadway_sqs/req_client.ex index a79d599..395bb6c 100644 --- a/lib/broadway_sqs/req_client.ex +++ b/lib/broadway_sqs/req_client.ex @@ -82,7 +82,11 @@ defmodule BroadwaySQS.ReqClient do defp delete_messages(messages, opts) do entries = Enum.map(messages, &delete_entry/1) - request!(:delete_message_batch, [opts.queue_url, entries], opts) + + request!( + SQS.delete_message_batch(opts.queue_url, entries, request_options(opts)), + :delete_message_batch + ) end defp change_message_visibilities(messages_with_timeouts, opts) do @@ -93,14 +97,16 @@ defmodule BroadwaySQS.ReqClient do |> Map.put("VisibilityTimeout", timeout) end) - request!(:change_message_visibility_batch, [opts.queue_url, entries], opts) + request!( + SQS.change_message_visibility_batch(opts.queue_url, entries, request_options(opts)), + :change_message_visibility_batch + ) end - defp request!(function, arguments, opts) do - case apply(SQS, function, arguments ++ [request_options(opts)]) do - {:ok, response} -> response - {:error, reason} -> raise "SQS request #{function} failed: #{inspect(reason)}" - end + defp request!({:ok, response}, _function), do: response + + defp request!({:error, reason}, function) do + raise "SQS request #{function} failed: #{inspect(reason)}" end defp delete_entry(message) do diff --git a/test/broadway_sqs/integration_test.exs b/test/broadway_sqs/integration_test.exs index c89ced0..a26a16d 100644 --- a/test/broadway_sqs/integration_test.exs +++ b/test/broadway_sqs/integration_test.exs @@ -23,56 +23,22 @@ defmodule BroadwaySQS.BroadwaySQS.IntegrationTest do end end - @receive_message_response """ - - - - - 7cd4d61a-2d9a-4922-9738-308af6126fea - receipt-handle-1 - 8cd6cfc2639481fee178bd04dd3628a7 - hello world - - - c431bcb8-3275-4cbb-a4a1-7bcbbc5773d1 - receipt-handle-2 - 35179a54ea587953021400eb0cd23201 - how are you? - - - - 251d6374-3eac-5128-b100-3b06e7db493a - - - """ - - @receive_message_empty_response """ - - - - - - 251d6374-3eac-5128-b100-3b06e7db493a - - - """ - - @delete_message_response """ - - - - - my-delete-message-batch-id-1 - - - my-delete-message-batch-id-2 - - - - 1164da49-dd83-5d9d-acaa-823b38f2d2f8 - - - """ + @receive_message_response %{ + "Messages" => [ + %{ + "MessageId" => "7cd4d61a-2d9a-4922-9738-308af6126fea", + "ReceiptHandle" => "receipt-handle-1", + "MD5OfBody" => "8cd6cfc2639481fee178bd04dd3628a7", + "Body" => "hello world" + }, + %{ + "MessageId" => "c431bcb8-3275-4cbb-a4a1-7bcbbc5773d1", + "ReceiptHandle" => "receipt-handle-2", + "MD5OfBody" => "35179a54ea587953021400eb0cd23201", + "Body" => "how are you?" + } + ] + } defmodule RequestCounter do use Agent @@ -95,49 +61,40 @@ defmodule BroadwaySQS.BroadwaySQS.IntegrationTest do end setup do - bypass = Bypass.open() - - Application.put_env(:ex_aws, :sqs, - scheme: "http", - host: "localhost", - port: bypass.port - ) - - on_exit(fn -> Application.delete_env(:ex_aws, :sqs) end) - - {:ok, bypass: bypass} + Req.Test.set_req_test_to_shared() + :ok end - test "consume messages from SQS and ack it", %{bypass: bypass} do + test "consume messages from SQS and ack it" do us = self() - Bypass.expect(bypass, fn conn -> - {:ok, body, conn} = Plug.Conn.read_body(conn) + Req.Test.expect(__MODULE__, 9, fn conn -> + payload = Jason.decode!(Req.Test.raw_body(conn)) + action = List.first(Conn.get_req_header(conn, "x-amz-target")) response = - case body do - "Action=ReceiveMessage" <> _rest -> + case action do + "AmazonSQS.ReceiveMessage" -> if RequestCounter.count_for(:receive_message) > 5 do - @receive_message_empty_response + %{} else RequestCounter.increment_for(:receive_message) @receive_message_response end - "Action=DeleteMessageBatch" <> _rest -> + "AmazonSQS.DeleteMessageBatch" -> + assert payload["Entries"] RequestCounter.increment_for(:delete_message_batch) send(us, :messages_deleted) - @delete_message_response + %{"Successful" => Enum.map(payload["Entries"], &Map.take(&1, ["Id"]))} end - conn - |> Conn.put_resp_header("content-type", "text/xml") - |> Conn.resp(200, response) + Req.Test.json(conn, response) end) {:ok, _} = RequestCounter.start_link(%{receive_message: 0, delete_message_batch: 0}) - {:ok, _consumer} = start_fake_consumer(bypass) + {:ok, _consumer} = start_fake_consumer() assert_receive {:message_handled, "hello world", %{receipt_handle: "receipt-handle-1"}}, 1_000 assert_receive {:message_handled, "how are you?", %{receipt_handle: "receipt-handle-2"}} @@ -152,20 +109,21 @@ defmodule BroadwaySQS.BroadwaySQS.IntegrationTest do assert RequestCounter.count_for(:delete_message_batch) == 3 end - defp start_fake_consumer(bypass) do + defp start_fake_consumer do Broadway.start_link(MyConsumer, name: MyConsumer, producer: [ module: {BroadwaySQS.Producer, - sqs_client: BroadwaySQS.ExAwsClient, + sqs_client: BroadwaySQS.ReqClient, max_number_of_messages: 2, config: [ access_key_id: "MY_AWS_ACCESS_KEY_ID", secret_access_key: "MY_AWS_SECRET_ACCESS_KEY", - region: "us-east-2" + region: "us-east-2", + plug: {Req.Test, __MODULE__} ], - queue_url: queue_endpoint_url(bypass)}, + queue_url: queue_endpoint_url()}, concurrency: 1 ], processors: [ @@ -178,7 +136,5 @@ defmodule BroadwaySQS.BroadwaySQS.IntegrationTest do ) end - defp queue_endpoint_url(bypass) do - "http://localhost:#{bypass.port}/my_queue" - end + defp queue_endpoint_url, do: "https://sqs.us-east-2.amazonaws.com/123456789012/my_queue" end diff --git a/test/broadway_sqs/req_client/request_test.exs b/test/broadway_sqs/req_client/request_test.exs deleted file mode 100644 index db9ee03..0000000 --- a/test/broadway_sqs/req_client/request_test.exs +++ /dev/null @@ -1,84 +0,0 @@ -defmodule BroadwaySQS.ReqClient.RequestTest do - use ExUnit.Case, async: true - - alias BroadwaySQS.ReqClient.Request - - @credentials [access_key_id: "access-key", secret_access_key: "secret-key"] - @request_opts [region: "eu-west-1", credentials: @credentials] - - test "sends a signed JSON SQS request and decodes the response" do - bypass = Bypass.open() - - Bypass.expect_once(bypass, "POST", "/", fn conn -> - {:ok, body, conn} = Plug.Conn.read_body(conn) - payload = Jason.decode!(body) - - assert Plug.Conn.get_req_header(conn, "content-type") == ["application/x-amz-json-1.0"] - assert Plug.Conn.get_req_header(conn, "x-amz-target") == ["AmazonSQS.ReceiveMessage"] - assert [authorization] = Plug.Conn.get_req_header(conn, "authorization") - assert String.starts_with?(authorization, "AWS4-HMAC-SHA256 ") - assert payload == %{"QueueUrl" => "http://localhost/queue"} - - Plug.Conn.resp(conn, 200, Jason.encode!(%{"Messages" => []})) - end) - - assert {:ok, %{"Messages" => []}} = - Request.call( - "AmazonSQS.ReceiveMessage", - %{"QueueUrl" => "http://localhost/queue"}, - Keyword.merge(@request_opts, endpoint: "http://localhost:#{bypass.port}") - ) - end - - test "sends a session token" do - bypass = Bypass.open() - - Bypass.expect_once(bypass, "POST", "/", fn conn -> - assert Plug.Conn.get_req_header(conn, "x-amz-security-token") == ["session-token"] - Plug.Conn.resp(conn, 200, Jason.encode!(%{})) - end) - - assert {:ok, %{}} = - Request.call( - "AmazonSQS.DeleteMessageBatch", - %{"QueueUrl" => "http://localhost/queue"}, - Keyword.merge(@request_opts, - endpoint: "http://localhost:#{bypass.port}", - credentials: Keyword.put(@credentials, :token, "session-token") - ) - ) - end - - test "returns decoded AWS errors for non-success responses" do - bypass = Bypass.open() - - Bypass.expect_once(bypass, "POST", "/", fn conn -> - Plug.Conn.resp(conn, 400, Jason.encode!(%{"__type" => "InvalidParameterValue"})) - end) - - assert {:error, {:http_error, 400, %{"__type" => "InvalidParameterValue"}}} = - Request.call( - "AmazonSQS.ChangeMessageVisibilityBatch", - %{"QueueUrl" => "http://localhost/queue"}, - Keyword.merge(@request_opts, endpoint: "http://localhost:#{bypass.port}") - ) - end - - test "returns an error when credentials are unavailable" do - assert {:error, :aws_credentials_not_found} = - Request.call( - "AmazonSQS.ReceiveMessage", - %{"QueueUrl" => "http://localhost/queue"}, - credentials: [] - ) - end - - test "returns an error when the AWS region is unavailable" do - assert {:error, :aws_region_not_found} = - Request.call( - "AmazonSQS.ReceiveMessage", - %{"QueueUrl" => "http://localhost/queue"}, - credentials: [access_key_id: "access-key", secret_access_key: "secret-key"] - ) - end -end diff --git a/test/broadway_sqs/req_client/sqs_test.exs b/test/broadway_sqs/req_client/sqs_test.exs index e156a89..5f1587a 100644 --- a/test/broadway_sqs/req_client/sqs_test.exs +++ b/test/broadway_sqs/req_client/sqs_test.exs @@ -1,6 +1,11 @@ defmodule BroadwaySQS.ReqClient.SQSTest do use ExUnit.Case, async: true + setup context do + Req.Test.set_req_test_from_context(context) + :ok + end + alias BroadwaySQS.ReqClient.SQS @request_options [ @@ -9,11 +14,10 @@ defmodule BroadwaySQS.ReqClient.SQSTest do ] test "receive_message builds the SQS JSON payload" do - bypass = Bypass.open() - queue_url = "http://localhost:#{bypass.port}/queue" + queue_url = queue_url() - Bypass.expect_once(bypass, "POST", "/queue", fn conn -> - {:ok, body, conn} = Plug.Conn.read_body(conn) + Req.Test.expect(__MODULE__, fn conn -> + body = Req.Test.raw_body(conn) assert Jason.decode!(body) == %{ "QueueUrl" => queue_url, @@ -25,7 +29,7 @@ defmodule BroadwaySQS.ReqClient.SQSTest do } assert Plug.Conn.get_req_header(conn, "x-amz-target") == ["AmazonSQS.ReceiveMessage"] - Plug.Conn.resp(conn, 200, Jason.encode!(%{"Messages" => []})) + Req.Test.json(conn, %{"Messages" => []}) end) options = %{ @@ -40,18 +44,17 @@ defmodule BroadwaySQS.ReqClient.SQSTest do SQS.receive_message( queue_url, options, - Keyword.put(@request_options, :endpoint, queue_url) + Keyword.merge(@request_options, endpoint: queue_url, plug: {Req.Test, __MODULE__}) ) end test "receive_message supports requesting all attributes" do - bypass = Bypass.open() - queue_url = "http://localhost:#{bypass.port}/queue" + queue_url = queue_url() - Bypass.expect_once(bypass, "POST", "/queue", fn conn -> - {:ok, body, conn} = Plug.Conn.read_body(conn) + Req.Test.expect(__MODULE__, fn conn -> + body = Req.Test.raw_body(conn) assert Jason.decode!(body)["AttributeNames"] == ["All"] - Plug.Conn.resp(conn, 200, Jason.encode!(%{"Messages" => []})) + Req.Test.json(conn, %{"Messages" => []}) end) options = %{max_number_of_messages: 1, attribute_names: :all} @@ -60,34 +63,32 @@ defmodule BroadwaySQS.ReqClient.SQSTest do SQS.receive_message( queue_url, options, - Keyword.put(@request_options, :endpoint, queue_url) + Keyword.merge(@request_options, endpoint: queue_url, plug: {Req.Test, __MODULE__}) ) end test "delete_message_batch builds the delete payload" do - bypass = Bypass.open() - queue_url = "http://localhost:#{bypass.port}/queue" + queue_url = queue_url() entries = [%{"Id" => "1", "ReceiptHandle" => "receipt-1"}] - Bypass.expect_once(bypass, "POST", "/queue", fn conn -> - {:ok, body, conn} = Plug.Conn.read_body(conn) + Req.Test.expect(__MODULE__, fn conn -> + body = Req.Test.raw_body(conn) assert Jason.decode!(body) == %{"QueueUrl" => queue_url, "Entries" => entries} assert Plug.Conn.get_req_header(conn, "x-amz-target") == ["AmazonSQS.DeleteMessageBatch"] - Plug.Conn.resp(conn, 200, Jason.encode!(%{"Successful" => [%{"Id" => "1"}]})) + Req.Test.json(conn, %{"Successful" => [%{"Id" => "1"}]}) end) assert {:ok, %{"Successful" => [%{"Id" => "1"}]}} = SQS.delete_message_batch( queue_url, entries, - Keyword.put(@request_options, :endpoint, queue_url) + Keyword.merge(@request_options, endpoint: queue_url, plug: {Req.Test, __MODULE__}) ) end test "change_message_visibility_batch builds the visibility payload" do - bypass = Bypass.open() - queue_url = "http://localhost:#{bypass.port}/queue" + queue_url = queue_url() entries = [ %{ @@ -97,8 +98,8 @@ defmodule BroadwaySQS.ReqClient.SQSTest do } ] - Bypass.expect_once(bypass, "POST", "/queue", fn conn -> - {:ok, body, conn} = Plug.Conn.read_body(conn) + Req.Test.expect(__MODULE__, fn conn -> + body = Req.Test.raw_body(conn) assert Jason.decode!(body) == %{"QueueUrl" => queue_url, "Entries" => entries} @@ -106,17 +107,19 @@ defmodule BroadwaySQS.ReqClient.SQSTest do "AmazonSQS.ChangeMessageVisibilityBatch" ] - Plug.Conn.resp(conn, 200, Jason.encode!(%{"Successful" => [%{"Id" => "1"}]})) + Req.Test.json(conn, %{"Successful" => [%{"Id" => "1"}]}) end) assert {:ok, %{"Successful" => [%{"Id" => "1"}]}} = SQS.change_message_visibility_batch( queue_url, entries, - Keyword.put(@request_options, :endpoint, queue_url) + Keyword.merge(@request_options, endpoint: queue_url, plug: {Req.Test, __MODULE__}) ) end + defp queue_url, do: "https://sqs.eu-west-1.amazonaws.com/123456789012/test-queue" + test "normalize_message converts SQS messages to the client format" do message = %{ "MessageId" => "message-id", diff --git a/test/broadway_sqs/req_client_test.exs b/test/broadway_sqs/req_client_test.exs index 8d74749..53ad491 100644 --- a/test/broadway_sqs/req_client_test.exs +++ b/test/broadway_sqs/req_client_test.exs @@ -12,8 +12,8 @@ defmodule BroadwaySQS.ReqClientTest do region: "eu-west-1" ] - setup do - Req.Test.set_req_test_from_context(__MODULE__) + setup context do + Req.Test.set_req_test_from_context(context) Req.Test.stub(__MODULE__, fn conn -> Req.Test.json(conn, %{}) end) :ok end @@ -24,6 +24,11 @@ defmodule BroadwaySQS.ReqClientTest do Req.Test.expect(__MODULE__, fn conn -> payload = Jason.decode!(Req.Test.raw_body(conn)) + assert Plug.Conn.get_req_header(conn, "content-type") == ["application/x-amz-json-1.0"] + assert Plug.Conn.get_req_header(conn, "x-amz-target") == ["AmazonSQS.ReceiveMessage"] + assert [authorization] = Plug.Conn.get_req_header(conn, "authorization") + assert String.starts_with?(authorization, "AWS4-HMAC-SHA256 ") + assert payload == %{ "QueueUrl" => queue_url, "MaxNumberOfMessages" => 2, @@ -89,6 +94,57 @@ defmodule BroadwaySQS.ReqClientTest do assert log =~ "Unable to fetch events from AWS queue #{queue_url}" end + test "sends a session token" do + queue_url = queue_url() + + Req.Test.expect(__MODULE__, fn conn -> + assert Plug.Conn.get_req_header(conn, "x-amz-security-token") == ["session-token"] + Req.Test.json(conn, %{"Messages" => []}) + end) + + {:ok, opts} = + ReqClient.init( + opts(queue_url, + config: Keyword.put(@config, :token, "session-token") + ) + ) + + assert ReqClient.receive_messages(1, opts) == [] + end + + test "logs when credentials are unavailable" do + queue_url = queue_url() + + log = + capture_log(fn -> + {:ok, opts} = ReqClient.init(opts(queue_url, config: [region: "eu-west-1"])) + assert ReqClient.receive_messages(1, opts) == [] + end) + + assert log =~ "aws_credentials_not_found" + end + + test "logs when the region is unavailable" do + queue_url = queue_url() + + log = + capture_log(fn -> + {:ok, opts} = + ReqClient.init( + opts(queue_url, + config: [ + access_key_id: "access-key", + secret_access_key: "secret-key" + ] + ) + ) + + assert ReqClient.receive_messages(1, opts) == [] + end) + + assert log =~ "aws_region_not_found" + end + test "acknowledges messages in delete batches" do queue_url = queue_url() test_pid = self() From 7d22a35b7a347b921238ef2de984493b04c80256 Mon Sep 17 00:00:00 2001 From: Milad Rastian Date: Wed, 12 Aug 2026 08:12:00 +0200 Subject: [PATCH 4/7] remove ExAwsClient Use ReqClient by default. Remove the unused dependencies and update the docs --- README.md | 6 +- examples/sqs_example/config/config.exs | 6 +- examples/sqs_example/lib/helpers.ex | 26 +- examples/sqs_example/mix.exs | 4 +- examples/sqs_example/mix.lock | 5 - lib/broadway_sqs/ex_aws_client.ex | 132 --------- lib/broadway_sqs/options.ex | 10 +- lib/broadway_sqs/producer.ex | 13 +- mix.exs | 4 - mix.lock | 36 +-- test/broadway_sqs/ex_aws_client_test.exs | 336 ----------------------- 11 files changed, 44 insertions(+), 534 deletions(-) delete mode 100644 lib/broadway_sqs/ex_aws_client.ex delete mode 100644 test/broadway_sqs/ex_aws_client_test.exs diff --git a/README.md b/README.md index 76642d2..7bb5ae0 100644 --- a/README.md +++ b/README.md @@ -8,14 +8,12 @@ For more details on using Broadway with Amazon SQS, please see the ## Installation -Add `:broadway_sqs` to the list of dependencies in `mix.exs` along with the HTTP -client of your choice (defaults to `:hackney`): +Add `:broadway_sqs` to the list of dependencies in `mix.exs`: ```elixir def deps do [ - {:broadway_sqs, "~> 0.7.1"}, - {:hackney, "~> 1.9"} + {:broadway_sqs, "~> 0.7.1"} ] end ``` diff --git a/examples/sqs_example/config/config.exs b/examples/sqs_example/config/config.exs index 8ff4fc8..d8231dd 100644 --- a/examples/sqs_example/config/config.exs +++ b/examples/sqs_example/config/config.exs @@ -3,13 +3,15 @@ use Mix.Config config :broadway_sqs_example, producer_module: {BroadwaySQS.Producer, - sqs_client: BroadwaySQS.ExAwsClient, + sqs_client: BroadwaySQS.ReqClient, config: [ # access_key_id: "YOUR_AWS_ACCESS_KEY_ID", # secret_access_key: "YOUR_AWS_SECRET_ACCESS_KEY" region: "us-east-2" ]}, int_queue: "TEST-int-queue", - string_queue: "TEST-string-queue" + string_queue: "TEST-string-queue", + region: "us-east-2", + sqs_endpoint: "https://sqs.us-east-2.amazonaws.com" import_config "#{Mix.env()}.exs" diff --git a/examples/sqs_example/lib/helpers.ex b/examples/sqs_example/lib/helpers.ex index d4b1489..fb8819a 100644 --- a/examples/sqs_example/lib/helpers.ex +++ b/examples/sqs_example/lib/helpers.ex @@ -1,27 +1,25 @@ defmodule BroadwaySQSExample.Helpers do def send_strings_sqs(queue, msg, amount) do - sqs_req = ExAws.SQS.send_message(queue, msg) - Enum.each(1..amount, fn _x -> Task.async(fn -> - ExAws.request(sqs_req, region: "us-east-2") + request("AmazonSQS.SendMessage", %{"QueueUrl" => queue, "MessageBody" => msg}) end) end) end def send_ints_sqs(queue, amount) do Enum.each(1..amount, fn x -> - sqs_req = ExAws.SQS.send_message(queue, x) - Task.async(fn -> - ExAws.request(sqs_req, region: "us-east-2") + request("AmazonSQS.SendMessage", %{ + "QueueUrl" => queue, + "MessageBody" => to_string(x) + }) end) end) end def create_sqs_queue(queue) do - sqs_req = ExAws.SQS.create_queue(queue) - ExAws.request(sqs_req, region: "us-east-2") + request("AmazonSQS.CreateQueue", %{"QueueName" => queue}) end def create_default_queues() do @@ -43,4 +41,16 @@ defmodule BroadwaySQSExample.Helpers do string_queue = Application.get_env(:broadway_sqs_example, :string_queue) send_strings_sqs(string_queue, "testing", 100) end + + defp request(action, payload) do + credentials = :aws_credentials.get_credentials() + region = Application.get_env(:broadway_sqs_example, :region, "us-east-2") + + BroadwaySQS.ReqClient.Request.call(action, payload, + credentials: credentials, + region: region, + endpoint: Application.get_env(:broadway_sqs_example, :sqs_endpoint), + queue_url: Application.get_env(:broadway_sqs_example, :sqs_endpoint) + ) + end end diff --git a/examples/sqs_example/mix.exs b/examples/sqs_example/mix.exs index c588e31..d952cf5 100644 --- a/examples/sqs_example/mix.exs +++ b/examples/sqs_example/mix.exs @@ -20,9 +20,7 @@ defmodule BroadwaySQSExample.MixProject do defp deps do [ - {:broadway_sqs, path: "../.."}, - {:hackney, "~> 1.9"}, - {:httpoison, "~> 0.13.0"} + {:broadway_sqs, path: "../.."} ] end end diff --git a/examples/sqs_example/mix.lock b/examples/sqs_example/mix.lock index 986b189..3382a40 100644 --- a/examples/sqs_example/mix.lock +++ b/examples/sqs_example/mix.lock @@ -2,17 +2,12 @@ "broadway": {:hex, :broadway, "0.6.2", "ef8e0d257420c72f0e600958cf95556835d9921ad14be333493083226458791a", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "f4f93704304a736c984cd6ed884f697415f68eb50906f4dc5d641926366ad8fa"}, "broadway_sqs": {:hex, :broadway_sqs, "0.1.0", "dd9d2d404ccbca9252fbbad54551bb73d685c1c37103e37adfa21267da888fe5", [:mix], [{:broadway, "~> 0.1", [hex: :broadway, repo: "hexpm", optional: false]}, {:ex_aws_sqs, "~> 2.0", [hex: :ex_aws_sqs, repo: "hexpm", optional: false]}, {:sweet_xml, "~> 0.6", [hex: :sweet_xml, repo: "hexpm", optional: false]}], "hexpm"}, "certifi": {:hex, :certifi, "2.5.3", "70bdd7e7188c804f3a30ee0e7c99655bc35d8ac41c23e12325f36ab449b70651", [:rebar3], [{:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm", "ed516acb3929b101208a9d700062d520f3953da3b6b918d866106ffa980e1c10"}, - "ex_aws": {:hex, :ex_aws, "2.1.6", "41ab8b4caa48035c96d07faa035d2d9de6df480e7e084c054e662ac888dcd4d4", [:mix], [{:configparser_ex, "~> 4.0", [hex: :configparser_ex, repo: "hexpm", optional: true]}, {:hackney, "~> 1.9", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:jsx, "~> 2.8", [hex: :jsx, repo: "hexpm", optional: true]}, {:sweet_xml, "~> 0.6", [hex: :sweet_xml, repo: "hexpm", optional: true]}], "hexpm", "a541bd042c1ee26412bb1e749ddf2a1c327e4fb7e382b1cd227e1b00eed3d469"}, - "ex_aws_sqs": {:hex, :ex_aws_sqs, "3.2.1", "fc6772b1cd894a73494498f73820f4171e88f48dadcb64c632d1413fb4592cdb", [:mix], [{:ex_aws, "~> 2.0", [hex: :ex_aws, repo: "hexpm", optional: false]}, {:saxy, "~> 1.1", [hex: :saxy, repo: "hexpm", optional: true]}, {:sweet_xml, ">= 0.0.0", [hex: :sweet_xml, repo: "hexpm", optional: true]}], "hexpm", "ae77e296dffc0608221f14287cea5621b4419b94794804fd20bbf6cf8c71561e"}, "gen_stage": {:hex, :gen_stage, "1.0.0", "51c8ae56ff54f9a2a604ca583798c210ad245f415115453b773b621c49776df5", [:mix], [], "hexpm", "1d9fc978db5305ac54e6f5fec7adf80cd893b1000cf78271564c516aa2af7706"}, - "hackney": {:hex, :hackney, "1.17.0", "717ea195fd2f898d9fe9f1ce0afcc2621a41ecfe137fae57e7fe6e9484b9aa99", [:rebar3], [{:certifi, "~>2.5", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~>6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~>1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~>1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~>1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~>0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "64c22225f1ea8855f584720c0e5b3cd14095703af1c9fbc845ba042811dc671c"}, - "httpoison": {:hex, :httpoison, "0.13.0", "bfaf44d9f133a6599886720f3937a7699466d23bb0cd7a88b6ba011f53c6f562", [:mix], [{:hackney, "~> 1.8", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm", "4846958172d6401c4f34ecc5c2c4607b5b0d90b8eec8f6df137ca4907942ed0f"}, "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~>0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"}, "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"}, "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm", "f278585650aa581986264638ebf698f8bb19df297f66ad91b18910dfc6e19323"}, "nimble_options": {:hex, :nimble_options, "0.3.5", "a4f6820cdcb4ee444afd78635f323e58e8a5ddf2fbbe9b9d283a99f972034bae", [:mix], [], "hexpm", "f5507cc90033a8d12769522009c80aa9164af6bab245dbd4ad421d008455f1e1"}, "parse_trans": {:hex, :parse_trans, "3.3.1", "16328ab840cc09919bd10dab29e431da3af9e9e7e7e6f0089dd5a2d2820011d8", [:rebar3], [], "hexpm", "07cd9577885f56362d414e8c4c4e6bdf10d43a8767abb92d24cbe8b24c54888b"}, - "saxy": {:hex, :saxy, "1.3.0", "61b52697a3235be68ce6f8ecc2c7032f3c01184b14d142a7d09270019e32fbf9", [:mix], [], "hexpm", "c9770a08c168be95c8d8249a9051dd5522641941ec1d9cb843e7d12dc101b6d2"}, "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.6", "cf344f5692c82d2cd7554f5ec8fd961548d4fd09e7d22f5b62482e5aeaebd4b0", [:make, :mix, :rebar3], [], "hexpm", "bdb0d2471f453c88ff3908e7686f86f9be327d065cc1ec16fa4540197ea04680"}, "telemetry": {:hex, :telemetry, "0.4.2", "2808c992455e08d6177322f14d3bdb6b625fbcfd233a73505870d8738a2f4599", [:rebar3], [], "hexpm", "2d1419bd9dda6a206d7b5852179511722e2b18812310d304620c7bd92a13fcef"}, "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.0", "bc84380c9ab48177092f43ac89e4dfa2c6d62b40b8bd132b1059ecc7232f9a78", [:rebar3], [], "hexpm", "25eee6d67df61960cf6a794239566599b09e17e668d3700247bc498638152521"}, diff --git a/lib/broadway_sqs/ex_aws_client.ex b/lib/broadway_sqs/ex_aws_client.ex deleted file mode 100644 index ea83a3b..0000000 --- a/lib/broadway_sqs/ex_aws_client.ex +++ /dev/null @@ -1,132 +0,0 @@ -defmodule BroadwaySQS.ExAwsClient do - @moduledoc """ - Default SQS client used by `BroadwaySQS.Producer` to communicate with AWS - SQS service. - - This client uses the `ExAws.SQS` library and implements the - `BroadwaySQS.SQSClient` and `Broadway.Acknowledger` behaviours which define - callbacks for receiving and acknowledging messages. - """ - - alias Broadway.{Message, Acknowledger} - require Logger - - @behaviour BroadwaySQS.SQSClient - @behaviour Acknowledger - - @max_num_messages_allowed_by_aws 10 - - @impl true - def init(opts) do - opts_map = opts |> Enum.into(%{ack_ref: opts[:broadway][:name]}) - - {:ok, opts_map} - end - - @impl true - def receive_messages(demand, opts) do - receive_messages_opts = build_receive_messages_opts(opts, demand) - - opts.queue_url - |> ExAws.SQS.receive_message(receive_messages_opts) - |> ExAws.request(opts.config) - |> wrap_received_messages(opts) - end - - @impl Acknowledger - def ack(ack_ref, successful, failed) do - ack_options = :persistent_term.get(ack_ref) - - messages_to_delete = - Enum.filter(successful, &ack?(&1, ack_options, :on_success)) ++ - Enum.filter(failed, &ack?(&1, ack_options, :on_failure)) - - messages_to_nack_with_timeout = - Enum.flat_map(successful, &nack(&1, ack_options, :on_success)) ++ - Enum.flat_map(failed, &nack(&1, ack_options, :on_failure)) - - messages_to_delete - |> Enum.chunk_every(@max_num_messages_allowed_by_aws) - |> Enum.each(&delete_messages(&1, ack_options)) - - messages_to_nack_with_timeout - |> Enum.chunk_every(@max_num_messages_allowed_by_aws) - |> Enum.each(&change_message_visibilities(&1, ack_options)) - end - - defp ack?(message, ack_options, option) do - {_, _, message_ack_options} = message.acknowledger - (message_ack_options[option] || Map.fetch!(ack_options, option)) == :ack - end - - defp nack(message, ack_options, option) do - {_, _, message_ack_options} = message.acknowledger - - case message_ack_options[option] || Map.fetch!(ack_options, option) do - {:nack, timeout} -> [{message, timeout}] - _ -> [] - end - end - - @impl Acknowledger - def configure(_ack_ref, ack_data, options) do - {:ok, Map.merge(ack_data, Map.new(options))} - end - - defp delete_messages(messages, ack_options) do - receipts = Enum.map(messages, &extract_message_receipt/1) - - ack_options.queue_url - |> ExAws.SQS.delete_message_batch(receipts) - |> ExAws.request!(ack_options.config) - end - - defp change_message_visibilities(messages_with_timeouts, ack_options) do - entries = - Enum.map(messages_with_timeouts, fn {message, timeout} -> - message - |> extract_message_receipt() - |> Map.put(:visibility_timeout, timeout) - end) - - ack_options.queue_url - |> ExAws.SQS.change_message_visibility_batch(entries) - |> ExAws.request!(ack_options.config) - end - - defp wrap_received_messages({:ok, %{body: body}}, %{ack_ref: ack_ref}) do - Enum.map(body.messages, fn message -> - metadata = Map.delete(message, :body) - acknowledger = build_acknowledger(message, ack_ref) - %Message{data: message.body, metadata: metadata, acknowledger: acknowledger} - end) - end - - defp wrap_received_messages({:error, reason}, %{queue_url: queue_url}) do - Logger.error("Unable to fetch events from AWS queue #{queue_url}. Reason: #{inspect(reason)}") - [] - end - - defp build_acknowledger(message, ack_ref) do - receipt = %{id: message.message_id, receipt_handle: message.receipt_handle} - {__MODULE__, ack_ref, %{receipt: receipt}} - end - - defp build_receive_messages_opts(opts, demand) do - max_number_of_messages = min(demand, opts[:max_number_of_messages]) - - [ - max_number_of_messages: max_number_of_messages, - wait_time_seconds: opts[:wait_time_seconds], - visibility_timeout: opts[:visibility_timeout], - attribute_names: opts[:attribute_names], - message_attribute_names: opts[:message_attribute_names] - ] - |> Enum.filter(fn {_, value} -> value end) - end - - defp extract_message_receipt(message) do - {_, _, %{receipt: receipt}} = message.acknowledger - receipt - end -end diff --git a/lib/broadway_sqs/options.ex b/lib/broadway_sqs/options.ex index 9092562..0c030a1 100644 --- a/lib/broadway_sqs/options.ex +++ b/lib/broadway_sqs/options.ex @@ -23,7 +23,7 @@ defmodule BroadwaySQS.Options do messages. Pay attention that all options passed to the producer will be forwarded to the client. """, - default: BroadwaySQS.ExAwsClient + default: BroadwaySQS.ReqClient ], receive_interval: [ type: :non_neg_integer, @@ -53,10 +53,10 @@ defmodule BroadwaySQS.Options do type: :keyword_list, default: [], doc: """ - A set of options that overrides the default ExAws configuration - options. The most commonly used options are: `:access_key_id`, `:secret_access_key`, - `:scheme`, `:region` and `:port`. For a complete list of configuration options and - their default values, please see the `ExAws` documentation. + A set of options for the SQS client. The `:region` option is used for + AWS Signature Version 4. Credentials are normally discovered through + `aws_credentials`; `:access_key_id`, `:secret_access_key`, and `:token` + can be provided for explicit credentials. """ ], max_number_of_messages: [ diff --git a/lib/broadway_sqs/producer.ex b/lib/broadway_sqs/producer.ex index 8cb6f32..f680e45 100644 --- a/lib/broadway_sqs/producer.ex +++ b/lib/broadway_sqs/producer.ex @@ -3,7 +3,7 @@ defmodule BroadwaySQS.Producer do A GenStage producer that continuously polls messages from a SQS queue and acknowledge them after being successfully processed. - By default this producer uses `BroadwaySQS.ExAwsClient` to talk to SQS but + By default this producer uses `BroadwaySQS.ReqClient` to talk to SQS but you can provide your client by implementing the `BroadwaySQS.SQSClient` behaviour. @@ -14,7 +14,7 @@ defmodule BroadwaySQS.Producer do Aside from `:receive_interval` and `:sqs_client` which are generic and apply to all producers (regardless of the client implementation), all other options are specific to - the `BroadwaySQS.ExAwsClient`, which is the default client. + the `BroadwaySQS.ReqClient`, which is the default client. #{NimbleOptions.docs(BroadwaySQS.Options.definition())} @@ -194,15 +194,6 @@ defmodule BroadwaySQS.Producer do def prepare_for_start(_module, broadway_opts) do {producer_module, client_opts} = broadway_opts[:producer][:module] - if Keyword.has_key?(client_opts, :queue_name) do - Logger.error( - "The option :queue_name has been removed in order to keep compatibility with " <> - "ex_aws_sqs >= v3.0.0. Please set the queue URL using the new :queue_url option." - ) - - exit(:invalid_config) - end - case NimbleOptions.validate(client_opts, BroadwaySQS.Options.definition()) do {:error, error} -> raise ArgumentError, format_error(error) diff --git a/mix.exs b/mix.exs index ec2ee18..a9dacf9 100644 --- a/mix.exs +++ b/mix.exs @@ -29,12 +29,8 @@ defmodule BroadwaySqs.MixProject do {:broadway, "~> 1.0"}, {:req, "~> 0.7.2"}, {:aws_credentials, "~> 1.0"}, - {:ex_aws_sqs, "~> 3.2.1 or ~> 3.3"}, {:nimble_options, "~> 0.3.7 or ~> 0.4 or ~> 1.0"}, {:telemetry, "~> 0.4.3 or ~> 1.0"}, - {:saxy, "~> 1.1"}, - {:hackney, "~> 1.9", only: [:dev, :test]}, - {:bypass, "~> 2.1.0", only: :test}, {:ex_doc, ">= 0.19.0", only: :docs} ] end diff --git a/mix.lock b/mix.lock index c113d20..dbf198b 100644 --- a/mix.lock +++ b/mix.lock @@ -1,42 +1,30 @@ %{ "aws_credentials": {:hex, :aws_credentials, "1.1.1", "4a28d7d2c01956dd9a2cc52a7edd5cb1e58343766715753d4b8d0d043771b95c", [:rebar3], [{:eini, "~> 2.2.5", [hex: :eini_beam, repo: "hexpm", optional: false]}, {:iso8601, "~> 1.3.4", [hex: :iso8601, repo: "hexpm", optional: false]}, {:jsx, "~> 3.1.0", [hex: :jsx, repo: "hexpm", optional: false]}], "hexpm", "8655e0e3c82c5ad659729ea8717ae1f0a087ea268e15aefa674fc3d781c1e9e2"}, - "broadway": {:hex, :broadway, "1.0.7", "7808f9e3eb6f53ca6d060f0f9d61012dd8feb0d7a82e62d087dd517b9b66fa53", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.3.7 or ~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "e76cfb0a7d64176c387b8b1ddbfb023e2ee8a63e92f43664d78e6d5d0b1177c6"}, - "bypass": {:hex, :bypass, "2.1.0", "909782781bf8e20ee86a9cabde36b259d44af8b9f38756173e8f5e2e1fabb9b1", [:mix], [{:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.0", [hex: :plug_cowboy, repo: "hexpm", optional: false]}, {:ranch, "~> 1.3", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "d9b5df8fa5b7a6efa08384e9bbecfe4ce61c77d28a4282f79e02f1ef78d96b80"}, - "certifi": {:hex, :certifi, "2.9.0", "6f2a475689dd47f19fb74334859d460a2dc4e3252a3324bd2111b8f0429e7e21", [:rebar3], [], "hexpm", "266da46bdb06d6c6d35fde799bcb28d36d985d424ad7c08b5bb48f5b5cdd4641"}, - "cowboy": {:hex, :cowboy, "2.10.0", "ff9ffeff91dae4ae270dd975642997afe2a1179d94b1887863e43f681a203e26", [:make, :rebar3], [{:cowlib, "2.12.1", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "3afdccb7183cc6f143cb14d3cf51fa00e53db9ec80cdcd525482f5e99bc41d6b"}, + "broadway": {:hex, :broadway, "1.3.0", "f75f6376159b74f55c5ba2629dac613e4fd79d9e71148ab5fbac8fdd7c999d2a", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.3.7 or ~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "bef3b4c5512d0072917b70239cbecf8f76a2587465a5b7c3e2b9ae18b4bc405b"}, + "cowboy": {:hex, :cowboy, "2.18.0", "bff388eb4d6356cb3f88c26e65b515976bf04b401b805d31550c3e60eba8fe18", [:make, :rebar3], [{:cowlib, ">= 2.19.0 and < 3.0.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, ">= 1.8.0 and < 3.0.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "62d0b26abcf455054972b0da242389c69d5982ce5914afb8c344517f667b9600"}, "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.4.0", "f239f68b588efa7707abce16a84d0d2acf3a0f50571f8bb7f56a15865aae820c", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7d98bac1ee4565d31b62d59f8823dfd8356a169e7fcbb83831b8a5397404c9de"}, - "cowlib": {:hex, :cowlib, "2.12.1", "a9fa9a625f1d2025fe6b462cb865881329b5caff8f1854d1cbc9f9533f00e1e1", [:make, :rebar3], [], "hexpm", "163b73f6367a7341b33c794c4e88e7dbfe6498ac42dcd69ef44c5bc5507c8db0"}, - "earmark_parser": {:hex, :earmark_parser, "1.4.39", "424642f8335b05bb9eb611aa1564c148a8ee35c9c8a8bba6e129d51a3e3c6769", [:mix], [], "hexpm", "06553a88d1f1846da9ef066b87b57c6f605552cfbe40d20bd8d59cc6bde41944"}, + "cowlib": {:hex, :cowlib, "2.19.0", "c9d11c9d035472e27a740c9f327786c61ed209269b4be0260d59d3ec07b8949f", [:make, :rebar3], [], "hexpm", "6dc66e3135b229193ea4dcb14294e79520c923d391315c9c962ef0b4bea72356"}, + "earmark_parser": {:hex, :earmark_parser, "1.4.46", "67607a0532e810c6f630a515c548d0b24949643f168cc556303bee4cf96105c7", [:mix], [], "hexpm", "9c44636e8a1c68c62f526b2dcd85d941dbbcee7ab82cf64ba06ce28bef8e89f5"}, "eini": {:hex, :eini_beam, "2.2.5", "28fa0a4eb7ff885cc388877b73fbd56fcca3b2eae4d81e47c47f317b900da1da", [:rebar3], [], "hexpm", "511e9207649f3becb5d945f1813615987899cf78fa130f92be07193a6a74ecb8"}, - "ex_aws": {:hex, :ex_aws, "2.4.3", "6c6d88ba7b9c07e3b0f4b70406d5fccb9f5358f5ef18138f7bd396f7863e8255", [:mix], [{:configparser_ex, "~> 4.0", [hex: :configparser_ex, repo: "hexpm", optional: true]}, {:hackney, "~> 1.16", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:jsx, "~> 2.8 or ~> 3.0", [hex: :jsx, repo: "hexpm", optional: true]}, {:mime, "~> 1.2 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:sweet_xml, "~> 0.7", [hex: :sweet_xml, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "67f61f8b6aec740150d483a21f551fabce26a481d9917305ed2bb47717007519"}, - "ex_aws_sqs": {:hex, :ex_aws_sqs, "3.4.0", "f7c4d0177c1c954776363d3dc05e5dfd37ddf0e2c65ec3f047e5c9c7dd1b71ac", [:mix], [{:ex_aws, "~> 2.1", [hex: :ex_aws, repo: "hexpm", optional: false]}, {:hackney, "~> 1.9", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:saxy, "~> 1.1", [hex: :saxy, repo: "hexpm", optional: true]}, {:sweet_xml, ">= 0.0.0", [hex: :sweet_xml, repo: "hexpm", optional: true]}], "hexpm", "b504482206ccaf767b714888e9d41a1cfcdcb241577985517114191c812f155a"}, - "ex_doc": {:hex, :ex_doc, "0.34.1", "9751a0419bc15bc7580c73fde506b17b07f6402a1e5243be9e0f05a68c723368", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "d441f1a86a235f59088978eff870de2e815e290e44a8bd976fe5d64470a4c9d2"}, + "ex_doc": {:hex, :ex_doc, "0.40.3", "4a972ffe64bc07dc605af487e98fc19b72a4185f55ca031b94c0552d6071c1d9", [:mix], [{:earmark_parser, "~> 1.4.44", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "2756e357742fecd9749b489b85d67c9ce99c465f2e75728d9e6dc8d704b973de"}, "finch": {:hex, :finch, "0.23.0", "e3f9287ac25a8832f848b144c2b57346aac65b205e2e0629a52adfe6507fd837", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.8", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "80e58d3f936f57e3fdf404f83a3642897ae6d9fb642934e46da4d8fe761b99d5"}, - "gen_stage": {:hex, :gen_stage, "1.2.1", "19d8b5e9a5996d813b8245338a28246307fd8b9c99d1237de199d21efc4c76a1", [:mix], [], "hexpm", "83e8be657fa05b992ffa6ac1e3af6d57aa50aace8f691fcf696ff02f8335b001"}, - "hackney": {:hex, :hackney, "1.18.1", "f48bf88f521f2a229fc7bae88cf4f85adc9cd9bcf23b5dc8eb6a1788c662c4f6", [:rebar3], [{:certifi, "~> 2.9.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~> 6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~> 1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.3.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "a4ecdaff44297e9b5894ae499e9a070ea1888c84afdd1fd9b7b2bc384950128e"}, + "gen_stage": {:hex, :gen_stage, "1.3.2", "7c77e5d1e97de2c6c2f78f306f463bca64bf2f4c3cdd606affc0100b89743b7b", [:mix], [], "hexpm", "0ffae547fa777b3ed889a6b9e1e64566217413d018cabd825f786e843ffe63e7"}, "hpax": {:hex, :hpax, "1.0.4", "777de5d433b0fbdc7c418159c8055910faa8047ffdb3d6b31098d2a46cd7685c", [:mix], [], "hexpm", "afc7cb142ebcc2d01ce7816190b98ce5dd49e799111b24249f3443d730f377ca"}, - "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"}, "iso8601": {:hex, :iso8601, "1.3.4", "7b1f095f86f6cf65e1e5a77872e8e8bf69bd58d4c3a415b3f77d9cc9423ecbb9", [:rebar3], [], "hexpm", "a334469c07f1c219326bc891a95f5eec8eb12dd8071a3fff56a7843cb20fae34"}, "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"}, "jsx": {:hex, :jsx, "3.1.0", "d12516baa0bb23a59bb35dccaf02a1bd08243fcbb9efe24f2d9d056ccff71268", [:rebar3], [], "hexpm", "0c5cc8fdc11b53cc25cf65ac6705ad39e54ecc56d1c22e4adb8f5a53fb9427f3"}, - "makeup": {:hex, :makeup, "1.1.2", "9ba8837913bdf757787e71c1581c21f9d2455f4dd04cfca785c70bbfff1a76a3", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cce1566b81fbcbd21eca8ffe808f33b221f9eee2cbc7a1706fc3da9ff18e6cac"}, - "makeup_elixir": {:hex, :makeup_elixir, "0.16.2", "627e84b8e8bf22e60a2579dad15067c755531fea049ae26ef1020cad58fe9578", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "41193978704763f6bbe6cc2758b84909e62984c7752b3784bd3c218bb341706b"}, - "makeup_erlang": {:hex, :makeup_erlang, "1.0.0", "6f0eff9c9c489f26b69b61440bf1b238d95badae49adac77973cbacae87e3c2e", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "ea7a9307de9d1548d2a72d299058d1fd2339e3d398560a0e46c27dab4891e4d2"}, - "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"}, + "makeup": {:hex, :makeup, "1.2.2", "882d46dc0905e9ff7abf2aab61a7e6b3dcc555533977d8a23b06019e6c89ac94", [:mix], [{:nimble_parsec, "~> 1.4", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "9a1a24e5b343b8ae16abea0822c10a6f75da27af7fa802ada5251f7579bfccfa"}, + "makeup_elixir": {:hex, :makeup_elixir, "1.0.1", "e928a4f984e795e41e3abd27bfc09f51db16ab8ba1aebdba2b3a575437efafc2", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "7284900d412a3e5cfd97fdaed4f5ed389b8f2b4cb49efc0eb3bd10e2febf9507"}, + "makeup_erlang": {:hex, :makeup_erlang, "1.1.0", "835f7e60792e08824cda445639555d7bf1bbbddb1b60b306e33cb6f6db24dc74", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "1cd6780fb1dd1a03979abaed0fe82712b0625118fd5257d3ebbf73f960c73c3c"}, "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"}, - "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm", "f278585650aa581986264638ebf698f8bb19df297f66ad91b18910dfc6e19323"}, "mint": {:hex, :mint, "1.9.3", "3337184d69179695c7a9f1714d92c11e629d36c8c037a21cf490131d3d150554", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "5f7c9342480c069dbbc4eeac3490303c9e01870ff01a7f1d29b6107054fc1e74"}, "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"}, - "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"}, + "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"}, "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, - "parse_trans": {:hex, :parse_trans, "3.3.1", "16328ab840cc09919bd10dab29e431da3af9e9e7e7e6f0089dd5a2d2820011d8", [:rebar3], [], "hexpm", "07cd9577885f56362d414e8c4c4e6bdf10d43a8767abb92d24cbe8b24c54888b"}, "plug": {:hex, :plug, "1.20.3", "56c480c633ec2ce10140e236e15233bf576e1d323887d7c96711bd02ab5160db", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "be266aee1b8536ef6409d58cf39a3121319f0ec47cfa1b24024485aa0e76ad76"}, - "plug_cowboy": {:hex, :plug_cowboy, "2.6.1", "9a3bbfceeb65eff5f39dab529e5cd79137ac36e913c02067dba3963a26efe9b2", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "de36e1a21f451a18b790f37765db198075c25875c64834bcc82d90b309eb6613"}, + "plug_cowboy": {:hex, :plug_cowboy, "2.9.0", "87e21e0d9054ced99c36d128f49e3ea2cd8b745fffb97de50bff99706087af4f", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.18", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "2002bafba4f3a45b55a58e68d70211b153a7ed18d37edb1ceb6e96e7a92c422e"}, "plug_crypto": {:hex, :plug_crypto, "2.2.0", "144014737daaf485407f5ed77daeaad74d651b216a28c87543f8cc7043f8efc8", [:mix], [], "hexpm", "83a95744ab1c75876542b6fab135fcc176280e0f301a111c1f757fddcec95d2c"}, - "ranch": {:hex, :ranch, "1.8.0", "8c7a100a139fd57f17327b6413e4167ac559fbc04ca7448e9be9057311597a1d", [:make, :rebar3], [], "hexpm", "49fbcfd3682fab1f5d109351b61257676da1a2fdbe295904176d5e521a2ddfe5"}, + "ranch": {:hex, :ranch, "1.8.1", "208169e65292ac5d333d6cdbad49388c1ae198136e4697ae2f474697140f201c", [:make, :rebar3], [], "hexpm", "aed58910f4e21deea992a67bf51632b6d60114895eb03bb392bb733064594dd0"}, "req": {:hex, :req, "0.7.2", "364eae2e5f5c984f2dac6d71c07f8c8c89ce0bc49c4d746dacb7a306823020de", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:finch, "~> 0.21", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "c9cdfa276b05d8db2a27fda5d233e6858b764d47189d76cbb186e130a871ae0b"}, - "saxy": {:hex, :saxy, "1.5.0", "0141127f2d042856f135fb2d94e0beecda7a2306f47546dbc6411fc5b07e28bf", [:mix], [], "hexpm", "ea7bb6328fbd1f2aceffa3ec6090bfb18c85aadf0f8e5030905e84235861cf89"}, - "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"}, "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, - "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.0", "bc84380c9ab48177092f43ac89e4dfa2c6d62b40b8bd132b1059ecc7232f9a78", [:rebar3], [], "hexpm", "25eee6d67df61960cf6a794239566599b09e17e668d3700247bc498638152521"}, } diff --git a/test/broadway_sqs/ex_aws_client_test.exs b/test/broadway_sqs/ex_aws_client_test.exs deleted file mode 100644 index 1154baa..0000000 --- a/test/broadway_sqs/ex_aws_client_test.exs +++ /dev/null @@ -1,336 +0,0 @@ -defmodule BroadwaySQS.ExAwsClientTest do - use ExUnit.Case - - alias BroadwaySQS.ExAwsClient - alias Broadway.Message - import ExUnit.CaptureLog - - defmodule FakeHttpClient do - @behaviour ExAws.Request.HttpClient - - def request(:post, url, "Action=ReceiveMessage" <> _ = body, _, _) do - send(self(), {:http_request_called, %{url: url, body: body}}) - - response_body = """ - - - - Id_1 - ReceiptHandle_1 - fake_md5 - Message 1 - - SenderId - 13 - - - ApproximateReceiveCount - 5 - - - TestStringAttribute - - Test - String - - - - - Id_2 - ReceiptHandle_2 - Message 2 - - - - """ - - {:ok, %{status_code: 200, body: response_body}} - end - - def request(:post, url, "Action=DeleteMessageBatch" <> _ = body, _, _) do - send(self(), {:http_request_called, %{url: url, body: body}}) - - {:ok, %{status_code: 200, body: ""}} - end - - def request(:post, url, "Action=ChangeMessageVisibilityBatch" <> _ = body, _, _) do - send(self(), {:http_request_called, %{url: url, body: body}}) - - {:ok, %{status_code: 200, body: ""}} - end - end - - defmodule FakeHttpClientWithError do - @behaviour ExAws.Request.HttpClient - - def request(:post, _url, "Action=ReceiveMessage" <> _, _, _) do - {:error, %{reason: "Fake error"}} - end - end - - describe "receive_messages/2" do - setup do - %{ - opts: [ - # will be injected by broadway at runtime - broadway: [name: :Broadway3], - queue_url: "my_queue", - config: [ - http_client: FakeHttpClient, - access_key_id: "FAKE_ID", - secret_access_key: "FAKE_KEY", - retries: [max_attempts: 0] - ] - ] - } - end - - test "returns a list of Broadway.Message with :data and :acknowledger set", %{opts: base_opts} do - {:ok, opts} = ExAwsClient.init(base_opts) - [message1, message2] = ExAwsClient.receive_messages(10, opts) - - assert message1.data == "Message 1" - assert message2.data == "Message 2" - - assert message1.acknowledger == - {ExAwsClient, opts.ack_ref, - %{receipt: %{id: "Id_1", receipt_handle: "ReceiptHandle_1"}}} - end - - test "add message_id, receipt_handle and md5_of_body to metadata", %{opts: base_opts} do - {:ok, opts} = ExAwsClient.init(base_opts) - [%{metadata: metadata} | _] = ExAwsClient.receive_messages(10, opts) - - assert metadata.message_id == "Id_1" - assert metadata.receipt_handle == "ReceiptHandle_1" - assert metadata.md5_of_body == "fake_md5" - end - - test "add attributes to metadata", %{opts: base_opts} do - {:ok, opts} = Keyword.put(base_opts, :attribute_names, :all) |> ExAwsClient.init() - - [%{metadata: metadata_1}, %{metadata: metadata_2} | _] = - ExAwsClient.receive_messages(10, opts) - - assert metadata_1.attributes == %{"sender_id" => 13, "approximate_receive_count" => 5} - assert metadata_2.attributes == [] - end - - test "add message_attributes to metadata", %{opts: base_opts} do - {:ok, opts} = Keyword.put(base_opts, :message_attribute_names, :all) |> ExAwsClient.init() - - [%{metadata: metadata_1}, %{metadata: metadata_2} | _] = - ExAwsClient.receive_messages(10, opts) - - assert metadata_1.message_attributes == %{ - "TestStringAttribute" => %{ - name: "TestStringAttribute", - data_type: "String", - string_value: "Test", - binary_value: "", - value: "Test" - } - } - - assert metadata_2.message_attributes == [] - end - - test "if the request fails, returns an empty list and log the error", %{opts: base_opts} do - {:ok, opts} = - base_opts - |> put_in([:config, :http_client], FakeHttpClientWithError) - |> ExAwsClient.init() - - assert capture_log(fn -> - assert ExAwsClient.receive_messages(10, opts) == [] - end) =~ - "[error] Unable to fetch events from AWS queue my_queue. Reason: \"Fake error\"" - end - - test "send a SQS/ReceiveMessage request with default options", %{opts: base_opts} do - {:ok, opts} = ExAwsClient.init(base_opts) - ExAwsClient.receive_messages(10, opts) - - assert_received {:http_request_called, %{body: body, url: url}} - assert body == "Action=ReceiveMessage&MaxNumberOfMessages=10&QueueUrl=my_queue" - assert url == "https://sqs.us-east-1.amazonaws.com/" - end - - test "request with custom :wait_time_seconds", %{opts: base_opts} do - {:ok, opts} = base_opts |> Keyword.put(:wait_time_seconds, 0) |> ExAwsClient.init() - ExAwsClient.receive_messages(10, opts) - - assert_received {:http_request_called, %{body: body, url: _url}} - assert body =~ "WaitTimeSeconds=0" - end - - test "request with custom :max_number_of_messages", %{opts: base_opts} do - {:ok, opts} = base_opts |> Keyword.put(:max_number_of_messages, 5) |> ExAwsClient.init() - ExAwsClient.receive_messages(10, opts) - - assert_received {:http_request_called, %{body: body, url: _url}} - assert body =~ "MaxNumberOfMessages=5" - end - - test "request with custom :config options", %{opts: base_opts} do - config = - Keyword.merge(base_opts[:config], - scheme: "http://", - host: "localhost", - port: 9324 - ) - - {:ok, opts} = Keyword.put(base_opts, :config, config) |> ExAwsClient.init() - - ExAwsClient.receive_messages(10, opts) - - assert_received {:http_request_called, %{url: url}} - assert url == "http://localhost:9324/" - end - end - - describe "ack/3" do - setup do - %{ - opts: [ - # will be injected by broadway at runtime - broadway: [name: :Broadway3], - queue_url: "my_queue", - config: [ - http_client: FakeHttpClient, - access_key_id: "FAKE_ID", - secret_access_key: "FAKE_KEY" - ], - on_success: :ack, - on_error: :noop - ] - } - end - - test "send a SQS/DeleteMessageBatch request", %{opts: base_opts} do - {:ok, opts} = ExAwsClient.init(base_opts) - ack_data_1 = %{receipt: %{id: "1", receipt_handle: "abc"}} - ack_data_2 = %{receipt: %{id: "2", receipt_handle: "def"}} - - fill_persistent_term(opts.ack_ref, base_opts) - - ExAwsClient.ack( - opts.ack_ref, - [ - %Message{acknowledger: {ExAwsClient, opts.ack_ref, ack_data_1}, data: nil}, - %Message{acknowledger: {ExAwsClient, opts.ack_ref, ack_data_2}, data: nil} - ], - [] - ) - - assert_received {:http_request_called, %{body: body, url: url}} - - assert body == - "Action=DeleteMessageBatch" <> - "&DeleteMessageBatchRequestEntry.1.Id=1&DeleteMessageBatchRequestEntry.1.ReceiptHandle=abc" <> - "&DeleteMessageBatchRequestEntry.2.Id=2&DeleteMessageBatchRequestEntry.2.ReceiptHandle=def&QueueUrl=my_queue" - - assert url == "https://sqs.us-east-1.amazonaws.com/" - end - - test "request with custom :on_success and :on_failure", %{opts: base_opts} do - {:ok, opts} = ExAwsClient.init(base_opts ++ [on_success: :noop, on_failure: :ack]) - - :persistent_term.put(opts.ack_ref, %{ - queue_url: opts[:queue_url], - config: opts[:config], - on_success: opts[:on_success], - on_failure: opts[:on_failure] - }) - - ack_data_1 = %{receipt: %{id: "1", receipt_handle: "abc"}} - ack_data_2 = %{receipt: %{id: "2", receipt_handle: "def"}} - ack_data_3 = %{receipt: %{id: "3", receipt_handle: "ghi"}} - ack_data_4 = %{receipt: %{id: "4", receipt_handle: "jkl"}} - - message1 = %Message{acknowledger: {ExAwsClient, opts.ack_ref, ack_data_1}, data: nil} - message2 = %Message{acknowledger: {ExAwsClient, opts.ack_ref, ack_data_2}, data: nil} - message3 = %Message{acknowledger: {ExAwsClient, opts.ack_ref, ack_data_3}, data: nil} - message4 = %Message{acknowledger: {ExAwsClient, opts.ack_ref, ack_data_4}, data: nil} - - ExAwsClient.ack( - opts.ack_ref, - [ - message1, - message2 |> Message.configure_ack(on_success: :ack) - ], - [ - message3, - message4 |> Message.configure_ack(on_failure: :noop) - ] - ) - - assert_received {:http_request_called, %{body: body}} - - assert body == - "Action=DeleteMessageBatch" <> - "&DeleteMessageBatchRequestEntry.1.Id=2&DeleteMessageBatchRequestEntry.1.ReceiptHandle=def" <> - "&DeleteMessageBatchRequestEntry.2.Id=3&DeleteMessageBatchRequestEntry.2.ReceiptHandle=ghi&QueueUrl=my_queue" - end - - test "request with custom :config options", %{opts: base_opts} do - config = - Keyword.merge(base_opts[:config], - scheme: "http://", - host: "localhost", - port: 9324 - ) - - {:ok, opts} = Keyword.put(base_opts, :config, config) |> ExAwsClient.init() - - :persistent_term.put(opts.ack_ref, %{ - queue_url: opts[:queue_url], - config: opts[:config], - on_success: opts[:on_success], - on_failure: opts[:on_failure] - }) - - ack_data = %{receipt: %{id: "1", receipt_handle: "abc"}} - message = %Message{acknowledger: {ExAwsClient, opts.ack_ref, ack_data}, data: nil} - - ExAwsClient.ack(opts.ack_ref, [message], []) - - assert_received {:http_request_called, %{url: url}} - assert url == "http://localhost:9324/" - end - - test "request with :nack strategy", %{opts: base_opts} do - {:ok, opts} = ExAwsClient.init(base_opts ++ [on_failure: {:nack, 10}]) - - :persistent_term.put(opts.ack_ref, %{ - queue_url: opts[:queue_url], - config: opts[:config], - on_success: opts[:on_success], - on_failure: opts[:on_failure] - }) - - ack_data = %{receipt: %{id: "1", receipt_handle: "abc"}} - message = %Message{acknowledger: {ExAwsClient, opts.ack_ref, ack_data}, data: nil} - - ExAwsClient.ack(opts.ack_ref, [], [message]) - - assert_received {:http_request_called, %{body: body}} - - assert body == - "Action=ChangeMessageVisibilityBatch" <> - "&ChangeMessageVisibilityBatchRequestEntry.1.Id=1" <> - "&ChangeMessageVisibilityBatchRequestEntry.1.ReceiptHandle=abc" <> - "&ChangeMessageVisibilityBatchRequestEntry.1.VisibilityTimeout=10" <> - "&QueueUrl=my_queue" - end - end - - defp fill_persistent_term(ack_ref, base_opts) do - :persistent_term.put(ack_ref, %{ - queue_url: base_opts[:queue_url], - config: base_opts[:config], - on_success: base_opts[:on_success] || :ack, - on_failure: base_opts[:on_failure] || :noop - }) - end -end From 824e52a4478ee27dbd20a55002ad7303d947761d Mon Sep 17 00:00:00 2001 From: Milad Rastian Date: Wed, 12 Aug 2026 09:02:56 +0200 Subject: [PATCH 5/7] update CHANGELOG --- CHANGELOG.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c3b94b..ea85d87 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,29 @@ # Changelog +## Unreleased + + * Replace the default `BroadwaySQS.ExAwsClient` with the Req-based + `BroadwaySQS.ReqClient`. + * Implement the SQS JSON API requests used by Broadway SQS with Req and AWS + * `ReceiveMessage` + * `DeleteMessageBatch` + * `ChangeMessageVisibilityBatch` + * Support credentials discovered through `aws_credentials` + * Remove the `ex_aws_sqs`, `ex_aws`, `hackney`, and `saxy` dependencies. + * Update the documentation and example application to use the Req-based + client. + +### Breaking changes + + * `BroadwaySQS.ExAwsClient` has been removed. The default client is now + `BroadwaySQS.ReqClient`. + * ExAws configuration is no longer used. Configure the AWS region with the + producer `:config` option, and provide credentials through + `aws_credentials` or the producer configuration options. + * Applications using `BroadwaySQS.ExAwsClient` directly or relying on + `ex_aws` application configuration must migrate to + `BroadwaySQS.ReqClient` and the new credential configuration. + ## v0.7.4 (2024-06-21) * Forward compatibility with Broadway v1.1 From 708ef719ac9abc519198ed8daccd440829966c84 Mon Sep 17 00:00:00 2001 From: Milad Rastian Date: Wed, 12 Aug 2026 23:45:24 +0200 Subject: [PATCH 6/7] update deps --- mix.exs | 1 + mix.lock | 5 ----- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/mix.exs b/mix.exs index a9dacf9..55c1647 100644 --- a/mix.exs +++ b/mix.exs @@ -31,6 +31,7 @@ defmodule BroadwaySqs.MixProject do {:aws_credentials, "~> 1.0"}, {:nimble_options, "~> 0.3.7 or ~> 0.4 or ~> 1.0"}, {:telemetry, "~> 0.4.3 or ~> 1.0"}, + {:plug, "~> 1.18", only: :test}, {:ex_doc, ">= 0.19.0", only: :docs} ] end diff --git a/mix.lock b/mix.lock index dbf198b..57c0f3f 100644 --- a/mix.lock +++ b/mix.lock @@ -1,9 +1,6 @@ %{ "aws_credentials": {:hex, :aws_credentials, "1.1.1", "4a28d7d2c01956dd9a2cc52a7edd5cb1e58343766715753d4b8d0d043771b95c", [:rebar3], [{:eini, "~> 2.2.5", [hex: :eini_beam, repo: "hexpm", optional: false]}, {:iso8601, "~> 1.3.4", [hex: :iso8601, repo: "hexpm", optional: false]}, {:jsx, "~> 3.1.0", [hex: :jsx, repo: "hexpm", optional: false]}], "hexpm", "8655e0e3c82c5ad659729ea8717ae1f0a087ea268e15aefa674fc3d781c1e9e2"}, "broadway": {:hex, :broadway, "1.3.0", "f75f6376159b74f55c5ba2629dac613e4fd79d9e71148ab5fbac8fdd7c999d2a", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.3.7 or ~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "bef3b4c5512d0072917b70239cbecf8f76a2587465a5b7c3e2b9ae18b4bc405b"}, - "cowboy": {:hex, :cowboy, "2.18.0", "bff388eb4d6356cb3f88c26e65b515976bf04b401b805d31550c3e60eba8fe18", [:make, :rebar3], [{:cowlib, ">= 2.19.0 and < 3.0.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, ">= 1.8.0 and < 3.0.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "62d0b26abcf455054972b0da242389c69d5982ce5914afb8c344517f667b9600"}, - "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.4.0", "f239f68b588efa7707abce16a84d0d2acf3a0f50571f8bb7f56a15865aae820c", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7d98bac1ee4565d31b62d59f8823dfd8356a169e7fcbb83831b8a5397404c9de"}, - "cowlib": {:hex, :cowlib, "2.19.0", "c9d11c9d035472e27a740c9f327786c61ed209269b4be0260d59d3ec07b8949f", [:make, :rebar3], [], "hexpm", "6dc66e3135b229193ea4dcb14294e79520c923d391315c9c962ef0b4bea72356"}, "earmark_parser": {:hex, :earmark_parser, "1.4.46", "67607a0532e810c6f630a515c548d0b24949643f168cc556303bee4cf96105c7", [:mix], [], "hexpm", "9c44636e8a1c68c62f526b2dcd85d941dbbcee7ab82cf64ba06ce28bef8e89f5"}, "eini": {:hex, :eini_beam, "2.2.5", "28fa0a4eb7ff885cc388877b73fbd56fcca3b2eae4d81e47c47f317b900da1da", [:rebar3], [], "hexpm", "511e9207649f3becb5d945f1813615987899cf78fa130f92be07193a6a74ecb8"}, "ex_doc": {:hex, :ex_doc, "0.40.3", "4a972ffe64bc07dc605af487e98fc19b72a4185f55ca031b94c0552d6071c1d9", [:mix], [{:earmark_parser, "~> 1.4.44", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "2756e357742fecd9749b489b85d67c9ce99c465f2e75728d9e6dc8d704b973de"}, @@ -22,9 +19,7 @@ "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"}, "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, "plug": {:hex, :plug, "1.20.3", "56c480c633ec2ce10140e236e15233bf576e1d323887d7c96711bd02ab5160db", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "be266aee1b8536ef6409d58cf39a3121319f0ec47cfa1b24024485aa0e76ad76"}, - "plug_cowboy": {:hex, :plug_cowboy, "2.9.0", "87e21e0d9054ced99c36d128f49e3ea2cd8b745fffb97de50bff99706087af4f", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.18", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "2002bafba4f3a45b55a58e68d70211b153a7ed18d37edb1ceb6e96e7a92c422e"}, "plug_crypto": {:hex, :plug_crypto, "2.2.0", "144014737daaf485407f5ed77daeaad74d651b216a28c87543f8cc7043f8efc8", [:mix], [], "hexpm", "83a95744ab1c75876542b6fab135fcc176280e0f301a111c1f757fddcec95d2c"}, - "ranch": {:hex, :ranch, "1.8.1", "208169e65292ac5d333d6cdbad49388c1ae198136e4697ae2f474697140f201c", [:make, :rebar3], [], "hexpm", "aed58910f4e21deea992a67bf51632b6d60114895eb03bb392bb733064594dd0"}, "req": {:hex, :req, "0.7.2", "364eae2e5f5c984f2dac6d71c07f8c8c89ce0bc49c4d746dacb7a306823020de", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:finch, "~> 0.21", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "c9cdfa276b05d8db2a27fda5d233e6858b764d47189d76cbb186e130a871ae0b"}, "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, } From 17c68512d9e1b3f56df52061d99c198f76931dea Mon Sep 17 00:00:00 2001 From: Milad Rastian Date: Wed, 12 Aug 2026 23:55:45 +0200 Subject: [PATCH 7/7] make BroadwaySQS.ReqClient as default sqs client --- CHANGELOG.md | 1 + examples/sqs_example/config/config.exs | 1 - lib/broadway_sqs/producer.ex | 4 +++- test/broadway_sqs/producer_test.exs | 10 ++++++++++ 4 files changed, 14 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea85d87..c7ce934 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ * Replace the default `BroadwaySQS.ExAwsClient` with the Req-based `BroadwaySQS.ReqClient`. + * Make `BroadwaySQS.ReqClient` as default sqs_client. * Implement the SQS JSON API requests used by Broadway SQS with Req and AWS * `ReceiveMessage` * `DeleteMessageBatch` diff --git a/examples/sqs_example/config/config.exs b/examples/sqs_example/config/config.exs index d8231dd..fc5f7db 100644 --- a/examples/sqs_example/config/config.exs +++ b/examples/sqs_example/config/config.exs @@ -3,7 +3,6 @@ use Mix.Config config :broadway_sqs_example, producer_module: {BroadwaySQS.Producer, - sqs_client: BroadwaySQS.ReqClient, config: [ # access_key_id: "YOUR_AWS_ACCESS_KEY_ID", # secret_access_key: "YOUR_AWS_SECRET_ACCESS_KEY" diff --git a/lib/broadway_sqs/producer.ex b/lib/broadway_sqs/producer.ex index f680e45..4be5fdc 100644 --- a/lib/broadway_sqs/producer.ex +++ b/lib/broadway_sqs/producer.ex @@ -174,10 +174,12 @@ defmodule BroadwaySQS.Producer do @behaviour Producer + @default_sqs_client BroadwaySQS.ReqClient + @impl true def init(opts) do receive_interval = opts[:receive_interval] - sqs_client = opts[:sqs_client] + sqs_client = Keyword.get(opts, :sqs_client, @default_sqs_client) {:ok, client_opts} = sqs_client.init(opts) {:producer, diff --git a/test/broadway_sqs/producer_test.exs b/test/broadway_sqs/producer_test.exs index a6c6ae9..07e1e3b 100644 --- a/test/broadway_sqs/producer_test.exs +++ b/test/broadway_sqs/producer_test.exs @@ -95,6 +95,16 @@ defmodule BroadwaySQS.BroadwaySQS.ProducerTest do end) end + test "uses ReqClient when sqs_client is not provided" do + assert {:producer, %{sqs_client: {BroadwaySQS.ReqClient, client_opts}}} = + BroadwaySQS.Producer.init( + queue_url: "https://sqs.amazonaws.com/0000000000/my_queue", + broadway: [name: __MODULE__] + ) + + assert client_opts[:queue_url] == "https://sqs.amazonaws.com/0000000000/my_queue" + end + test "when the queue url is nil" do assert_raise( ArgumentError,