Add websocket support with Gun

Expand cowboy routing so /ws[...] are redirected to a Gun Websocket handler. To facilitate this a ws macro is added to matcher that sets up 'route' by generating an ID. This ID is a query parameter appended to the redirect URI and mapped to backend host, port and path.
mu-semtech · Jul 21, 2021 · 72c4243 · 72c4243
1 parent 213c4a1
commit 72c4243
Show file tree

Hide file tree

Showing 12 changed files with 269 additions and 61 deletions.
diff --git a/config/config.exs b/config/config.exs
@@ -44,7 +44,11 @@ config :dispatcher,
   # log whenever a layer starts processing
   log_layer_start_processing: CH.system_boolean("LOG_LAYER_START_PROCESSING"),
   # log whenever a layer matched, and if no matching layer was found
-  log_layer_matching: CH.system_boolean("LOG_LAYER_MATCHING")
+  log_layer_matching: CH.system_boolean("LOG_LAYER_MATCHING"),
+  log_ws_all: CH.system_boolean("LOG_WS_ALL"),
+  log_ws_backend: CH.system_boolean("LOG_WS_BACKEND"),
+  log_ws_frontend: CH.system_boolean("LOG_WS_FRONTEND"),
+  log_ws_unhandled: CH.system_boolean("LOG_WS_UNHANDLED")
 
 # It is also possible to import configuration files, relative to this
 # directory. For example, you can emulate configuration per environment

diff --git a/lib/dispatcher.ex b/lib/dispatcher.ex
@@ -1,11 +1,11 @@
 defmodule Dispatcher do
   use Matcher
 
-  define_accept_types [
-    text: [ "text/*" ],
-    html: [ "text/html", "application/xhtml+html" ],
-    json: [ "application/json", "application/vnd.api+json" ]
-  ]
+  define_accept_types(
+    text: ["text/*"],
+    html: ["text/html", "application/xhtml+html"],
+    json: ["application/json", "application/vnd.api+json"]
+  )
 
   # get "/*_rest", %{ accept: %{ html: true } } do
   #   Proxy.forward conn, [], "http://static/ember-app/index.html"
@@ -16,35 +16,39 @@ defmodule Dispatcher do
   # end
 
   post "/hello/erika", %{} do
-    Plug.Conn.send_resp conn, 401, "FORBIDDEN"
+    Plug.Conn.send_resp(conn, 401, "FORBIDDEN")
   end
 
   # 200 microservice dispatching
 
-  match "/hello/erika", %{ accept: %{ json: true } } do
-    Plug.Conn.send_resp conn, 200, "{ \"message\": \"Hello Erika\" }"
+  match "/hello/erika", %{accept: %{json: true}} do
+    Plug.Conn.send_resp(conn, 200, "{ \"message\": \"Hello Erika\" }\n")
   end
 
-  match "/hello/erika", %{ accept: %{ html: true } } do
-    Plug.Conn.send_resp conn, 200, "<html><head><title>Hello</title></head><body>Hello Erika</body></html>"
+  match "/hello/erika", %{accept: %{html: true}} do
+    Plug.Conn.send_resp(
+      conn,
+      200,
+      "<html><head><title>Hello</title></head><body>Hello Erika</body></html>"
+    )
   end
 
   # 404 routes
 
-  match "/hello/aad/*_rest", %{ accept: %{ json: true } } do
-    Plug.Conn.send_resp conn, 200, "{ \"message\": \"Hello Aad\" }"
+  match "/hello/aad/*_rest", %{accept: %{json: true}} do
+    Plug.Conn.send_resp(conn, 200, "{ \"message\": \"Hello Aad\" }")
   end
 
-  match "/*_rest", %{ accept: %{ json: true }, last_call: true } do
-    Plug.Conn.send_resp conn, 404, "{ \"errors\": [ \"message\": \"Not found\", \"status\": 404 } ] }"
-  end
+  # Websocket example route
+  # This forwards to /ws?target=<...>
+  # Then forwards websocket from /ws?target=<...> to ws://localhost:7999
 
-  match "/*_rest", %{ accept: %{ html: true }, last_call: true } do
-    Plug.Conn.send_resp conn, 404, "<html><head><title>Not found</title></head><body>No acceptable response found</body></html>"
+  match "/ws2" do
+    ws(conn, "ws://localhost:7999")
   end
 
-  match "/*_rest", %{ last_call: true } do
-    Plug.Conn.send_resp conn, 404, "No response found"
-  end
 
+  match "__", %{last_call: true} do
+    send_resp(conn, 404, "Route not found.  See config/dispatcher.ex")
+  end
 end
diff --git a/lib/dispatcher/log.ex b/lib/dispatcher/log.ex
@@ -2,6 +2,10 @@ defmodule Dispatcher.Log do
   @type log_name ::
           :log_layer_start_processing
           | :log_layer_matching
+          | :log_ws_all
+          | :log_ws_backend
+          | :log_ws_frontend
+          | :log_ws_unhandled
 
   @spec log(log_name, any()) :: any()
   def log(name, content) do

diff --git a/lib/manipulators/add_x_rewrite_url_header.ex b/lib/manipulators/add_x_rewrite_url_header.ex
@@ -2,14 +2,14 @@ defmodule Manipulators.AddXRewriteUrlHeader do
   @behaviour ProxyManipulator
 
   @impl true
-  def headers( headers, {frontend_conn, _backend_conn} = connection ) do
+  def headers(headers, {frontend_conn, _backend_conn} = connection) do
     new_headers = [{"x-rewrite-url", frontend_conn.request_path} | headers]
     {new_headers, connection}
   end
 
   @impl true
-  def chunk(_,_), do: :skip
+  def chunk(_, _), do: :skip
 
   @impl true
-  def finish(_,_), do: :skip
+  def finish(_, _), do: :skip
 end
diff --git a/lib/manipulators/remove_accept_encoding_header.ex b/lib/manipulators/remove_accept_encoding_header.ex
@@ -3,9 +3,9 @@ defmodule Manipulators.RemoveAcceptEncodingHeader do
 
   @impl true
   def headers(headers, connection) do
-    headers =
-      headers
-      |> Enum.reject( &match?( {"accept_encoding", _}, &1 ) )
+    # headers =
+    #   headers
+    #   |> Enum.reject( &match?( {"accept_encoding", _}, &1 ) )
     {headers, connection}
   end
 

diff --git a/lib/matcher.ex b/lib/matcher.ex
@@ -2,15 +2,20 @@ alias Dispatcher.Log
 
 defmodule Matcher do
   defmacro __using__(_opts) do
+    # Set this attribute _BEFORE_ any code is ran
+    Module.register_attribute(__CALLER__.module, :websocket, accumulate: true)
+
     quote do
       require Matcher
       import Matcher
+      import Plug.Router, only: [forward: 2]
       import Plug.Conn, only: [send_resp: 3]
       import Proxy, only: [forward: 3]
 
       def layers do
-        [ :service, :last_call ]
+        [:service, :last_call]
       end
+
       defoverridable layers: 0
 
       def dispatch(conn) do
@@ -28,6 +33,34 @@ defmodule Matcher do
     end
   end
 
+  defmacro ws(conn, host) do
+    # host = "ws://localhost:8000/test"
+
+    parsed =
+      URI.parse(host)
+      |> Log.inspect(:log_ws_all, label: "Creating websocket route")
+
+    id = for _ <- 1..24, into: "", do: <<Enum.random('0123456789abcdef')>>
+
+    host = parsed.host || "localhost"
+    port = parsed.port || 80
+    path = parsed.path || "/"
+
+    Module.put_attribute(__CALLER__.module, :websocket, %{
+      host: host,
+      port: port,
+      path: path,
+      id: id
+    })
+
+    # Return redirect things
+    quote do
+      unquote(conn)
+      |> Plug.Conn.resp(:found, "")
+      |> Plug.Conn.put_resp_header("location", "/ws?target=" <> unquote(id))
+    end
+  end
+
   defmacro get(path, options \\ quote(do: %{}), do: block) do
     quote do
       match_method(get, unquote(path), unquote(options), do: unquote(block))
@@ -98,7 +131,6 @@ defmodule Matcher do
   defmacro __before_compile__(_env) do
     matchers =
       Module.get_attribute(__CALLER__.module, :matchers)
-      # |> IO.inspect(label: "Discovered matchers")
       |> Enum.map(fn {call, path, options, block} ->
         make_match_method(call, path, options, block, __CALLER__)
       end)
@@ -110,7 +142,18 @@ defmodule Matcher do
         end
       end
 
-    [last_match_def | matchers]
+    socket_dict_f =
+      quote do
+        def websockets() do
+          Enum.reduce(@websocket, %{}, fn x, acc -> Map.put(acc, x.id, x) end)
+        end
+
+        def get_websocket(id) do
+          Enum.find(@websocket, fn x -> x.id == id end)
+        end
+      end
+
+    [socket_dict_f, last_match_def | matchers]
     |> Enum.reverse()
   end
 
@@ -171,24 +214,29 @@ defmodule Matcher do
 
           new_accept =
             case value do
-              [item] -> # convert item
+              # convert item
+              [item] ->
                 {:%{}, [], [{item, true}]}
+
               [_item | _rest] ->
                 raise "Multiple items in accept arrays are not supported."
+
               {:%{}, _, _} ->
                 value
             end
 
           new_list =
             list
-            |> Keyword.drop( [:accept] )
-            |> Keyword.merge( [accept: new_accept] )
+            |> Keyword.drop([:accept])
+            |> Keyword.merge(accept: new_accept)
 
           {:%{}, any, new_list}
         else
           options
         end
-      _ -> options
+
+      _ ->
+        options
     end
   end
 
@@ -223,8 +271,6 @@ defmodule Matcher do
             str -> str
           end).()
 
-    # |> IO.inspect(label: "call name")
-
     # Creates the variable(s) for the parsed path
     process_derived_path_elements = fn elements ->
       reversed_elements = Enum.reverse(elements)
@@ -310,7 +356,6 @@ defmodule Matcher do
   def dispatch_call(conn, accept_types, layers_fn, call_handler) do
     # Extract core info
     {method, path, accept_header, host} = extract_core_info_from_conn(conn)
-    # |> IO.inspect(label: "extracted header")
 
     # Extract core request info
     accept_hashes =
@@ -321,10 +366,11 @@ defmodule Matcher do
 
     # layers |> IO.inspect(label: "layers" )
     # Try to find a solution in each of the layers
-    layers = layers_fn.()
-    |> Log.inspect(:log_available_layers, "Available layers")
+    layers =
+      layers_fn.()
+      |> Log.inspect(:log_available_layers, "Available layers")
 
-    reverse_host = Enum.reverse( host )
+    reverse_host = Enum.reverse(host)
 
     response_conn =
       layers
@@ -432,7 +478,7 @@ defmodule Matcher do
   defp sort_and_group_accept_headers(accept) do
     accept
     |> safe_parse_accept_header()
-    # |> IO.inspect(label: "parsed_accept_header")
+    |> IO.inspect(label: "parsed_accept_header")
     |> Enum.sort_by(&elem(&1, 3))
     |> Enum.group_by(&elem(&1, 3))
     |> Map.to_list()

diff --git a/lib/mu_dispatcher.ex b/lib/mu_dispatcher.ex
@@ -10,11 +10,23 @@ defmodule MuDispatcher do
     port = 80
 
     children = [
-      {Plug.Cowboy, scheme: :http, plug: PlugRouterDispatcher, options: [port: port]}
+      # this is kinda strange, but the 'plug:' field is not used when 'dispatch:' is provided (my understanding)
+      {Plug.Adapters.Cowboy,
+       scheme: :http, plug: PlugRouterDispatcher, options: [dispatch: dispatch, port: port]}
     ]
 
     Logger.info("Mu Dispatcher starting on port #{port}")
 
     Supervisor.start_link(children, strategy: :one_for_one)
   end
+
+  defp dispatch do
+    [
+      {:_,
+       [
+         {"/ws/[...]", WebsocketHandler, %{}},
+         {:_, Plug.Cowboy.Handler, {PlugRouterDispatcher, []}}
+       ]}
+    ]
+  end
 end
diff --git a/lib/plug_router_dispatcher.ex b/lib/plug_router_dispatcher.ex
@@ -1,3 +1,5 @@
+alias Dispatcher.Log
+
 defmodule PlugRouterDispatcher do
   use Plug.Router
 
@@ -6,6 +8,6 @@ defmodule PlugRouterDispatcher do
   plug(:dispatch)
 
   match _ do
-    Dispatcher.dispatch( conn )
+    Dispatcher.dispatch(conn)
   end
 end
diff --git a/lib/proxy.ex b/lib/proxy.ex
@@ -1,5 +1,8 @@
 defmodule Proxy do
-  @request_manipulators [Manipulators.AddXRewriteUrlHeader,Manipulators.RemoveAcceptEncodingHeader]
+  @request_manipulators [
+    Manipulators.AddXRewriteUrlHeader,
+    Manipulators.RemoveAcceptEncodingHeader
+  ]
   @response_manipulators [Manipulators.AddVaryHeader]
   @manipulators ProxyManipulatorSettings.make_settings(
                   @request_manipulators,
@@ -13,6 +16,7 @@ defmodule Proxy do
       conn,
       path,
       base,
-      @manipulators)
+      @manipulators
+    )
   end
 end