Merge pull request #157 from membraneframework/release/v0.24.0

Release membrane_rtp_plugin v0.24.0

Author: FelonEkonom

README.md

@@ -15,7 +15,7 @@ The package can be installed by adding `membrane_rtp_plugin` to your list of dep
 ```elixir
 def deps do
   [
-    {:membrane_rtp_plugin, "~> 0.23.2"},
+	  {:membrane_rtp_plugin, "~> 0.24.0"},
     {:ex_libsrtp, ">= 0.0.0"} # required only if SRTP/SRTCP support is needed
   ]
 end

lib/membrane/rtcp/parser.ex

@@ -11,12 +11,12 @@ defmodule Membrane.RTCP.Parser do
 
   def_input_pad :input,
     accepted_format: %RemoteStream{type: :packetized, content_format: cf} when cf in [nil, RTCP],
-    demand_mode: :auto
+    flow_control: :auto
 
-  def_output_pad :output, accepted_format: RTCP, demand_mode: :auto
+  def_output_pad :output, accepted_format: RTCP, flow_control: :auto
 
   def_output_pad :receiver_report_output,
-    mode: :push,
+    flow_control: :push,
     accepted_format: %RemoteStream{type: :packetized, content_format: RTCP}
 
   @impl true
@@ -36,7 +36,7 @@ defmodule Membrane.RTCP.Parser do
   end
 
   @impl true
-  def handle_process(:input, %Buffer{payload: payload, metadata: metadata}, _ctx, state) do
+  def handle_buffer(:input, %Buffer{payload: payload, metadata: metadata}, _ctx, state) do
     payload
     |> RTCP.Packet.parse()
     |> case do

lib/membrane/rtcp/receiver.ex

@@ -18,8 +18,8 @@ defmodule Membrane.RTCP.Receiver do
   alias Membrane.Time
   alias Membrane.{RTCP, RTP}
 
-  def_input_pad :input, accepted_format: _any, demand_mode: :auto
-  def_output_pad :output, accepted_format: _any, demand_mode: :auto
+  def_input_pad :input, accepted_format: _any, flow_control: :auto
+  def_output_pad :output, accepted_format: _any, flow_control: :auto
 
   def_options local_ssrc: [spec: RTP.ssrc_t()],
               remote_ssrc: [spec: RTP.ssrc_t()],
@@ -113,7 +113,7 @@ defmodule Membrane.RTCP.Receiver do
       interarrival_jitter: trunc(stats.interarrival_jitter),
       last_sr_timestamp: Map.get(state.sr_info, :cut_wallclock_ts, 0),
       # delay_since_sr is expressed in 1/65536 seconds, see https://tools.ietf.org/html/rfc3550#section-6.4.1
-      delay_since_sr: Time.round_to_seconds(65_536 * delay_since_sr)
+      delay_since_sr: Time.as_seconds(65_536 * delay_since_sr, :round)
     }
 
     packet = %RTCP.ReceiverReportPacket{ssrc: state.local_ssrc, reports: [report_block]}
@@ -149,7 +149,7 @@ defmodule Membrane.RTCP.Receiver do
   def handle_event(pad, event, ctx, state), do: super(pad, event, ctx, state)
 
   @impl true
-  def handle_process(:input, buffer, _ctx, state) do
+  def handle_buffer(:input, buffer, _ctx, state) do
     {[buffer: {:output, buffer}], state}
   end
 

lib/membrane/rtcp/transport_feedback_packet/twcc.ex

@@ -271,7 +271,7 @@ defmodule Membrane.RTCP.TransportFeedbackPacket.TWCC do
   defp scale_delta(delta) do
     # Deltas are represented as multiples of 250μs
     # https://datatracker.ietf.org/doc/html/draft-holmer-rmcat-transport-wide-cc-extensions-01#section-3.1.5
-    delta |> Time.round_to_microseconds() |> div(250)
+    delta |> Time.as_microseconds(:round) |> div(250)
   end
 
   defp delta_to_packet_status(scaled_delta) do

lib/membrane/rtp/depayloader_bin.ex

@@ -18,12 +18,10 @@ defmodule Membrane.RTP.DepayloaderBin do
               ]
 
   def_input_pad :input,
-    accepted_format: RTP,
-    demand_unit: :buffers
+    accepted_format: RTP
 
   def_output_pad :output,
-    accepted_format: _any,
-    demand_unit: :buffers
+    accepted_format: _any
 
   @impl true
   def handle_init(_ctx, opts) do

lib/membrane/rtp/header_generator.ex

@@ -15,9 +15,9 @@ defmodule Membrane.RTP.HeaderGenerator do
   @max_seq_num Bitwise.bsl(1, 16) - 1
   @max_timestamp Bitwise.bsl(1, 32) - 1
 
-  def_input_pad :input, accepted_format: RTP, demand_mode: :auto
+  def_input_pad :input, accepted_format: RTP, flow_control: :auto
 
-  def_output_pad :output, accepted_format: RTP, demand_mode: :auto
+  def_output_pad :output, accepted_format: RTP, flow_control: :auto
 
   def_options ssrc: [spec: RTP.ssrc_t()],
               payload_type: [spec: RTP.payload_type_t()],
@@ -63,13 +63,13 @@ defmodule Membrane.RTP.HeaderGenerator do
   end
 
   @impl true
-  def handle_process(:input, buffer, _ctx, state) do
+  def handle_buffer(:input, buffer, _ctx, state) do
     {rtp_metadata, metadata} = Map.pop(buffer.metadata, :rtp, %{})
 
     rtp_offset =
       buffer.pts
-      |> Ratio.mult(state.clock_rate)
-      |> Membrane.Time.round_to_seconds()
+      |> Numbers.mult(state.clock_rate)
+      |> Membrane.Time.as_seconds(:round)
 
     rtp_timestamp = rem(state.init_timestamp + rtp_offset, @max_timestamp + 1)
 

lib/membrane/rtp/inbound_packet_tracker.ex

@@ -21,8 +21,8 @@ defmodule Membrane.RTP.InboundPacketTracker do
   @max_s24_val Bitwise.bsl(1, 23) - 1
   @min_s24_val -Bitwise.bsl(1, 23)
 
-  def_input_pad :input, accepted_format: _any, demand_mode: :auto
-  def_output_pad :output, accepted_format: _any, demand_mode: :auto
+  def_input_pad :input, accepted_format: _any, flow_control: :auto
+  def_output_pad :output, accepted_format: _any, flow_control: :auto
 
   def_options clock_rate: [
                 spec: Membrane.RTP.clock_rate_t()
@@ -74,7 +74,7 @@ defmodule Membrane.RTP.InboundPacketTracker do
   end
 
   @impl true
-  def handle_process(:input, buffer, _ctx, %State{} = state) do
+  def handle_buffer(:input, buffer, _ctx, %State{} = state) do
     seq_num = buffer.metadata.rtp.sequence_number
 
     {delta, packet_index, tracker} = SequenceNumberTracker.track(state.seq_num_tracker, seq_num)
@@ -187,7 +187,7 @@ defmodule Membrane.RTP.InboundPacketTracker do
     # Algorithm from https://tools.ietf.org/html/rfc3550#appendix-A.8
     arrival_ts = Map.get(metadata, :arrival_ts, Time.vm_time())
     buffer_ts = metadata.rtp.timestamp
-    arrival = arrival_ts |> Time.as_seconds() |> Ratio.mult(clock_rate) |> Ratio.trunc()
+    arrival = arrival_ts |> Time.as_seconds() |> Numbers.mult(clock_rate) |> Ratio.trunc()
     transit = arrival - buffer_ts
 
     {jitter, transit} =

lib/membrane/rtp/jitter_buffer.ex

@@ -15,9 +15,9 @@ defmodule Membrane.RTP.JitterBuffer do
 
   @timestamp_limit Bitwise.bsl(1, 32)
 
-  def_output_pad :output, accepted_format: RTP, demand_mode: :auto
+  def_output_pad :output, accepted_format: RTP, flow_control: :auto
 
-  def_input_pad :input, accepted_format: RTP, demand_mode: :auto
+  def_input_pad :input, accepted_format: RTP, flow_control: :auto
 
   @default_latency 200 |> Time.milliseconds()
 
@@ -66,7 +66,7 @@ defmodule Membrane.RTP.JitterBuffer do
     Process.send_after(
       self(),
       :initial_latency_passed,
-      state.latency |> Time.round_to_milliseconds()
+      state.latency |> Time.as_milliseconds(:round)
     )
 
     {[], %{state | waiting?: true}}
@@ -83,7 +83,7 @@ defmodule Membrane.RTP.JitterBuffer do
   end
 
   @impl true
-  def handle_process(:input, buffer, _context, %State{store: store, waiting?: true} = state) do
+  def handle_buffer(:input, buffer, _context, %State{store: store, waiting?: true} = state) do
     state =
       case BufferStore.insert_buffer(store, buffer) do
         {:ok, result} ->
@@ -98,7 +98,7 @@ defmodule Membrane.RTP.JitterBuffer do
   end
 
   @impl true
-  def handle_process(:input, buffer, _context, %State{store: store} = state) do
+  def handle_buffer(:input, buffer, _context, %State{store: store} = state) do
     case BufferStore.insert_buffer(store, buffer) do
       {:ok, result} ->
         state = %State{state | store: result}
@@ -147,7 +147,7 @@ defmodule Membrane.RTP.JitterBuffer do
 
         buffer_ts ->
           since_insertion = Time.monotonic_time() - buffer_ts
-          send_after_time = max(0, latency - since_insertion) |> Time.round_to_milliseconds()
+          send_after_time = max(0, latency - since_insertion) |> Time.as_milliseconds(:round)
           Process.send_after(self(), :send_buffers, send_after_time)
       end
 

lib/membrane/rtp/outbound_rtx_controller.ex

@@ -13,12 +13,12 @@ defmodule Membrane.RTP.OutboundRtxController do
 
   def_input_pad :input,
     availability: :always,
-    demand_mode: :auto,
+    flow_control: :auto,
     accepted_format: _any
 
   def_output_pad :output,
     availability: :always,
-    demand_mode: :auto,
+    flow_control: :auto,
     accepted_format: _any
 
   @max_store_size 300
@@ -37,7 +37,7 @@ defmodule Membrane.RTP.OutboundRtxController do
   end
 
   @impl true
-  def handle_process(:input, buffer, _ctx, state) when byte_size(buffer.payload) > 0 do
+  def handle_buffer(:input, buffer, _ctx, state) when byte_size(buffer.payload) > 0 do
     idx = seq_num_to_index(buffer.metadata.rtp.sequence_number)
 
     state = put_in(state, [:store, idx], {nil, buffer})
@@ -46,7 +46,7 @@ defmodule Membrane.RTP.OutboundRtxController do
   end
 
   @impl true
-  def handle_process(:input, buffer, _ctx, state), do: {[forward: buffer], state}
+  def handle_buffer(:input, buffer, _ctx, state), do: {[forward: buffer], state}
 
   @impl true
   def handle_event(

lib/membrane/rtp/outbound_tracking_serializer.ex

@@ -16,21 +16,21 @@ defmodule Membrane.RTP.OutboundTrackingSerializer do
   alias Membrane.RTCP.TransportFeedbackPacket.NACK
   alias Membrane.RTP.Session.SenderReport
 
-  def_input_pad :input, accepted_format: RTP, demand_mode: :auto
+  def_input_pad :input, accepted_format: RTP, flow_control: :auto
 
   def_output_pad :output,
     accepted_format: %RemoteStream{type: :packetized, content_format: RTP},
-    demand_mode: :auto
+    flow_control: :auto
 
   def_input_pad :rtcp_input,
     availability: :on_request,
     accepted_format: _any,
-    demand_mode: :auto
+    flow_control: :auto
 
   def_output_pad :rtcp_output,
     availability: :on_request,
     accepted_format: %RemoteStream{type: :packetized, content_format: RTCP},
-    demand_mode: :auto
+    flow_control: :auto
 
   def_options ssrc: [spec: RTP.ssrc_t()],
               payload_type: [spec: RTP.payload_type_t()],
@@ -209,7 +209,7 @@ defmodule Membrane.RTP.OutboundTrackingSerializer do
   end
 
   @impl true
-  def handle_process(:input, %Buffer{} = buffer, _ctx, state) do
+  def handle_buffer(:input, %Buffer{} = buffer, _ctx, state) do
     state = update_stats(buffer, state)
 
     %{rtp: rtp_metadata} = buffer.metadata

lib/membrane/rtp/parser.ex

@@ -44,12 +44,12 @@ defmodule Membrane.RTP.Parser do
   def_input_pad :input,
     accepted_format:
       %RemoteStream{type: :packetized, content_format: cf} when cf in [nil, RTP, RTCP],
-    demand_mode: :auto
+    flow_control: :auto
 
-  def_output_pad :output, accepted_format: RTP, demand_mode: :auto
+  def_output_pad :output, accepted_format: RTP, flow_control: :auto
 
   def_output_pad :rtcp_output,
-    mode: :push,
+    flow_control: :push,
     accepted_format: %RemoteStream{content_format: RTCP, type: :packetized},
     availability: :on_request
 
@@ -86,7 +86,7 @@ defmodule Membrane.RTP.Parser do
   end
 
   @impl true
-  def handle_process(:input, %Buffer{payload: payload, metadata: metadata} = buffer, _ctx, state) do
+  def handle_buffer(:input, %Buffer{payload: payload, metadata: metadata} = buffer, _ctx, state) do
     with :rtp <- RTP.Packet.identify(payload),
          {:ok,
           %{packet: packet, padding_size: padding_size, total_header_size: total_header_size}} <-

lib/membrane/rtp/payloader_bin.ex

@@ -7,9 +7,9 @@ defmodule Membrane.RTP.PayloaderBin do
 
   alias Membrane.RTP
 
-  def_input_pad :input, accepted_format: _any, demand_unit: :buffers
+  def_input_pad :input, accepted_format: _any
 
-  def_output_pad :output, accepted_format: RTP, demand_unit: :buffers
+  def_output_pad :output, accepted_format: RTP
 
   def_options payloader: [
                 spec: module(),

lib/membrane/rtp/rtx_parser.ex

@@ -16,8 +16,8 @@ defmodule Membrane.RTP.RTXParser do
 
   alias Membrane.{Buffer, RTP}
 
-  def_input_pad :input, accepted_format: RTP, demand_mode: :auto
-  def_output_pad :output, accepted_format: RTP, demand_mode: :auto
+  def_input_pad :input, accepted_format: RTP, flow_control: :auto
+  def_output_pad :output, accepted_format: RTP, flow_control: :auto
 
   def_options original_payload_type: [
                 description:
@@ -47,7 +47,7 @@ defmodule Membrane.RTP.RTXParser do
   end
 
   @impl true
-  def handle_process(:input, %Buffer{payload: payload} = buffer, _ctx, state)
+  def handle_buffer(:input, %Buffer{payload: payload} = buffer, _ctx, state)
       when byte_size(payload) >= 2 do
     <<original_seq_num::16, original_payload::binary>> = payload
 
@@ -76,7 +76,7 @@ defmodule Membrane.RTP.RTXParser do
   end
 
   @impl true
-  def handle_process(:input, %Buffer{payload: payload, metadata: metadata}, _ctx, state) do
+  def handle_buffer(:input, %Buffer{payload: payload, metadata: metadata}, _ctx, state) do
     # Ignore empty buffers, most likely used for bandwidth estimation
     if byte_size(payload) > 0 do
       Membrane.Logger.warning(

lib/membrane/rtp/session/sender_report.ex

@@ -100,8 +100,8 @@ defmodule Membrane.RTP.Session.SenderReport do
 
     rtp_offset =
       (timestamp - sender_stats.timestamp)
-      |> Ratio.mult(sender_stats.clock_rate)
-      |> Time.round_to_seconds()
+      |> Numbers.mult(sender_stats.clock_rate)
+      |> Time.as_seconds(:round)
 
     rtp_timestamp = rem(sender_stats.rtp_timestamp + rtp_offset, @timestamp_limit)
 

lib/membrane/rtp/session_bin.ex

@@ -180,7 +180,6 @@ defmodule Membrane.RTP.SessionBin do
   end
 
   def_input_pad :input,
-    demand_unit: :buffers,
     accepted_format: _any,
     availability: :on_request,
     options: [
@@ -215,13 +214,11 @@ defmodule Membrane.RTP.SessionBin do
     ]
 
   def_input_pad :rtp_input,
-    demand_unit: :buffers,
     accepted_format:
       %RemoteStream{type: :packetized, content_format: cf} when cf in [RTP, RTCP, nil],
     availability: :on_request
 
   def_output_pad :output,
-    demand_unit: :buffers,
     accepted_format: _any,
     availability: :on_request,
     options: [
@@ -286,7 +283,6 @@ defmodule Membrane.RTP.SessionBin do
     ]
 
   def_output_pad :rtp_output,
-    demand_unit: :buffers,
     accepted_format: %RemoteStream{type: :packetized, content_format: RTP},
     availability: :on_request,
     options: [
@@ -322,12 +318,10 @@ defmodule Membrane.RTP.SessionBin do
     ]
 
   def_output_pad :rtcp_receiver_output,
-    demand_unit: :buffers,
     accepted_format: %RemoteStream{type: :packetized, content_format: RTCP},
     availability: :on_request
 
   def_output_pad :rtcp_sender_output,
-    demand_unit: :buffers,
     accepted_format: %RemoteStream{type: :packetized, content_format: RTCP},
     availability: :on_request