add streaming support for list_object_versions

implemented stream_object_versions!/3 in lazy.ex to enable streaming
through paginated object version results using dual-marker pagination
(key_marker + version_id_marker). updated list_object_versions to support
streaming via ExAws.stream!() similar to list_objects and list_objects_v2.

tests now use per-test buckets to eliminate flakiness from shared state.
cleanup simplified to use list_object_versions which works for both
versioned and non-versioned buckets (non-versioned objects have version_id
of "null"). using request!() for cleanup makes failures visible.

Author: velimir

lib/ex_aws/s3.ex

@@ -229,12 +229,39 @@ defmodule ExAws.S3 do
     )
   end
 
-  @doc "List metadata about all versions of the objects in a bucket."
+  @type list_object_versions_opts :: [
+          {:delimiter, binary}
+          | {:key_marker, binary}
+          | {:version_id_marker, binary}
+          | {:max_keys, 0..1000}
+          | {:prefix, binary}
+          | {:encoding_type, binary}
+        ]
+
+  @doc """
+  List metadata about all versions of the objects in a bucket.
+
+  Can be streamed.
+
+  ## Examples
+  ```
+  S3.list_object_versions("my-bucket") |> ExAws.request
+
+  S3.list_object_versions("my-bucket") |> ExAws.stream!
+  S3.list_object_versions("my-bucket", prefix: "backup/") |> ExAws.stream!
+  ```
+  """
   @spec list_object_versions(bucket :: binary) :: ExAws.Operation.S3.t()
-  @spec list_object_versions(bucket :: binary, opts :: Keyword.t()) ::
+  @spec list_object_versions(bucket :: binary, opts :: list_object_versions_opts) ::
           ExAws.Operation.S3.t()
+  @params [:delimiter, :key_marker, :version_id_marker, :max_keys, :prefix, :encoding_type]
   def list_object_versions(bucket, opts \\ []) do
-    request(:get, bucket, "/", [resource: "versions", params: opts],
+    params =
+      opts
+      |> format_and_take(@params)
+
+    request(:get, bucket, "/", [resource: "versions", params: params],
+      stream_builder: &ExAws.S3.Lazy.stream_object_versions!(bucket, opts, &1),
       parser: &ExAws.S3.Parsers.parse_object_versions/1
     )
   end

lib/ex_aws/s3/lazy.ex

@@ -54,6 +54,37 @@ defmodule ExAws.S3.Lazy do
     )
   end
 
+  def stream_object_versions!(bucket, opts, config) do
+    request_fun = fn fun_opts ->
+      ExAws.S3.list_object_versions(bucket, Keyword.merge(opts, fun_opts))
+      |> ExAws.request!(config)
+      |> Map.get(:body)
+    end
+
+    Stream.resource(
+      fn -> {request_fun, []} end,
+      fn
+        :quit ->
+          {:halt, nil}
+
+        {fun, args} ->
+          case fun.(args) do
+            results = %{is_truncated: "true"} ->
+              {add_version_results(results),
+               {fun,
+                [
+                  key_marker: results[:next_key_marker],
+                  version_id_marker: results[:next_version_id_marker]
+                ]}}
+
+            results ->
+              {add_version_results(results), :quit}
+          end
+      end,
+      & &1
+    )
+  end
+
   def add_results(results, opts) do
     case Keyword.get(opts, :stream_prefixes, nil) do
       nil -> results.contents
@@ -68,4 +99,8 @@ defmodule ExAws.S3.Lazy do
   end
 
   def next_marker(%{next_marker: marker}), do: marker
+
+  def add_version_results(results) do
+    (results[:versions] || []) ++ (results[:delete_markers] || [])
+  end
 end