diff --git a/lib/live_beats/application.ex b/lib/live_beats/application.ex index 4d52e9a..d971645 100644 --- a/lib/live_beats/application.ex +++ b/lib/live_beats/application.ex @@ -29,7 +29,9 @@ defmodule LiveBeats.Application do name: PresenceClient}, # Start the Endpoint (http/https) LiveBeatsWeb.Endpoint, - {LiveBeats.SongsCleaner, count: 7, interval: :day} + {LiveBeats.SongsCleaner, count: 7, interval: :day}, + LiveBeats.UserTracker + # Start a worker by calling: LiveBeats.Worker.start_link(arg) # {LiveBeats.Worker, arg} diff --git a/lib/live_beats/presence/presence_client.ex b/lib/live_beats/presence/presence_client.ex index e72c2a2..340fbfd 100644 --- a/lib/live_beats/presence/presence_client.ex +++ b/lib/live_beats/presence/presence_client.ex @@ -41,12 +41,14 @@ defmodule LiveBeats.PresenceClient do @impl Phoenix.Presence.Client def handle_join(topic, _key, presence, state) do + LiveBeats.UserTracker.presence_joined(presence) local_broadcast(topic, {__MODULE__, %{user_joined: presence}}) {:ok, state} end @impl Phoenix.Presence.Client def handle_leave(topic, _key, presence, state) do + LiveBeats.UserTracker.presence_left(presence) local_broadcast(topic, {__MODULE__, %{user_left: presence}}) {:ok, state} end diff --git a/lib/live_beats/user_tracker.ex b/lib/live_beats/user_tracker.ex new file mode 100644 index 0000000..3a2151c --- /dev/null +++ b/lib/live_beats/user_tracker.ex @@ -0,0 +1,109 @@ +defmodule LiveBeats.UserTracker do + @moduledoc """ + Send active users updates using a polling interval. + """ + + use GenServer + @pubsub LiveBeats.PubSub + @poll_interval :timer.seconds(30) + + def subscribe() do + Phoenix.PubSub.subscribe(@pubsub, topic()) + end + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + def list_active_users() do + GenServer.call(__MODULE__, :list_users) + end + + def presence_joined(presence) do + GenServer.call(__MODULE__, {:presence_joined, presence}) + end + + def presence_left(presence) do + GenServer.call(__MODULE__, {:presence_left, presence}) + end + + @impl true + def init(_opts) do + {:ok, + schedule_updates(%{ + active_users: %{}, + user_leaves: [], + user_joins: [] + })} + end + + @impl true + def handle_call(:list_users, _from, state) do + {:reply, list_users(state), state} + end + + @impl true + def handle_call({:presence_joined, presence}, _from, state) do + {:reply, :ok, handle_join(state, presence)} + end + + @impl true + def handle_call({:presence_left, presence}, _from, state) do + {:reply, :ok, handle_leave(state, presence)} + end + + @impl true + def handle_info(:send_updates, state) do + leaves = state.user_leaves -- state.user_joins + joins = state.user_joins -- state.user_leaves + + broadcast_updates(leaves, joins) + + # cleaning joins and leaves for each interval + new_state = %{state | user_leaves: [], user_joins: []} + {:noreply, schedule_updates(new_state)} + end + + defp schedule_updates(state) do + Process.send_after(self(), :send_updates, @poll_interval) + state + end + + defp handle_join(state, %{user: user}) do + if Map.has_key?(state.active_users, user.id) do + state + else + updated_active_users = Map.put_new(state.active_users, user.id, user) + updated_user_joins = [user | state.user_joins] + + %{state | active_users: updated_active_users, user_joins: updated_user_joins} + end + end + + defp handle_leave(state, %{user: user, metas: metas}) do + if Map.has_key?(state.active_users, user.id) and metas == [] do + updated_active_users = Map.delete(state.active_users, user.id) + updated_user_leaves = [user | state.user_leaves] + + %{state | active_users: updated_active_users, user_leaves: updated_user_leaves} + else + state + end + end + + defp topic() do + "active_users" + end + + defp broadcast_updates(leaves, joins) do + Phoenix.PubSub.local_broadcast( + @pubsub, + topic(), + {LiveBeats.UserTracker, %{user_leaves: leaves, user_joins: joins}} + ) + end + + defp list_users(state) do + Enum.map(state.active_users, fn {_key, value} -> value end) + end +end diff --git a/lib/live_beats_web/live/nav.ex b/lib/live_beats_web/live/nav.ex index af1da2b..82997b9 100644 --- a/lib/live_beats_web/live/nav.ex +++ b/lib/live_beats_web/live/nav.ex @@ -1,16 +1,22 @@ defmodule LiveBeatsWeb.Nav do import Phoenix.LiveView - - alias LiveBeats.MediaLibrary + alias LiveBeats.UserTracker alias LiveBeatsWeb.{ProfileLive, SettingsLive} def on_mount(:default, _params, _session, socket) do - {:cont, - socket - |> assign(active_users: MediaLibrary.list_active_profiles(limit: 20)) - |> assign(:region, System.get_env("FLY_REGION")) - |> attach_hook(:active_tab, :handle_params, &handle_active_tab_params/3) - |> attach_hook(:ping, :handle_event, &handle_event/3)} + if connected?(socket) do + UserTracker.subscribe() + end + + socket = + socket + |> assign(:active_users, UserTracker.list_active_users()) + |> assign(:region, System.get_env("FLY_REGION")) + |> attach_hook(:active_tab, :handle_params, &handle_active_tab_params/3) + |> attach_hook(:ping, :handle_event, &handle_event/3) + |> attach_hook(:active_users, :handle_info, &handle_info/2) + + {:cont, socket} end defp handle_active_tab_params(params, _url, socket) do @@ -37,6 +43,19 @@ defmodule LiveBeatsWeb.Nav do defp handle_event(_, _, socket), do: {:cont, socket} + defp handle_info({UserTracker, %{user_leaves: leaves, user_joins: joins}}, socket) do + updated_socket = + Enum.reduce(leaves, socket, fn user, socket -> + socket + |> push_event("remove-el", %{id: "mobile-active-users-container-#{user.id}"}) + |> push_event("remove-el", %{id: "desktop-active-users-container-#{user.id}"}) + end) + + {:halt, update(updated_socket, :active_users, &(joins ++ &1))} + end + + defp handle_info(_params, socket), do: {:cont, socket} + defp current_user_profile_username(socket) do if user = socket.assigns.current_user do user.username diff --git a/lib/live_beats_web/live/profile_live.ex b/lib/live_beats_web/live/profile_live.ex index f56a290..9f9f1bc 100644 --- a/lib/live_beats_web/live/profile_live.ex +++ b/lib/live_beats_web/live/profile_live.ex @@ -107,7 +107,7 @@ defmodule LiveBeatsWeb.ProfileLive do |> list_songs() |> assign_presences() - {:ok, socket, temporary_assigns: [songs: [], presences: []]} + {:ok, socket, temporary_assigns: [songs: [], presences: [], active_users: []]} end def handle_params(params, _url, socket) do diff --git a/lib/live_beats_web/live/settings_live.ex b/lib/live_beats_web/live/settings_live.ex index d18c529..0591bba 100644 --- a/lib/live_beats_web/live/settings_live.ex +++ b/lib/live_beats_web/live/settings_live.ex @@ -70,7 +70,7 @@ defmodule LiveBeatsWeb.SettingsLive do def mount(_parmas, _session, socket) do changeset = Accounts.change_settings(socket.assigns.current_user, %{}) - {:ok, assign(socket, changeset: changeset)} + {:ok, assign(socket, changeset: changeset), temporary_assigns: [active_users: []]} end def handle_event("validate", %{"user" => params}, socket) do diff --git a/lib/live_beats_web/views/layout_view.ex b/lib/live_beats_web/views/layout_view.ex index 69fc6f8..f44d807 100644 --- a/lib/live_beats_web/views/layout_view.ex +++ b/lib/live_beats_web/views/layout_view.ex @@ -13,10 +13,17 @@ defmodule LiveBeatsWeb.LayoutView do