DO NOT MERGE Use Cowboy's new direct data delivery for HTTP/2 Websocket

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 ->
@@ -84,7 +100,7 @@ init(Req, Opts) ->
                                    stats_timer = rabbit_event:init_stats_timer()},
                     WsOpts0 = proplists:get_value(ws_opts, Opts, #{}),
                     WsOpts  = maps:merge(#{compress => true}, WsOpts0),
-                    {?MODULE, Req1, State, WsOpts}
+                    {?MODULE, Req1, State, WsOpts#{data_delivery => relay}}
             end
     end.
 

deps/rabbitmq_web_stomp/src/rabbit_web_stomp_handler.erl

@@ -95,7 +95,7 @@ init(Req0, Opts) ->
         socket             = SockInfo,
         peername           = PeerAddr,
         auth_hd            = cowboy_req:header(<<"authorization">>, Req)
-    }, WsOpts}.
+    }, WsOpts#{data_delivery => relay}}.
 
 websocket_init(State) ->
     process_flag(trap_exit, true),

rabbitmq-components.mk

@@ -40,8 +40,8 @@ endif
 # all projects use the same versions. It avoids conflicts.
 
 dep_accept = hex 0.3.5
-dep_cowboy = hex 2.13.0
-dep_cowlib = hex 2.14.0
+dep_cowboy = git https://github.com/ninenines/cowboy direct-data_delivery-for-h2-websocket
+dep_cowlib = git https://github.com/ninenines/cowlib master
 dep_credentials_obfuscation = hex 3.5.0
 dep_cuttlefish = hex 3.5.0
 dep_gen_batch_server = hex 0.8.8