membrane_udp_plugin

defmodule Membrane.UDP.Sink do
  @moduledoc """
  Element that sends buffers received on the input pad over a UDP socket.

  The local and destination addresses are provided at init via the element's
  options; the destination can additionally be changed at runtime by returning
  a `:notify_child` action with a `t:set_destination_notification/0`:

  ```elixir
  {[notify_child: {:sink, {:set_destination, peer_ip, peer_port}}], state}
  ```
  """
  use Membrane.Sink

  alias Membrane.Buffer
  alias Membrane.UDP.{CommonSocketBehaviour, Socket}

  @type destination_port :: 1..65_535

  @type set_destination_notification ::
          {:set_destination, :inet.ip_address(), destination_port()}

  def_options destination_address: [
                spec: :inet.ip_address(),
                description: "An IP Address that the packets will be sent to."
              ],
              destination_port_no: [
                spec: destination_port(),
                description: "A UDP port number of a target."
              ],
              local_address: [
                spec: :inet.socket_address(),
                default: :any,
                description: """
                An IP Address set for a UDP socket used to sent packets. It allows to specify 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 for the socket used to sent packets. If set to `0` (default)
                the underlying OS will assign a free UDP port.
                """
              ],
              local_socket: [
                spec: :gen_udp.socket() | nil,
                default: nil,
                description: """
                Already opened UDP socket, if provided it will be used instead of creating
                and opening a new one.
                """
              ]

  def_input_pad :input, accepted_format: _any

  # Private API

  @impl true
  def handle_init(_context, %__MODULE__{} = options) do
    %__MODULE__{
      destination_address: dst_address,
      destination_port_no: dst_port_no,
      local_address: local_address,
      local_port_no: local_port_no,
      local_socket: local_socket
    } = options

    :ok = CommonSocketBehaviour.validate_destination!(dst_address, dst_port_no)

    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,
        socket_handle: local_socket
      }
    }

    {[], state}
  end

  @impl true
  def handle_playing(_context, state) do
    {[], state}
  end

  @impl true
  def handle_buffer(:input, %Buffer{payload: payload}, _context, state) do
    %{dst_socket: dst_socket, local_socket: local_socket} = state

    case Socket.send(dst_socket, local_socket, payload) do
      :ok -> {[], state}
      {:error, cause} -> raise "Error sending UDP packet, reason: #{inspect(cause)}"
    end
  end

  @impl true
  def handle_parent_notification({:set_destination, ip, port}, _ctx, state) do
    :ok = CommonSocketBehaviour.validate_destination!(ip, port)

    state =
      state
      |> put_in([:dst_socket, :ip_address], ip)
      |> put_in([:dst_socket, :port_no], port)

    {[], state}
  end

  @impl true
  defdelegate handle_setup(context, state), to: CommonSocketBehaviour
end