Add Membrane.AWS.S3.Source

.gitignore

@@ -1,3 +1,5 @@
+
+.env
 compile_commands.json
 .gdb_history
 bundlex.sh

README.md

@@ -1,35 +1,42 @@
-# Membrane Template Plugin
+# Membrane AWS Plugin
 
-[![Hex.pm](https://img.shields.io/hexpm/v/membrane_template_plugin.svg)](https://hex.pm/packages/membrane_template_plugin)
-[![API Docs](https://img.shields.io/badge/api-docs-yellow.svg?style=flat)](https://hexdocs.pm/membrane_template_plugin)
-[![CircleCI](https://circleci.com/gh/membraneframework/membrane_template_plugin.svg?style=svg)](https://circleci.com/gh/membraneframework/membrane_template_plugin)
+[![Hex.pm](https://img.shields.io/hexpm/v/membrane_aws_plugin.svg)](https://hex.pm/packages/membrane_aws_plugin)
+[![API Docs](https://img.shields.io/badge/api-docs-yellow.svg?style=flat)](https://hexdocs.pm/membrane_aws_plugin)
+[![CircleCI](https://circleci.com/gh/jellyfish-dev/membrane_aws_plugin.svg?style=svg)](https://circleci.com/gh/jellyfish-dev/membrane_aws_plugin)
+[![codecov](https://codecov.io/gh/jellyfish-dev/membrane_aws_plugin/branch/main/graph/badge.svg?token=ANWFKV2EDP)](https://codecov.io/gh/jellyfish-dev/membrane_aws_plugin)
 
-This repository contains a template for new plugins.
+This repository contains Membrane element that interacts with AWS.
+Currently implemented are:
+- `Membrane.AWS.S3.Source`
 
-Check out different branches for other flavors of this template.
 
 It's a part of the [Membrane Framework](https://membrane.stream).
 
 ## Installation
 
-The package can be installed by adding `membrane_template_plugin` to your list of dependencies in `mix.exs`:
+The package can be installed by adding `membrane_aws_plugin` to your list of dependencies in `mix.exs`:
 
 ```elixir
 def deps do
   [
-    {:membrane_template_plugin, "~> 0.1.0"}
+    {:membrane_aws_plugin, "~> 0.1.0"}
   ]
 end
 ```
 
-## Usage
+## Usage example
 
-TODO
+The `example/` folder contains an example usage of `Membrane.AWS.S3.Source`.
+
+This demo downloads a file from S3 and saves it locally. It requires that you set these environment variables: `BUCKET`, `FILE_PATH`, `ACCESS_KEY_ID`, `SECRET_ACCESS_KEY`, `REGION`.
+```bash
+$ elixir examples/source_example.exs
+```
 
 ## Copyright and License
 
-Copyright 2020, [Software Mansion](https://swmansion.com/?utm_source=git&utm_medium=readme&utm_campaign=membrane_template_plugin)
+Copyright 2024, [Software Mansion](https://swmansion.com/?utm_source=git&utm_medium=readme&utm_campaign=membrane_aws_plugin)
 
-[![Software Mansion](https://logo.swmansion.com/logo?color=white&variant=desktop&width=200&tag=membrane-github)](https://swmansion.com/?utm_source=git&utm_medium=readme&utm_campaign=membrane_template_plugin)
+[![Software Mansion](https://logo.swmansion.com/logo?color=white&variant=desktop&width=200&tag=membrane-github)](https://swmansion.com/?utm_source=git&utm_medium=readme&utm_campaign=membrane_aws_plugin)
 
 Licensed under the [Apache License, Version 2.0](LICENSE)

example/source_example.exs

@@ -0,0 +1,48 @@
+Mix.install([
+  {:membrane_aws_plugin, path: __DIR__ |> Path.join("..") |> Path.expand()},
+  :membrane_core,
+  :membrane_file_plugin,
+  :hackney
+])
+
+# In this example, the pipeline will download file from S3 and save it to a local file
+
+defmodule Example do
+  use Membrane.Pipeline
+
+  @impl true
+  def handle_init(_ctx, options) do
+    structure = [
+      child(:s3_source, %Membrane.AWS.S3.Source{
+        bucket: System.fetch_env!("BUCKET"),
+        path: System.fetch_env!("FILE_PATH"),
+        aws_config: [
+          access_key_id: System.fetch_env!("ACCESS_KEY_ID"),
+          secret_access_key: System.fetch_env!("SECRET_ACCESS_KEY"),
+          region: System.fetch_env!("REGION")
+        ]
+      })
+      |> child(:file_sink, %Membrane.File.Sink{location: "source.example"})
+    ]
+
+    {[spec: structure], %{}}
+  end
+
+  # Next two functions are only a logic for terminating a pipeline when it's done, you don't need to worry
+  @impl true
+  def handle_element_end_of_stream(:file_sink, _pad, _ctx, state) do
+    {[terminate: :normal], state}
+  end
+
+  def handle_element_end_of_stream(_element, _pad, _ctx, state) do
+    {[], state}
+  end
+end
+
+{:ok, _supervisor, pipeline} = Membrane.Pipeline.start_link(Example)
+monitor_ref = Process.monitor(pipeline)
+
+receive do
+  {:DOWN, ^monitor_ref, :process, _pid, _reason} ->
+    :ok
+end

lib/s3/s3_source.ex

@@ -0,0 +1,110 @@
+defmodule Membrane.AWS.S3.Source do
+  @moduledoc """
+  Element that reads file from a S3 bucket and sends it through the output pad.
+  """
+  use Membrane.Source
+
+  alias Membrane.{Buffer, RemoteStream}
+
+  @default_max_concurrency 8
+  @task_timeout_milliseconds 60_000
+
+  def_options aws_config: [
+                spec: Keyword.t(),
+                description: """
+                Config to ExAWS. For more information refer to [`ExAws.Config`](https://github.com/ex-aws/ex_aws/blob/main/lib/ex_aws/config.ex).
+                """,
+                default: []
+              ],
+              bucket: [
+                spec: binary(),
+                description: "Name of bucket"
+              ],
+              path: [
+                spec: binary(),
+                description: "Path to file in bucket"
+              ],
+              opts: [
+                spec: [
+                  max_concurrency: pos_integer(),
+                  chunk_size: pos_integer(),
+                  timeout: pos_integer()
+                ],
+                description: "File download opts",
+                default: []
+              ],
+              cached_chunks_number: [
+                spec: non_neg_integer(),
+                description: "Number of chunks cached before demand",
+                default: 10
+              ]
+
+  def_output_pad :output, accepted_format: %RemoteStream{type: :bytestream}, flow_control: :manual
+
+  @impl true
+  def handle_init(_context, opts) do
+    state =
+      Map.merge(opts, %{
+        aws_config: ExAws.Config.new(:s3, opts.aws_config),
+        chunks_stream: nil,
+        chunks: :queue.new()
+      })
+
+    {[], state}
+  end
+
+  @impl true
+  def handle_playing(_ctx, state) do
+    chunks_stream = ExAws.S3.Download.build_chunk_stream(state, state.aws_config)
+
+    {take_chunks, chunks_stream} = Enum.split(chunks_stream, state.cached_chunks_number)
+
+    downloaded_chunks =
+      take_chunks
+      |> Task.async_stream(
+        &download_chunk(state, &1),
+        max_concurrency: state.opts[:max_concurrency] || @default_max_concurrency,
+        timeout: state.opts[:timeout] || @task_timeout_milliseconds
+      )
+      |> Enum.map(fn {:ok, chunk} -> chunk end)
+      |> Enum.reduce(state.chunks, fn chunk, acc -> :queue.in(chunk, acc) end)
+
+    {[stream_format: {:output, %RemoteStream{type: :bytestream}}],
+     %{state | chunks_stream: chunks_stream, chunks: downloaded_chunks}}
+  end
+
+  @impl true
+  def handle_demand(_pad, _size, _unit, _ctx, state) do
+    {chunks, chunks_stream} = update_chunks(state)
+
+    case :queue.out(chunks) do
+      {:empty, _chunks} ->
+        {[end_of_stream: :output], state}
+
+      {{:value, payload}, chunks} ->
+        {[buffer: {:output, %Buffer{payload: payload}}] ++ [redemand: :output],
+         %{state | chunks_stream: chunks_stream, chunks: chunks}}
+    end
+  end
+
+  defp update_chunks(state) do
+    {boundaries, chunks_stream} = Enum.split(state.chunks_stream, 1)
+
+    chunks =
+      case boundaries do
+        [boundaries] ->
+          chunk = download_chunk(state, boundaries)
+          :queue.in(chunk, state.chunks)
+
+        [] ->
+          state.chunks
+      end
+
+    {chunks, chunks_stream}
+  end
+
+  defp download_chunk(state, boundaries) do
+    {_start_byte, chunk} = ExAws.S3.Download.get_chunk(state, boundaries, state.aws_config)
+    chunk
+  end
+end