channel.ex

defmodule WebsocketApi.MerchantLocationChannel do
    use WebsocketApi.Web, :channel
    alias WebsocketApi.{
        DeviceWatcher,
        MerchantLocation,
        MerchantLocationDevice,
        Repo
    }
    require Logger
    @moduledoc false

    def join("merchant_location:" <> uuid, payload, socket) do
        session_key = Map.get(payload, "session_key", "")
        device_id = Map.get(payload, "poynt_device_id", "")
        session = Repo.get_by(WebsocketApi.Session, session_key: session_key)

        if session do
            merchant_location = Repo.get_by(MerchantLocation, uuid: uuid)
            device = if String.length(String.trim(device_id)) != 0 do
                params = %{
                  merchant_location_id: merchant_location.id,
                  name: "poynt",
                  identifier: device_id,
                  is_connected: true
                }
                
                {:ok, device} = MerchantLocationDevice.insert_or_update(params, %{identifier: device_id})
                
                Logger.warn("Monitoring poynt device #{device_id} for merchant: #{uuid}")
                :ok = DeviceWatcher.monitor(:merchant_location, self(),
                    {__MODULE__, :device_disconnected, [uuid, device_id]}
                )
                
                WebsocketApi.Endpoint.broadcast(
                  "merchant_location:#{uuid}",
                  "publish",
                  %{device_connected: device_id}
                )
                device
            else
                Repo.get_by(
                    MerchantLocationDevice,
                    %{is_connected: true, merchant_location_id: merchant_location.id}
                )
            end
            Logger.info("Joined appointment channel for merchant: #{uuid}")
            echo_payload = Map.merge(payload, %{device_connected: device && device.identifier})
            
            {
                :ok,
                echo_payload,
                socket
            }
        else
            {:error, "unauthorized"}
        end
    end

    def handle_in("ping", payload, socket) do
        {:reply, {:ok, payload}, socket}
    end

    def handle_out(event, payload, socket) do
        push socket, event, payload
        {:noreply, socket}
    end
    
    def device_disconnected(uuid, device_id) do
      Logger.warn("Device connection dropped - merchant: #{uuid}, device: #{device_id}")

      MerchantLocationDevice
      |> Repo.get_by(%{identifier: device_id})
      |> MerchantLocationDevice.update(%{is_connected: false})

      WebsocketApi.Endpoint.broadcast(
        "merchant_location:#{uuid}",
        "publish",
        %{device_disconnected: device_id}
      )
    end
end

channel_watcher.ex

defmodule WebsocketApi.DeviceWatcher do
  use GenServer
  @moduledoc """
  This dude watches a pid, and when it gets killed
  triggers a callback function defined by the instantiator.
  
  Good for use with monitoring websocket connections
  and performing some action when the connection is dropped
  """

  ## Client API

  def monitor(server_name, pid, mfa) do
    GenServer.call(server_name, {:monitor, pid, mfa})
  end

  def demonitor(server_name, pid) do
    GenServer.call(server_name, {:demonitor, pid})
  end

  ## Server API

  def start_link(name) do
    GenServer.start_link(__MODULE__, [], name: name)
  end

  def init(_) do
    Process.flag(:trap_exit, true)
    {:ok, %{channels: Map.new()}}
  end

  def handle_call({:monitor, pid, mfa}, _from, state) do
    Process.link(pid)
    {:reply, :ok, put_channel(state, pid, mfa)}
  end

  def handle_call({:demonitor, pid}, _from, state) do
    case Map.fetch(state.channels, pid) do
      :error       -> {:reply, :ok, state}
      {:ok,  _mfa} ->
        Process.unlink(pid)
        {:reply, :ok, drop_channel(state, pid)}
    end
  end

  def handle_info({:EXIT, pid, _reason}, state) do
    case Map.fetch(state.channels, pid) do
      :error -> {:noreply, state}
      {:ok, {mod, func, args}} ->
        Task.start_link(fn -> apply(mod, func, args) end)
        {:noreply, drop_channel(state, pid)}
    end
  end

  defp drop_channel(state, pid) do
    %{state | channels: Map.delete(state.channels, pid)}
  end

  defp put_channel(state, pid, mfa) do
    %{state | channels: Map.put(state.channels, pid, mfa)}
  end
end