chore: use built-in phoenix websocket (#1011)

Author: jyeshe

config/dev.exs

@@ -7,14 +7,12 @@ config :ae_mdw, AeMdwWeb.Endpoint,
   watchers: [],
   check_origin: false
 
-config :ae_mdw, AeMdwWeb.Endpoint,
-  live_reload: [
-    patterns: [
-      ~r"priv/gettext/.*(po)$",
-      ~r"lib/test_web/(live|views)/.*(ex)$",
-      ~r"lib/test_web/templates/.*(eex)$"
-    ]
-  ]
+config :ae_mdw, AeMdwWeb.WebsocketEndpoint,
+  http: [port: 4001],
+  debug_errors: true,
+  code_reloader: false,
+  watchers: [],
+  check_origin: false
 
 # ## SSL Support
 #

config/prod.secret.exs

@@ -18,6 +18,13 @@ config :ae_mdw, AeMdwWeb.Endpoint,
   ],
   secret_key_base: secret_key_base
 
+config :ae_mdw, AeMdwWeb.WebsocketEndpoint,
+  http: [
+    port: String.to_integer(System.get_env("WS_PORT") || "4001"),
+    transport_options: [socket_opts: [:inet6]]
+  ],
+  secret_key_base: secret_key_base
+
 # ## Using releases (Elixir v1.9+)
 #
 # If you are doing OTP releases, you need to instruct Phoenix

config/test.exs

@@ -13,5 +13,9 @@ config :ae_mdw, AeMdwWeb.Endpoint,
   http: [port: 4002],
   server: false
 
+config :ae_mdw, AeMdwWeb.WebsocketEndpoint,
+  http: [port: 4003],
+  server: true
+
 # Log warnings
 config :logger, level: :warn

lib/ae_mdw_web/supervisor.ex

@@ -7,7 +7,10 @@ defmodule AeMdwWeb.Supervisor do
 
   @impl true
   def init([]) do
-    children = [AeMdwWeb.Endpoint]
+    children = [
+      AeMdwWeb.Endpoint,
+      AeMdwWeb.WebsocketEndpoint
+    ]
 
     Supervisor.init(children, strategy: :one_for_one)
   end

lib/ae_mdw_web/websocket/broadcaster.ex

@@ -172,10 +172,7 @@ defmodule AeMdwWeb.Websocket.Broadcaster do
   end
 
   defp broadcast(channel, msg) do
-    Riverside.LocalDelivery.deliver(
-      {:channel, channel},
-      {:text, msg}
-    )
+    AeMdwWeb.Websocket.SocketHandler.channel_broadcast(channel, msg)
   end
 
   defp encode_message(payload, "Transactions", source),
@@ -258,7 +255,7 @@ defmodule AeMdwWeb.Websocket.Broadcaster do
     |> Enum.uniq()
   end
 
-  defp already_processed?(type_hash), do: EtsCache.get(@hashes_table, type_hash) != nil
+  defp already_processed?(type_hash), do: EtsCache.member(@hashes_table, type_hash)
 
   defp push_hash(type_hash), do: EtsCache.put(@hashes_table, type_hash, :ok)
 end

lib/ae_mdw_web/websocket/chain_listener.ex

@@ -61,32 +61,32 @@ defmodule AeMdwWeb.Websocket.ChainListener do
 
   @impl GenServer
   def handle_info({:DOWN, _ref, _type, pid, _info}, state) do
-    case :ets.member(@subs_main, pid) do
-      true ->
-        # main socket channel dies, all connections are disconnected. Clean all tables.
-        Enum.each(
-          [@subs_main, @subs_channel_targets, @subs_target_channels, @subs_pids],
-          &:ets.delete_all_objects/1
-        )
-
-      false ->
-        spec =
-          Ex2ms.fun do
-            {{^pid, sub}, _} -> sub
-          end
-
-        for sub <- :ets.select(@subs_channel_targets, spec) do
-          key_to_delete = {sub, pid}
-          :ets.delete(@subs_target_channels, key_to_delete)
+    Process.demonitor(pid, [:flush])
+
+    # last connection down
+    if :ets.info(@subs_pids, :size) <= 1 do
+      Enum.each(
+        [@subs_main, @subs_channel_targets, @subs_target_channels, @subs_pids],
+        &:ets.delete_all_objects/1
+      )
+    else
+      spec =
+        Ex2ms.fun do
+          {{^pid, sub}, _} -> sub
         end
 
-        spec_ =
-          Ex2ms.fun do
-            {{^pid, _}, _} -> true
-          end
+      for sub <- :ets.select(@subs_channel_targets, spec) do
+        key_to_delete = {sub, pid}
+        :ets.delete(@subs_target_channels, key_to_delete)
+      end
 
-        :ets.select_delete(@subs_channel_targets, spec_)
-        :ets.delete(@subs_pids, pid)
+      spec_ =
+        Ex2ms.fun do
+          {{^pid, _}, _} -> true
+        end
+
+      :ets.select_delete(@subs_channel_targets, spec_)
+      :ets.delete(@subs_pids, pid)
     end
 
     {:noreply, state}