-
Notifications
You must be signed in to change notification settings - Fork 2
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
UDP Endpoint #28
Merged
+225
−84
Merged
UDP Endpoint #28
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit
Hold shift + click to select a range
5f6b87f
Add UDP.Endpoint
Rados13 bd4d7b6
Update to core 1.0
Rados13 e980427
Remove duplication in tests
Rados13 9293923
Update docs
Rados13 25f4fa1
Add recbuf socket option in SocketFactory
Rados13 61323fd
Remove SocketFactory
Rados13 7b96431
Fix warnings
Rados13 File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| defmodule Membrane.UDP.Endpoint do | ||
| @moduledoc """ | ||
| Element that sends buffers received on the input pad over a UDP socket and | ||
| reads packets from a UDP socket and sends their payloads through the output pad. | ||
| """ | ||
| use Membrane.Endpoint | ||
|
|
||
| import Mockery.Macro | ||
|
|
||
| alias Membrane.{Buffer, RemoteStream} | ||
| alias Membrane.UDP.{CommonSocketBehaviour, Socket} | ||
|
|
||
| def_options destination_address: [ | ||
| spec: :inet.ip_address(), | ||
| description: "An IP Address that the packets will be sent to." | ||
| ], | ||
| destination_port_no: [ | ||
| spec: :inet.port_number(), | ||
| description: "A UDP port number of a target." | ||
| ], | ||
| local_address: [ | ||
| spec: :inet.socket_address(), | ||
| default: :any, | ||
| description: """ | ||
| This address is used in two cases: | ||
| * An IP Address set for a UDP socket used to sent packets. | ||
| * An IP Address on which the socket will listen. | ||
| In both cases, it allows to choose which network interface to use if there's more than one. | ||
| """ | ||
| ], | ||
| local_port_no: [ | ||
| spec: :inet.port_number(), | ||
| default: 0, | ||
| description: """ | ||
| A UDP port number used when opening a receiving socket and for sending packets. | ||
| """ | ||
| ], | ||
| recv_buffer_size: [ | ||
| spec: pos_integer(), | ||
| default: 1024 * 1024, | ||
| description: """ | ||
| Size of the receive buffer. Packages of size greater than this buffer will be truncated | ||
| """ | ||
| ] | ||
|
|
||
| def_input_pad :input, | ||
| accepted_format: _any, | ||
| flow_control: :manual, | ||
| demand_unit: :buffers | ||
|
|
||
| def_output_pad :output, accepted_format: %RemoteStream{type: :packetized}, flow_control: :push | ||
|
|
||
| # Private API | ||
|
|
||
| @impl true | ||
| def handle_init(_context, %__MODULE__{} = opts) do | ||
| %__MODULE__{ | ||
| destination_address: dst_address, | ||
| destination_port_no: dst_port_no, | ||
| local_address: local_address, | ||
| local_port_no: local_port_no | ||
| } = opts | ||
|
|
||
| state = %{ | ||
| dst_socket: %Socket{ | ||
| ip_address: dst_address, | ||
| port_no: dst_port_no | ||
| }, | ||
| local_socket: %Socket{ | ||
| ip_address: local_address, | ||
| port_no: local_port_no, | ||
| sock_opts: [recbuf: opts.recv_buffer_size] | ||
| } | ||
| } | ||
|
|
||
| {[], state} | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_playing(_context, state) do | ||
| {[demand: :input, stream_format: {:output, %RemoteStream{type: :packetized}}], state} | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_buffer(:input, %Buffer{payload: payload}, _context, state) do | ||
| %{dst_socket: dst_socket, local_socket: local_socket} = state | ||
|
|
||
| case mockable(Socket).send(dst_socket, local_socket, payload) do | ||
| :ok -> {[demand: :input], state} | ||
| {:error, cause} -> raise "Error sending UDP packet, reason: #{inspect(cause)}" | ||
| end | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_parent_notification( | ||
| {:udp, _socket_handle, _addr, _port_no, _payload} = meta, | ||
| ctx, | ||
| state | ||
| ) do | ||
| handle_info(meta, ctx, state) | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_info( | ||
| {:udp, _socket_handle, address, port_no, payload}, | ||
| %{playback: :playing}, | ||
| state | ||
| ) do | ||
| metadata = | ||
| Map.new() | ||
| |> Map.put(:udp_source_address, address) | ||
| |> Map.put(:udp_source_port, port_no) | ||
| |> Map.put(:arrival_ts, Membrane.Time.vm_time()) | ||
|
|
||
| actions = [buffer: {:output, %Buffer{payload: payload, metadata: metadata}}] | ||
|
|
||
| {actions, state} | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_info( | ||
| {:udp, _socket_handle, _address, _port_no, _payload}, | ||
| _ctx, | ||
| state | ||
| ) do | ||
| {[], state} | ||
| end | ||
|
|
||
| @impl true | ||
| defdelegate handle_setup(context, state), to: CommonSocketBehaviour | ||
| end | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,29 +1,31 @@ | ||
| defmodule Membrane.UDP.SourceTest do | ||
| use ExUnit.Case | ||
|
|
||
| alias Membrane.UDP.Source | ||
| alias Membrane.UDP.{Endpoint, Source} | ||
|
|
||
| test "parses udp message" do | ||
| example_binary_payload = "Hi there, I am binary" | ||
| sender_port = 6666 | ||
| sender_address = {192, 168, 0, 1} | ||
| state = :unchanged | ||
| message = {:udp, 5000, sender_address, sender_port, example_binary_payload} | ||
| for module <- [Endpoint, Source] do | ||
| test "parses udp message #{inspect(module)} element" do | ||
| example_binary_payload = "Hi there, I am binary" | ||
| sender_port = 6666 | ||
| sender_address = {192, 168, 0, 1} | ||
| state = :unchanged | ||
| message = {:udp, 5000, sender_address, sender_port, example_binary_payload} | ||
|
|
||
| assert {actions, ^state} = | ||
| Source.handle_parent_notification(message, %{playback: :playing}, state) | ||
| assert {actions, ^state} = | ||
| unquote(module).handle_parent_notification(message, %{playback: :playing}, state) | ||
|
|
||
| assert {:output, buffer} = Keyword.get(actions, :buffer) | ||
| assert {:output, buffer} = Keyword.get(actions, :buffer) | ||
|
|
||
| assert %Membrane.Buffer{ | ||
| payload: ^example_binary_payload, | ||
| metadata: %{ | ||
| udp_source_address: ^sender_address, | ||
| udp_source_port: ^sender_port, | ||
| arrival_ts: arrival_ts | ||
| } | ||
| } = buffer | ||
| assert %Membrane.Buffer{ | ||
| payload: ^example_binary_payload, | ||
| metadata: %{ | ||
| udp_source_address: ^sender_address, | ||
| udp_source_port: ^sender_port, | ||
| arrival_ts: arrival_ts | ||
| } | ||
| } = buffer | ||
|
|
||
| assert_in_delta(arrival_ts, Membrane.Time.vm_time(), 2 |> Membrane.Time.milliseconds()) | ||
| assert_in_delta(arrival_ts, Membrane.Time.vm_time(), 2 |> Membrane.Time.milliseconds()) | ||
| end | ||
| end | ||
| end |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let's make it auto