Implement streaming API

.github/workflows/ci.yml

@@ -38,11 +38,18 @@ jobs:
   dialyzer:
     runs-on: ubuntu-latest
 
-    container:
-      image: bitwalker/alpine-elixir:1.10.4
+    strategy:
+      matrix:
+        elixir: ["1.13.x"]
+        otp: ["24.x"]
 
     steps:
       - uses: actions/checkout@v1
+      - name: Set up Elixir
+        uses: erlef/setup-beam@v1
+        with:
+          elixir-version: ${{ matrix.elixir }}
+          otp-version: ${{ matrix.otp }}
       - uses: actions/cache@v2
         with:
           path: _build

lib/supplement/storages/disk.ex

@@ -3,6 +3,14 @@ defmodule Capsule.Storages.Disk do
 
   @behaviour Storage
 
+  @impl Storage
+
+  def stream!(id, opts \\ []) do
+    opts
+    |> path_in_root(id)
+    |> File.stream!()
+  end
+
   @impl Storage
   def put(upload, opts \\ []) do
     with path <- Path.join(opts[:prefix] || "/", Upload.name(upload)),

lib/supplement/storages/ram.ex

@@ -30,6 +30,13 @@ defmodule Capsule.Storages.RAM do
   def read(id, _opts \\ []),
     do: {:ok, id |> decode_pid! |> StringIO.contents() |> elem(0)}
 
+  @impl Storage
+  def stream!(id, _opts \\ []) do
+    id
+    |> decode_pid!
+    |> IO.stream(:line)
+  end
+
   defp decompose_id(id), do: String.split(id, "/", parts: 2)
 
   defp decode_pid!(id) do

lib/supplement/storages/s3.ex

@@ -40,6 +40,14 @@ defmodule Capsule.Storages.S3 do
     end
   end
 
+  @impl Storage
+  def stream!(id, opts \\ []) do
+    opts
+    |> config(:bucket)
+    |> Client.download_file(id, :memory)
+    |> ex_aws_module().stream!()
+  end
+
   defp config(opts, key) do
     Application.fetch_env!(:capsule, __MODULE__)
     |> Keyword.merge(opts)

mix.exs

@@ -30,7 +30,7 @@ defmodule Supplement.MixProject do
   # Run "mix help deps" to learn about dependencies.
   defp deps do
     [
-      {:capsule, "~> 0.9"},
+      {:capsule, github: "elixir-capsule/capsule"},
       {:ex_aws, "~> 2.0", optional: true},
       {:ex_aws_s3, "~> 2.0", optional: true},
       {:mox, "~> 1.0", only: [:test]},

mix.lock

@@ -1,14 +1,14 @@
 %{
   "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"},
-  "capsule": {:hex, :capsule, "0.9.0", "bafacd356462889cc6e816cd1c756727da563b88ad5f0811182771aca6772e61", [:mix], [], "hexpm", "acb4f8b05c510993d09c133eb2becadf1814e80249192b5e6e9951f2cf6f08c7"},
+  "capsule": {:git, "https://github.com/elixir-capsule/capsule.git", "5e59791447220ce3511249998f8e91c2bac3f652", []},
   "cowboy": {:hex, :cowboy, "2.9.0", "865dd8b6607e14cf03282e10e934023a1bd8be6f6bacf921a7e2a96d800cd452", [:make, :rebar3], [{:cowlib, "2.11.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "2c729f934b4e1aa149aff882f57c6372c15399a20d54f65c8d67bef583021bde"},
   "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.11.0", "0b9ff9c346629256c42ebe1eeb769a83c6cb771a6ee5960bd110ab0b9b872063", [:make, :rebar3], [], "hexpm", "2b3e9da0b21c4565751a6d4901c20d1b4cc25cbb7fd50d91d2ab6dd287bc86a9"},
   "dialyxir": {:hex, :dialyxir, "1.0.0", "6a1fa629f7881a9f5aaf3a78f094b2a51a0357c843871b8bc98824e7342d00a5", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "aeb06588145fac14ca08d8061a142d52753dbc2cf7f0d00fc1013f53f8654654"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.12", "b245e875ec0a311a342320da0551da407d9d2b65d98f7a9597ae078615af3449", [:mix], [], "hexpm", "711e2cc4d64abb7d566d43f54b78f7dc129308a63bc103fbd88550d2174b3160"},
   "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},
-  "ex_aws": {:hex, :ex_aws, "2.1.5", "f0f3f3e8a93e32b895be7981d34da14194effeae2ba9ba206e87d211f283fa62", [: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", "0f0357f116fc3ca6e962c0be46cc42e9a16622ecd26fcc6c8b97b114d3f81949"},
-  "ex_aws_s3": {:hex, :ex_aws_s3, "2.0.2", "c0258bbdfea55de4f98f0b2f0ca61fe402cc696f573815134beb1866e778f47b", [:mix], [{:ex_aws, "~> 2.0", [hex: :ex_aws, repo: "hexpm", optional: false]}, {:sweet_xml, ">= 0.0.0", [hex: :sweet_xml, repo: "hexpm", optional: true]}], "hexpm", "0569f5b211b1a3b12b705fe2a9d0e237eb1360b9d76298028df2346cad13097a"},
+  "ex_aws": {:hex, :ex_aws, "2.5.0", "1785e69350b16514c1049330537c7da10039b1a53e1d253bbd703b135174aec3", [: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", "971b86e5495fc0ae1c318e35e23f389e74cf322f2c02d34037c6fc6d405006f1"},
+  "ex_aws_s3": {:hex, :ex_aws_s3, "2.5.2", "cee302b8e9ee198cc0d89f1de2a7d6a8921e1a556574476cf5590d2156590fe3", [:mix], [{:ex_aws, "~> 2.0", [hex: :ex_aws, repo: "hexpm", optional: false]}, {:sweet_xml, ">= 0.0.0", [hex: :sweet_xml, repo: "hexpm", optional: true]}], "hexpm", "cc5bd945a22a99eece4721d734ae2452d3717e81c357a781c8574663254df4a1"},
   "ex_doc": {:hex, :ex_doc, "0.23.0", "a069bc9b0bf8efe323ecde8c0d62afc13d308b1fa3d228b65bca5cf8703a529d", [:mix], [{:earmark_parser, "~> 1.4.0", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}], "hexpm", "f5e2c4702468b2fd11b10d39416ddadd2fcdd173ba2a0285ebd92c39827a5a16"},
   "makeup": {:hex, :makeup, "1.0.5", "d5a830bc42c9800ce07dd97fa94669dfb93d3bf5fcf6ea7a0c67b2e0e4a7f26c", [:mix], [{:nimble_parsec, "~> 0.5 or ~> 1.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cfa158c02d3f5c0c665d0af11512fed3fba0144cf1aadee0f2ce17747fba2ca9"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.15.0", "98312c9f0d3730fde4049985a1105da5155bfe5c11e47bdc7406d88e01e4219b", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.1", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "75ffa34ab1056b7e24844c90bfc62aaf6f3a37a15faa76b07bc5eba27e4a8b4a"},

test/storages/disk_test.exs

@@ -56,4 +56,14 @@ defmodule Capsule.Storages.DiskTest do
       on_exit(fn -> File.rm!("tmp/path") end)
     end
   end
+
+  describe "stream!/1" do
+    test "returns enum of file contents" do
+      File.write!("tmp/path", "data")
+
+      assert "data" = "path" |> Disk.stream!() |> Enum.join()
+
+      on_exit(fn -> File.rm!("tmp/path") end)
+    end
+  end
 end

test/storages/ram_test.exs

@@ -49,6 +49,14 @@ defmodule Capsule.Storages.RAMTest do
     end
   end
 
+  describe "stream/1" do
+    setup :build_ram_file
+
+    test "result can be enumerated", %{id: id} do
+      assert ["some data"] = id |> RAM.stream!() |> Enum.to_list()
+    end
+  end
+
   defp build_ram_file(_context) do
     {:ok, pid} = StringIO.open("some data")
 

test/storages/s3_test.exs

@@ -59,4 +59,16 @@ defmodule Capsule.Storages.S3Test do
       assert {:ok, _} = S3.read("fake", bucket: "other")
     end
   end
+
+  describe "stream!/1" do
+    test "returns ex aws stream" do
+      stub(ExAwsMock, :stream!, fn _ ->
+        Stream.transform([0, 1], 0, fn i, acc ->
+          if acc < 3, do: {[i], acc + 1}, else: {:halt, acc}
+        end)
+      end)
+
+      assert [0, 1] = "fake" |> S3.stream!() |> Enum.to_list()
+    end
+  end
 end