627 support for streaming via stdio

.formatter.exs

@@ -1,6 +1,6 @@
 [
   inputs: [
-    "{lib,test,config}/**/*.{ex,exs}",
+    "{lib,test,config,examples}/**/*.{ex,exs}",
     ".formatter.exs",
     "*.exs"
   ],

config/config.exs

@@ -1,5 +1,7 @@
 import Config
 
+config :logger, backends: []
+
 if config_env() == :test do
   config :membrane_file_plugin, :file_impl, Membrane.File.CommonMock
 else

examples/file_to_pipe.exs

@@ -0,0 +1,78 @@
+# defmodule Mix.Shell.Ultraquiet do
+#   @behaviour Mix.Shell
+
+#   def print_app, do: :ok
+
+#   def error(_message), do: :ok
+
+#   def info(_message), do: :ok
+
+#   def print_app(_message), do: :ok
+
+#   def prompt(_message), do: :ok
+
+#   def yes?(_message, _opts \\ []), do: :ok
+
+#   def cmd(command, opts \\ []) do
+#     Mix.Shell.cmd(command, opts, fn data -> data end)
+#   end
+# end
+
+# {:ok, config} = :logger.get_handler_config(:default)
+# config_dev = put_in(config, [:config, :type], :standard_error)
+
+# :ok =
+#   :logger.set_handler_config(:default, put_in(config, [:config, :type], :standard_error))
+
+# :logger.remove_handler(:default)
+# :ok =
+#   :logger.add_handler(:default, :logger_std_h, put_in(config, [:config, :type], :standard_error))
+
+# Logger.configure(device: :standard_error)
+# Logger.Backends.Internal.configure(device: :standard_error)
+
+Mix.start()
+Mix.shell(Mix.Shell.Quiet)
+
+Mix.install(
+  [
+    {:membrane_file_plugin, path: "."}
+  ]
+  # force: true
+)
+
+defmodule FileExamplePipeline do
+  @doc """
+  Example pipeline that reads its source code file and outputs it to /tmp/test.
+  """
+
+  use Membrane.Pipeline
+
+  @doc false
+  @impl true
+  def handle_init(_ctx, %{target: target, input: input}) do
+    spec =
+      child(%Membrane.File.Source{location: input})
+      |> child(:sink, %Membrane.File.Sink{location: :stdout})
+
+    {[spec: spec], %{target: target}}
+  end
+
+  @impl true
+  def handle_element_end_of_stream(:sink, :input, _ctx, state) do
+    send(state.target, :done)
+    {[], state}
+  end
+end
+
+# # redirect membrane logs to stderr
+Membrane.File.Sink.redirect_logs()
+
+[input] = System.argv()
+
+{:ok, _supervisor, pid} =
+  Membrane.Pipeline.start_link(FileExamplePipeline, %{input: input, target: self()})
+
+receive do
+  :done -> Membrane.Pipeline.terminate(pid)
+end

examples/pipe_to_file.exs

@@ -0,0 +1,47 @@
+Mix.start()
+Mix.shell(Mix.Shell.Quiet)
+
+Mix.install(
+  [
+    {:membrane_file_plugin, path: "."}
+  ]
+  # force: true
+)
+
+[output, chunk_size_str | _] = System.argv()
+{chunk_size, ""} = Integer.parse(chunk_size_str)
+
+defmodule PipeToFile do
+  @doc """
+  Example pipeline that reads its source code file and outputs it to /tmp/test.
+  """
+
+  use Membrane.Pipeline
+
+  @doc false
+  @impl true
+  def handle_init(_ctx, %{target: target, output: output, chunk_size: chunk_size}) do
+    spec =
+      child(%Membrane.File.Source{location: :stdin, chunk_size: chunk_size})
+      |> child(:sink, %Membrane.File.Sink{location: output})
+
+    {[spec: spec], %{target: target}}
+  end
+
+  @impl true
+  def handle_element_end_of_stream(:sink, :input, _ctx, state) do
+    send(state.target, :done)
+    {[], state}
+  end
+end
+
+{:ok, _supervisor, pid} =
+  Membrane.Pipeline.start_link(PipeToFile, %{
+    target: self(),
+    output: output,
+    chunk_size: chunk_size
+  })
+
+receive do
+  :done -> Membrane.Pipeline.terminate(pid)
+end

examples/sink_and_source.exs

@@ -1,31 +1,34 @@
 Mix.install([
-  {:membrane_core, "~> 0.11"},
+  {:membrane_core, "~> 1.0"},
   {:membrane_file_plugin, path: Path.expand(__DIR__ <> "/..")}
 ])
 
 defmodule FileExamplePipeline do
+  @doc """
+  Example pipeline that reads its source code file and outputs it to /tmp/test.
+  """
+
   use Membrane.Pipeline
 
   @doc false
   @impl true
   def handle_init(_ctx, target) do
-    spec = [
+    spec =
       child(:file_src, %Membrane.File.Source{location: __ENV__.file})
       |> child(:file_sink, %Membrane.File.Sink{location: "/tmp/test"})
-    ]
 
-    {[spec: spec, playback: :playing], %{target: target}}
+    {[spec: spec], %{target: target}}
   end
 
   @impl true
-  def handle_element_end_of_stream({:file_sink, :input}, _ctx, state) do
+  def handle_element_end_of_stream(:file_sink, :input, _ctx, state) do
     send(state.target, :done)
     {[], state}
   end
 end
 
-{:ok, _supervisor_pid, pid} = FileExamplePipeline.start_link(self())
+{:ok, _supervisor_pid, pid} = Membrane.Pipeline.start_link(FileExamplePipeline, self())
 
 receive do
-  :done -> FileExamplePipeline.terminate(pid)
+  :done -> Membrane.Pipeline.terminate(pid)
 end

examples/sink_multi.exs

@@ -1,10 +1,16 @@
 Mix.install([
-  {:membrane_core, "~> 0.11"},
+  {:membrane_core, "~> 1.0"},
   {:membrane_file_plugin, path: Path.expand(__DIR__ <> "/..")}
 ])
 
 # Filter responsible for generating split events
 defmodule Splitter do
+  @moduledoc """
+  Receives buffer and splits it into two buffers
+  of size `head_size` and `buffer.size - head_size`,
+  sending a split event to multisink in between.
+  """
+
   use Membrane.Filter
 
   alias Membrane.Buffer
@@ -29,7 +35,7 @@ defmodule Splitter do
       buffer: {:output, %Buffer{payload: tail}}
     ]
 
-    { actions, %{split?: false}}
+    {actions, %{split?: false}}
   end
 
   def handle_buffer(:input, buffer, _ctx, %{split?: false}) do
@@ -40,6 +46,11 @@ end
 :ok = File.write!("input.bin", <<0::integer-unit(8)-size(1024)>>)
 
 defmodule SinkMultiExamplePipeline do
+  @moduledoc """
+  Example pipeline that reads a binary file
+  and performs a multisink split when sending it forward.
+  """
+
   use Membrane.Pipeline
 
   @doc false
@@ -51,22 +62,22 @@ defmodule SinkMultiExamplePipeline do
       |> child(:file_sink, %Membrane.File.Sink.Multi{location: "/tmp/output", extension: ".bin"})
     ]
 
-    {[spec: spec, playback: :playing], %{target: target}}
+    {[spec: spec], %{target: target}}
   end
 
   @impl true
-  def handle_element_end_of_stream({:file_sink, :input}, _ctx, state) do
+  def handle_element_end_of_stream(:file_sink, :input, _ctx, state) do
     send(state.target, :done)
     {[], state}
   end
 
-  def handle_element_end_of_stream(_other, _ctx, state) do
+  def handle_element_end_of_stream(_elem, _pad, _ctx, state) do
     {[], state}
   end
 end
 
-{:ok, _supervisor_pid, pid} = SinkMultiExamplePipeline.start_link(self())
+{:ok, _supervisor_pid, pid} = Membrane.Pipeline.start_link(SinkMultiExamplePipeline, self())
 
 receive do
-  :done -> SinkMultiExamplePipeline.terminate(pid)
+  :done -> Membrane.Pipeline.terminate(pid)
 end

lib/membrane_file/sink.ex

@@ -1,12 +1,19 @@
 defmodule Membrane.File.Sink do
   @moduledoc """
   Element that creates a file and stores incoming buffers there (in binary format).
+  Can also be used as a pipe to standard output by setting location to :stdout,
+  though this requires additional configuration.
 
   When `Membrane.File.SeekSinkEvent` is received, the element starts writing buffers starting
   from `position`. By default, it overwrites previously stored bytes. You can set `insert?`
   field of the event to `true` to start inserting new buffers without overwriting previous ones.
   Please note, that inserting requires rewriting the file, what negatively impacts performance.
   For more information refer to `Membrane.File.SeekSinkEvent` moduledoc.
+
+  Pipeline logs are directed to standard output by default. To separate them from the sink's output
+  we recommend redirecting the logger to standard error. For simple use cases using the default logger
+  configuration (like stand-alone scripts) this can be achieved by simply calling redirect_logs/1.
+  See examples/file_to_pipe.exs for a working example.
   """
   use Membrane.Sink
 
@@ -15,12 +22,33 @@ defmodule Membrane.File.Sink do
   @common_file Membrane.File.CommonFileBehaviour.get_impl()
 
   def_options location: [
-                spec: Path.t(),
-                description: "Path of the output file"
+                spec: Path.t() | :stdout,
+                description: "Path of the output file or :stdout"
               ]
 
   def_input_pad :input, flow_control: :manual, demand_unit: :buffers, accepted_format: _any
 
+  @spec redirect_logs() :: :ok
+  def redirect_logs() do
+    {:ok, config} = :logger.get_handler_config(:default)
+    :ok = :logger.remove_handler(:default)
+
+    :ok =
+      :logger.add_handler(
+        :default,
+        :logger_std_h,
+        put_in(config, [:config, :type], :standard_error)
+      )
+  end
+
+  @impl true
+  def handle_init(_ctx, %__MODULE__{location: :stdout}) do
+    {[],
+     %{
+       location: :stdout
+     }}
+  end
+
   @impl true
   def handle_init(_ctx, %__MODULE__{location: location}) do
     {[],
@@ -32,6 +60,11 @@ defmodule Membrane.File.Sink do
      }}
   end
 
+  @impl true
+  def handle_setup(_ctx, %{location: :stdout} = state) do
+    {[], state}
+  end
+
   @impl true
   def handle_setup(_ctx, %{location: location} = state) do
     fd = @common_file.open!(location, [:read, :write])
@@ -45,12 +78,23 @@ defmodule Membrane.File.Sink do
     {[demand: :input], state}
   end
 
+  @impl true
+  def handle_buffer(:input, buffer, _ctx, %{location: :stdout} = state) do
+    :ok = @common_file.write!(:stdio, buffer)
+    {[demand: :input], state}
+  end
+
   @impl true
   def handle_buffer(:input, buffer, _ctx, %{fd: fd} = state) do
     :ok = @common_file.write!(fd, buffer)
     {[demand: :input], state}
   end
 
+  @impl true
+  def handle_event(:input, %SeekSinkEvent{}, _ctx, %{location: :stdout} = _state) do
+    raise "Seek event not supported for :stdout sink"
+  end
+
   @impl true
   def handle_event(:input, %SeekSinkEvent{insert?: insert?, position: position}, _ctx, state) do
     state =
@@ -63,11 +107,21 @@ defmodule Membrane.File.Sink do
 
   def handle_event(pad, event, ctx, state), do: super(pad, event, ctx, state)
 
+  @impl true
+  def handle_end_of_stream(:input, _ctx, %{location: :stdout} = state) do
+    {[], state}
+  end
+
   @impl true
   def handle_end_of_stream(:input, _ctx, state) do
     {[], do_merge_and_close(state)}
   end
 
+  @impl true
+  def handle_terminate_request(_ctx, %{location: :stdout} = state) do
+    {[terminate: :normal], state}
+  end
+
   @impl true
   def handle_terminate_request(_ctx, state) do
     {[terminate: :normal], do_merge_and_close(state)}