Native STOMP - Make in runnable after the latest refactorings, fix queue type deduction

Author: ikavgo

deps/rabbitmq_stomp/src/rabbit_stomp_processor.erl

@@ -99,9 +99,13 @@
 
 -export_type ([process_frame_result/0]).
 
+-export([adapter_name/1]).
+
 %%----------------------------------------------------------------------------
 %% Public API
 %%----------------------------------------------------------------------------
+adapter_name(#state{cfg = #cfg{adapter_info = #amqp_adapter_info{name = Name}}}) ->
+  Name.
 
 -spec initial_state(
   #stomp_configuration{},
@@ -313,7 +317,7 @@ process_connect(Implicit, Frame,
                                                   adapter_info   = AdapterInfo}}) ->
     process_request(
       fun(StateN) ->
-              maybe
+              Res1 = maybe
                   {ok, Version} = negotiate_version(Frame),
                   FT = frame_transformer(Version),
                   Frame1 = FT(Frame),
@@ -322,12 +326,12 @@ process_connect(Implicit, Frame,
                   VHost = login_header(Frame1, ?HEADER_HOST, DefaultVHost),
                   Heartbeat = login_header(Frame1, ?HEADER_HEART_BEAT, "0,0"),
                   {ProtoName, _} = AdapterInfo#amqp_adapter_info.protocol,
-                  StateN1 = StateN#state{cfg = #cfg{vhost = VHost,
-                                                    adapter_info = AdapterInfo#amqp_adapter_info{
-                                                                     protocol = {ProtoName, Version}},
-                                                    frame_transformer = FT,
-                                                    auth_mechanism = Auth,
-                                                    auth_login = Username}},
+                  StateN1 = StateN#state{cfg = Config#cfg{vhost = VHost,
+                                                          adapter_info = AdapterInfo#amqp_adapter_info{
+                                                                           protocol = {ProtoName, Version}},
+                                                          frame_transformer = FT,
+                                                          auth_mechanism = Auth,
+                                                          auth_login = Username}},
                   {Username, AuthProps} = auth_props_for_creds(Creds, StateN1),
                   {ok, User} ?= rabbit_access_control:check_user_login(Username, AuthProps),
                   {ok, AuthzCtx} ?= check_vhost_access(VHost, User, PeerIp),
@@ -347,10 +351,10 @@ process_connect(Implicit, Frame,
                                false -> [{?HEADER_SERVER, server_header()} | Headers]
                            end,
                            "",
-                           StateN1#state{cfg = #cfg{
-                                          session_id = SessionId,
-                                                  version    = Version
-                                                 },
+                           StateN1#state{cfg = StateN1#state.cfg#cfg{
+                                                               session_id = SessionId,
+                                                               version    = Version
+                                                              },
                                          user = User,
                                          authz_ctx = AuthzCtx}),
                   self() ! connection_created,
@@ -375,14 +379,15 @@ process_connect(Implicit, Frame,
                       rabbit_log:warning("STOMP login failed for user '~ts': "
                                          "this user's access is restricted to localhost", [EUsername]),
                       error("Bad CONNECT", "non-loopback access denied", State)
-              end
-              case {Res, Implicit} of
+              end,
+              case {Res1, Implicit} of
                   {{ok, _, StateN2}, implicit} ->
                       self() ! connection_created, ok(StateN2);
                   _                            ->
-                      self() ! connection_created, Res
+                      self() ! connection_created, Res1
 
-              end,
+              end
+      end,
       State).
 
 creds(_, _, #cfg{default_login       = DefLogin,
@@ -659,7 +664,7 @@ maybe_delete_durable_sub_queue({topic, Name}, Frame,
             {ok, Id} = rabbit_stomp_frame:header(Frame, ?HEADER_ID),
             QName = rabbit_stomp_util:subscription_queue_name(Name, Id, Frame),
             QRes = rabbit_misc:r(VHost, queue, list_to_binary(QName)),
-            io:format("Durable QRes: ~p~n", [QRes]),
+            ?LOG_DEBUG("Durable QRes: ~p~n", [QRes]),
             delete_queue(QRes, Username),
             ok(State);
         false ->
@@ -753,6 +758,7 @@ do_subscribe(Destination, DestHdr, Frame,
                     try
                         {ok, State1} = consume_queue(QueueName, #{no_ack => (AckMode == auto),
                                                                   prefetch_count => Prefetch,
+                                                                  mode => {simple_prefetch, Prefetch},
                                                                   consumer_tag => ConsumerTag,
                                                                   exclusive_consume => false,
                                                                   args => Arguments},
@@ -913,16 +919,16 @@ do_send(Destination, _DestHdr,
 
             Message = rabbit_message_interceptor:intercept(Message0),
 
-            io:format("Message: ~p~n", [Message]),
+            %% io:format("Message: ~p~n", [Message]),
 
             QNames = rabbit_exchange:route(Exchange, Message, #{return_binding_keys => true}),
-            io:format("QNames ~p~n", [QNames]),
+            %% io:format("QNames ~p~n", [QNames]),
 
             Delivery = {Message, DeliveryOptions, QNames},
-            io:format("Delivery: ~p~n", [Delivery]),
+            %% io:format("Delivery: ~p~n", [Delivery]),
             deliver_to_queues(ExchangeName, Delivery, State2);
         {error, _} = Err ->
-            io:format("Err ~p~n", [Err]),
+            %% io:format("Err ~p~n", [Err]),
             Err
     end.
 
@@ -941,7 +947,7 @@ deliver_to_queues(XName,
     Qs0 = rabbit_amqqueue:lookup_many(RoutedToQNames),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
     MsgSeqNo = maps:get(correlation, Options, undefined),
-    io:format("Qs: ~p~n", [Qs]),
+    %% io:format("Qs: ~p~n", [Qs]),
     case rabbit_queue_type:deliver(Qs, Message, Options, QStates0) of
         {ok, QStates, Actions} ->
             rabbit_global_counters:messages_routed(stomp, length(Qs)),
@@ -1183,10 +1189,10 @@ deliver_to_client(ConsumerTag, Ack, Msgs, State) ->
 
 deliver_one_to_client(ConsumerTag, _Ack, {QName, QPid, MsgId, Redelivered, MsgCont0} = _Msg,
                       State = #state{queue_states = QStates,
-                                          delivery_tag = DeliveryTag}) ->
+                                     delivery_tag = DeliveryTag}) ->
 
-    [RoutingKey | _] = mc:get_annotation(routing_keys, MsgCont0),
-    ExchangeNameBin = mc:get_annotation(exchange, MsgCont0),
+    [RoutingKey | _] = mc:routing_keys(MsgCont0),
+    ExchangeNameBin = mc:exchange(MsgCont0),
     MsgCont = mc:convert(mc_amqpl, MsgCont0),
     Content = mc:protocol_state(MsgCont),
     Delivery = #'basic.deliver'{consumer_tag = ConsumerTag,
@@ -1338,7 +1344,7 @@ ensure_reply_queue(TempQueueId, State = #state{reply_queues  = RQS,
 %%----------------------------------------------------------------------------
 
 ensure_receipt(Frame = #stomp_frame{command = Command}, State) ->
-    io:format("ER Frame: ~p~n", [Frame]),
+    %% io:format("ER Frame: ~p~n", [Frame]),
     case rabbit_stomp_frame:header(Frame, ?HEADER_RECEIPT) of
         {ok, Id}  -> do_receipt(Command, Id, State);
         not_found -> State
@@ -1653,7 +1659,7 @@ delete_queue(QRes, Username) ->
     case rabbit_amqqueue:with(
            QRes,
            fun (Q) ->
-                   io:format("Delete queue ~p~n", [rabbit_queue_type:delete(Q, false, false, Username)])
+                   rabbit_queue_type:delete(Q, false, false, Username)
            end,
            fun (not_found) ->
                    ok;
@@ -1693,7 +1699,7 @@ ensure_binding(QName, {Exchange, RoutingKey}, _State = #state{cfg = #cfg{
               ok ->
                   ok
           end,
-    io:format("rabbit_binding:add ~p ~p~n", [Binding, Res]),
+    %% io:format("rabbit_binding:add ~p ~p~n", [Binding, Res]),
     Res.
 
 check_resource_access(User, Resource, Perm, Context) ->
@@ -1911,18 +1917,22 @@ new_amqqueue(QNameBin0, Type, Params0, _State = #state{user = #user{username = U
                  false -> [{auto_delete, true}, {exclusive, true} | Params0];
                  true  -> Params0
              end,
-
-    amqqueue:new(QName,
-                 none,
-                 proplists:get_value(durable, Params, false),
-                 proplists:get_value(auto_delete, Params, false),
-                 case proplists:get_value(exclusive, Params, false) of
-                     false -> none;
-                     true -> self()
-                 end,
-                 proplists:get_value(arguments, Params, []),
-                 VHost,
-                 #{user => Username}).
+    Args = proplists:get_value(arguments, Params, []),
+
+    AMQ =  amqqueue:new(QName,
+                        none,
+                        proplists:get_value(durable, Params, false),
+                        proplists:get_value(auto_delete, Params, false),
+                        case proplists:get_value(exclusive, Params, false) of
+                            false -> none;
+                            true -> self()
+                        end,
+                        Args,
+                        VHost,
+                        #{user => Username},
+                        rabbit_amqqueue:get_queue_type(Args)),
+    %% io:format("~p", [AMQ]),
+    AMQ.
 
 
 to_url([])  -> [];

deps/rabbitmq_stomp/src/rabbit_stomp_reader.erl

@@ -73,7 +73,6 @@ init([SupHelperPid, Ref, Configuration]) ->
             rabbit_log_connection:info("accepting STOMP connection ~tp (~ts)",
                 [self(), ConnName]),
 
-            ParseState = rabbit_stomp_frame:initial_state(),
             ParserConfig = #stomp_parser_config{
                               max_headers        = Configuration#stomp_configuration.max_headers,
                               max_header_length  = Configuration#stomp_configuration.max_header_length,
@@ -403,19 +402,19 @@ processor_args(Configuration, Sock) ->
      ssl_login_name(RealSocket, Configuration), PeerAddr}.
 
 adapter_info(Sock) ->
-case rabbit_net:socket_ends(Socket, inbound) of
-        {ok, {PeerIp, PeerPort, Ip, Port}} ->
-#amqp_adapter_info{protocol        = {'STOMP', 0},
-                       name            = Name,
-                       host            = Host,
-                       port            = Port,
-                       peer_host       = PeerHost,
-                       peer_port       = PeerPort,
-                       additional_info = maybe_ssl_info(Sock)}
-            process_connect(ConnectPacket, Socket, ConnName, SendFun, SocketEnds);
-        {error, Reason} ->
-            {error, {socket_ends, Reason}}
-    end.
+%% case rabbit_net:socket_ends(Socket, inbound) of
+%%         {ok, {PeerIp, PeerPort, Ip, Port}} ->
+%% #amqp_adapter_info{protocol        = {'STOMP', 0},
+%%                        name            = Name,
+%%                        host            = Host,
+%%                        port            = Port,
+%%                        peer_host       = PeerHost,
+%%                        peer_port       = PeerPort,
+%%                        additional_info = maybe_ssl_info(Sock)}
+%%             process_connect(ConnectPacket, Socket, ConnName, SendFun, SocketEnds);
+%%         {error, Reason} ->
+%%             {error, {socket_ends, Reason}}
+%%     end.
 
     amqp_connection:socket_adapter_info(Sock, {'STOMP', 0}).
 

deps/rabbitmq_stomp/test/python_SUITE.erl

@@ -31,13 +31,11 @@ groups() ->
     ].
 
 init_per_suite(Config) ->
-    DataDir = ?config(data_dir, Config),
     {ok, _} = rabbit_ct_helpers:exec(["pip", "install", "-r", requirements_path(Config),
-                                                        "--target", deps_path(Config)]),
+                                                            "--target", deps_path(Config)]),
     Config.
 
 end_per_suite(Config) ->
-    DataDir = ?config(data_dir, Config),
     ok = file:del_dir_r(deps_path(Config)),
     Config.
 
@@ -82,8 +80,10 @@ run(Config, Test) ->
     StompPort = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_stomp),
     StompPortTls = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_stomp_tls),
     AmqpPort = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
+    MgmtPort = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mgmt),
     NodeName = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
     os:putenv("AMQP_PORT", integer_to_list(AmqpPort)),
+    os:putenv("MGMT_PORT", integer_to_list(MgmtPort)),
     os:putenv("STOMP_PORT", integer_to_list(StompPort)),
     os:putenv("STOMP_PORT_TLS", integer_to_list(StompPortTls)),
     os:putenv("RABBITMQ_NODENAME", atom_to_list(NodeName)),

deps/rabbitmq_stomp/test/python_SUITE_data/src/requirements.txt

@@ -1,3 +1,3 @@
 stomp.py==8.1.0
 pika==1.1.0
-
+rabbitman===0.1.0

deps/rabbitmq_stomp/test/python_SUITE_data/src/x_queue_type_quorum.py

@@ -10,7 +10,7 @@
 import time
 import os
 import re
-
+import rabbitman
 
 class TestUserGeneratedQueueName(base.BaseTest):
 
@@ -34,6 +34,11 @@ def test_quorum_queue(self):
         # let the quorum queue some time to start
         time.sleep(5)
 
+        client = rabbitman.Client(f'http://localhost:{(os.environ["MGMT_PORT"])}', 'guest', 'guest')
+        queue = client.get_queues_by_vhost_and_name("/", queueName)
+
+        self.assertEqual(queue['type'], 'quorum')
+
         connection = pika.BlockingConnection(
                 pika.ConnectionParameters(host='127.0.0.1', port=int(os.environ["AMQP_PORT"])))
         channel = connection.channel()

deps/rabbitmq_stomp/test/python_SUITE_data/src/x_queue_type_stream.py

@@ -10,7 +10,7 @@
 import time
 import os
 import re
-import urllib.request, json
+import rabbitman
 
 class TestUserGeneratedQueueName(base.BaseTest):
 
@@ -34,12 +34,17 @@ def test_stream_queue(self):
                     'id': 1234,
                     'prefetch-count': 10
                 },
-               ack="client"
+                ack="client"
                 )
 
         # let the stream queue some time to start
         time.sleep(5)
 
+        client = rabbitman.Client(f'http://localhost:{(os.environ["MGMT_PORT"])}', 'guest', 'guest')
+        queue = client.get_queues_by_vhost_and_name("/", queueName)
+
+        self.assertEqual(queue['type'], 'stream')
+
         connection = pika.BlockingConnection(
                 pika.ConnectionParameters(host='127.0.0.1', port=int(os.environ["AMQP_PORT"])))
         channel = connection.channel()

deps/rabbitmq_web_stomp/src/rabbit_web_stomp_handler.erl

@@ -177,7 +177,7 @@ websocket_info({start_heartbeats, _},
                State = #state{heartbeat_mode = no_heartbeat}) ->
     {ok, State};
 websocket_info({start_heartbeats, {0, 0}}, State) ->
-    {ok, State#state{timeout_sec = {0, 0}}};
+    {ok, State};
 websocket_info({start_heartbeats, {SendTimeout, ReceiveTimeout}},
                State = #state{socket         = Sock,
                               heartbeat_sup  = SupPid,
@@ -187,8 +187,7 @@ websocket_info({start_heartbeats, {SendTimeout, ReceiveTimeout}},
     ReceiveFun = fun() -> Self ! client_timeout end,
     Heartbeat = rabbit_heartbeat:start(SupPid, Sock, SendTimeout,
                                        SendFun, ReceiveTimeout, ReceiveFun),
-    {ok, State#state{heartbeat = Heartbeat,
-                     timeout_sec = {0, 0}}};
+    {ok, State#state{heartbeat = Heartbeat}};
 websocket_info(connection_created, State) ->
     Infos = infos(?INFO_ITEMS ++ ?OTHER_METRICS, State),
     ?LOG_DEBUG("Connection created infos ~p", [Infos]),
@@ -379,9 +378,8 @@ info_internal(garbage_collection, _State) ->
 info_internal(reductions, _State) ->
     {reductions, Reductions} = erlang:process_info(self(), reductions),
     Reductions;
-info_internal(timeout, #state{timeout_sec = {_, Receive}}) ->
-    Receive;
-info_internal(timeout, #state{timeout_sec = undefined}) ->
+info_internal(timeout, State) ->
+    %% TODO: real value
     0;
 info_internal(conn_name, #state{conn_name = Val}) ->
     rabbit_data_coercion:to_binary(Val);