Implement Req Client

CHANGELOG.md

@@ -1,5 +1,30 @@
 # Changelog
 
+## Unreleased
+
+  * 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`
+    * `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

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
 ```

config/config.exs

@@ -0,0 +1,3 @@
+import Config
+
+import_config "#{config_env()}.exs"

config/dev.exs

@@ -0,0 +1 @@
+import Config

config/test.exs

@@ -0,0 +1,3 @@
+import Config
+
+config :aws_credentials, credential_providers: []

examples/sqs_example/config/config.exs

@@ -3,13 +3,14 @@ use Mix.Config
 config :broadway_sqs_example,
   producer_module:
     {BroadwaySQS.Producer,
-     sqs_client: BroadwaySQS.ExAwsClient,
      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"

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

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

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"},

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: [

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())}
 
@@ -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,
@@ -194,15 +196,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)