HTTP/2 Web-MQTT

Author: lhoguin

deps/rabbit_common/src/rabbit_net.erl

@@ -117,7 +117,10 @@ getstat(Sock, Stats) when is_port(Sock) ->
 getstat({rabbit_proxy_socket, Sock, _}, Stats) when ?IS_SSL(Sock) ->
     ssl:getstat(Sock, Stats);
 getstat({rabbit_proxy_socket, Sock, _}, Stats) when is_port(Sock) ->
-    inet:getstat(Sock, Stats).
+    inet:getstat(Sock, Stats);
+%% Virtual sockets; we do not have access to the underlying socket.
+getstat({rabbit_virtual_socket, _}, _) ->
+    {error, enotsup}.
 
 recv(Sock) when ?IS_SSL(Sock) ->
     recv(Sock, {ssl, ssl_closed, ssl_error});
@@ -180,7 +183,9 @@ peername(Sock)   when ?IS_SSL(Sock) -> ssl:peername(Sock);
 peername(Sock)   when is_port(Sock) -> inet:peername(Sock).
 
 peercert(Sock)   when ?IS_SSL(Sock) -> ssl:peercert(Sock);
-peercert(Sock)   when is_port(Sock) -> nossl.
+peercert(Sock)   when is_port(Sock) -> nossl;
+peercert({rabbit_virtual_socket, #{cert := undefined}}) -> nossl;
+peercert({rabbit_virtual_socket, #{cert := Cert}}) -> Cert.
 
 connection_string(Sock, Direction) ->
     case socket_ends(Sock, Direction) of
@@ -219,7 +224,15 @@ socket_ends({rabbit_proxy_socket, Sock, ProxyInfo}, Direction) ->
          } ->
             {ok, {rdns(FromAddress), FromPort,
                   rdns(ToAddress),   ToPort}}
-    end.
+    end;
+socket_ends({rabbit_virtual_socket, SocketInfo}, _) ->
+    #{
+        src_address := FromAddress,
+        src_port := FromPort,
+        dest_address := ToAddress,
+        dest_port := ToPort
+    } = SocketInfo,
+    {ok, {rdns(FromAddress), FromPort, rdns(ToAddress), ToPort}}.
 
 maybe_ntoab(Addr) when is_tuple(Addr) -> rabbit_misc:ntoab(Addr);
 maybe_ntoab(Host)                     -> Host.

deps/rabbitmq_mqtt/src/rabbit_mqtt_keepalive.erl

@@ -41,6 +41,14 @@ handle({init, IntervalSecs, Sock}, _State) ->
                         socket = Sock,
                         recv_oct = RecvOct,
                         received = true}};
+        %% Virtual sockets (such as Websocket/WebTransport)
+        %% do not support getstat.
+        {error, enotsup} ->
+            {ok, #state{interval_secs = IntervalSecs,
+                        timer = undefined,
+                        socket = Sock,
+                        recv_oct = 0,
+                        received = true}};
         {error, _} = Err ->
             Err
     end;

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl

@@ -135,7 +135,7 @@ start_tls_listener(TLSConf0, CowboyOpts) ->
       num_acceptors => get_env(num_ssl_acceptors, 10),
       num_conns_sups => get_env(num_conns_sup, 1)
      },
-    case cowboy:start_tls(RanchRef, RanchTransportOpts, CowboyOpts) of
+    case cowboy:start_tls(RanchRef, RanchTransportOpts, CowboyOpts#{enable_connect_protocol => true}) of
         {ok, _} ->
             ok;
         {error, {already_started, _}} ->

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_handler.erl

@@ -56,9 +56,25 @@
 upgrade(Req, Env, Handler, HandlerState) ->
     upgrade(Req, Env, Handler, HandlerState, #{}).
 
+%% We simulate PROXY headers when HTTP/2 is used to have src/dest.
+upgrade(Req=#{version := 'HTTP/2', peer := Peer, sock := Sock, cert := Cert},
+        Env, Handler, HandlerState, Opts) ->
+    logger:error("~p ~p ~p ~p ~p", [Req, Env, Handler, HandlerState, Opts]),
+    {SrcAddr, SrcPort} = Peer,
+    {DestAddr, DestPort} = Sock,
+    SocketInfo = #{
+        src_address => SrcAddr,
+        src_port => SrcPort,
+        dest_address => DestAddr,
+        dest_port => DestPort,
+        cert => Cert
+    },
+    VirtualSocket = {rabbit_virtual_socket, SocketInfo},
+    cowboy_websocket:upgrade(Req, Env, Handler, HandlerState#state{socket = VirtualSocket}, Opts);
 upgrade(Req, Env, Handler, HandlerState, Opts) ->
     cowboy_websocket:upgrade(Req, Env, Handler, HandlerState, Opts).
 
+%% @todo This is only called for HTTP/1.1.
 takeover(Parent, Ref, Socket, Transport, Opts, Buffer, {Handler, HandlerState}) ->
     Sock = case HandlerState#state.socket of
                undefined ->