STOMP: move everything to binaries

Author: ikavgo

deps/rabbitmq_stomp/include/rabbit_stomp_headers.hrl

@@ -5,57 +5,57 @@
 %% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 %%
 
--define(HEADER_ACCEPT_VERSION, "accept-version").
--define(HEADER_ACK, "ack").
--define(HEADER_AMQP_MESSAGE_ID, "amqp-message-id").
--define(HEADER_APP_ID, "app-id").
--define(HEADER_AUTO_DELETE, "auto-delete").
--define(HEADER_CONTENT_ENCODING, "content-encoding").
--define(HEADER_CONTENT_LENGTH, "content-length").
--define(HEADER_CONTENT_TYPE, "content-type").
--define(HEADER_CORRELATION_ID, "correlation-id").
--define(HEADER_DESTINATION, "destination").
--define(HEADER_DURABLE, "durable").
--define(HEADER_EXPIRATION, "expiration").
--define(HEADER_EXCLUSIVE, "exclusive").
--define(HEADER_HEART_BEAT, "heart-beat").
--define(HEADER_HOST, "host").
--define(HEADER_ID, "id").
--define(HEADER_LOGIN, "login").
--define(HEADER_MESSAGE_ID, "message-id").
--define(HEADER_PASSCODE, "passcode").
--define(HEADER_PERSISTENT, "persistent").
--define(HEADER_PREFETCH_COUNT, "prefetch-count").
--define(HEADER_X_STREAM_OFFSET, "x-stream-offset").
--define(HEADER_X_STREAM_FILTER, "x-stream-filter").
--define(HEADER_X_STREAM_MATCH_UNFILTERED, "x-stream-match-unfiltered").
--define(HEADER_PRIORITY, "priority").
--define(HEADER_X_PRIORITY, "x-priority").
--define(HEADER_RECEIPT, "receipt").
--define(HEADER_REDELIVERED, "redelivered").
--define(HEADER_REPLY_TO, "reply-to").
--define(HEADER_SERVER, "server").
--define(HEADER_SESSION, "session").
--define(HEADER_SUBSCRIPTION, "subscription").
--define(HEADER_TIMESTAMP, "timestamp").
--define(HEADER_TRANSACTION, "transaction").
--define(HEADER_TYPE, "type").
--define(HEADER_USER_ID, "user-id").
--define(HEADER_VERSION, "version").
--define(HEADER_X_DEAD_LETTER_EXCHANGE, "x-dead-letter-exchange").
--define(HEADER_X_DEAD_LETTER_ROUTING_KEY, "x-dead-letter-routing-key").
--define(HEADER_X_EXPIRES, "x-expires").
--define(HEADER_X_MAX_LENGTH, "x-max-length").
--define(HEADER_X_MAX_AGE, "x-max-age").
--define(HEADER_X_MAX_LENGTH_BYTES, "x-max-length-bytes").
--define(HEADER_X_STREAM_MAX_SEGMENT_SIZE_BYTES, "x-stream-max-segment-size-bytes").
--define(HEADER_X_MAX_PRIORITY, "x-max-priority").
--define(HEADER_X_MESSAGE_TTL, "x-message-ttl").
--define(HEADER_X_QUEUE_NAME, "x-queue-name").
--define(HEADER_X_QUEUE_TYPE, "x-queue-type").
--define(HEADER_X_STREAM_FILTER_SIZE_BYTES, "x-stream-filter-size-bytes").
+-define(HEADER_ACCEPT_VERSION, <<"accept-version">>).
+-define(HEADER_ACK, <<"ack">>).
+-define(HEADER_AMQP_MESSAGE_ID, <<"amqp-message-id">>).
+-define(HEADER_APP_ID, <<"app-id">>).
+-define(HEADER_AUTO_DELETE, <<"auto-delete">>).
+-define(HEADER_CONTENT_ENCODING, <<"content-encoding">>).
+-define(HEADER_CONTENT_LENGTH, <<"content-length">>).
+-define(HEADER_CONTENT_TYPE, <<"content-type">>).
+-define(HEADER_CORRELATION_ID, <<"correlation-id">>).
+-define(HEADER_DESTINATION, <<"destination">>).
+-define(HEADER_DURABLE, <<"durable">>).
+-define(HEADER_EXPIRATION, <<"expiration">>).
+-define(HEADER_EXCLUSIVE, <<"exclusive">>).
+-define(HEADER_HEART_BEAT, <<"heart-beat">>).
+-define(HEADER_HOST, <<"host">>).
+-define(HEADER_ID, <<"id">>).
+-define(HEADER_LOGIN, <<"login">>).
+-define(HEADER_MESSAGE_ID, <<"message-id">>).
+-define(HEADER_PASSCODE, <<"passcode">>).
+-define(HEADER_PERSISTENT, <<"persistent">>).
+-define(HEADER_PREFETCH_COUNT, <<"prefetch-count">>).
+-define(HEADER_X_STREAM_OFFSET, <<"x-stream-offset">>).
+-define(HEADER_X_STREAM_FILTER, <<"x-stream-filter">>).
+-define(HEADER_X_STREAM_MATCH_UNFILTERED, <<"x-stream-match-unfiltered">>).
+-define(HEADER_PRIORITY, <<"priority">>).
+-define(HEADER_X_PRIORITY, <<"x-priority">>).
+-define(HEADER_RECEIPT, <<"receipt">>).
+-define(HEADER_REDELIVERED, <<"redelivered">>).
+-define(HEADER_REPLY_TO, <<"reply-to">>).
+-define(HEADER_SERVER, <<"server">>).
+-define(HEADER_SESSION, <<"session">>).
+-define(HEADER_SUBSCRIPTION, <<"subscription">>).
+-define(HEADER_TIMESTAMP, <<"timestamp">>).
+-define(HEADER_TRANSACTION, <<"transaction">>).
+-define(HEADER_TYPE, <<"type">>).
+-define(HEADER_USER_ID, <<"user-id">>).
+-define(HEADER_VERSION, <<"version">>).
+-define(HEADER_X_DEAD_LETTER_EXCHANGE, <<"x-dead-letter-exchange">>).
+-define(HEADER_X_DEAD_LETTER_ROUTING_KEY, <<"x-dead-letter-routing-key">>).
+-define(HEADER_X_EXPIRES, <<"x-expires">>).
+-define(HEADER_X_MAX_LENGTH, <<"x-max-length">>).
+-define(HEADER_X_MAX_AGE, <<"x-max-age">>).
+-define(HEADER_X_MAX_LENGTH_BYTES, <<"x-max-length-bytes">>).
+-define(HEADER_X_STREAM_MAX_SEGMENT_SIZE_BYTES, <<"x-stream-max-segment-size-bytes">>).
+-define(HEADER_X_MAX_PRIORITY, <<"x-max-priority">>).
+-define(HEADER_X_MESSAGE_TTL, <<"x-message-ttl">>).
+-define(HEADER_X_QUEUE_NAME, <<"x-queue-name">>).
+-define(HEADER_X_QUEUE_TYPE, <<"x-queue-type">>).
+-define(HEADER_X_STREAM_FILTER_SIZE_BYTES, <<"x-stream-filter-size-bytes">>).
 
--define(MESSAGE_ID_SEPARATOR, "@@").
+-define(MESSAGE_ID_SEPARATOR, <<"@@">>).
 
 -define(HEADERS_NOT_ON_SEND, [?HEADER_MESSAGE_ID]).
 
@@ -83,14 +83,14 @@
                        ]).
 
 
--define(QUEUE_PREFIX, "/queue").
--define(TOPIC_PREFIX, "/topic").
--define(EXCHANGE_PREFIX, "/exchange").
--define(AMQQUEUE_PREFIX, "/amq/queue").
--define(TEMP_QUEUE_PREFIX, "/temp-queue").
-%% reply queues names can have slashes in the content so no further
+-define(QUEUE_PREFIX, <<"/queue">>).
+-define(TOPIC_PREFIX, <<"/topic">>).
+-define(EXCHANGE_PREFIX, <<"/exchange">>).
+-define(AMQQUEUE_PREFIX, <<"/amq/queue">>).
+-define(TEMP_QUEUE_PREFIX, <<"/temp-queue">>).
+%% Reply queues names can have slashes in the content so no further
 %% parsing happens.
--define(REPLY_QUEUE_PREFIX, "/reply-queue/").
+-define(REPLY_QUEUE_PREFIX, <<"/reply-queue/">>).
 
 %%-------------------------------------------------
 

deps/rabbitmq_stomp/src/rabbit_stomp_frame.erl

@@ -96,8 +96,8 @@ initial_state(Config) -> {none, Config}.
 -record(ps, {acc     = [] :: [byte()],
              acc_len = 0  :: non_neg_integer(),
              cmd          :: atom() | binary() | undefined,
-             hdrs    = [] :: [{string(), string()}],
-             hdrname      :: string() | undefined,
+             hdrs    = [] :: [{binary(), binary()}],
+             hdrname      :: binary() | undefined,
              config       :: #stomp_parser_config{}}).
 
 %%
@@ -173,15 +173,13 @@ parse_hdr(Bin, S = #ps{config = #stomp_parser_config{max_header_length = MaxHL}}
             case byte_size(Name) > MaxHL orelse byte_size(Value) > MaxHL of
                 true  -> {error, {max_header_length, MaxHL}};
                 false ->
-                    Hdrs = insert_header(S#ps.hdrs,
-                                         binary_to_list(Name),
-                                         binary_to_list(Value)),
+                    Hdrs = insert_header(S#ps.hdrs, Name, Value),
                     parse_headers(Rest, S#ps{hdrs = Hdrs})
             end;
         has_escapes ->
             parse_hdrname_esc(Bin, S#ps{acc = [], acc_len = 0});
         {no_value, Name} ->
-            {error, {header_no_value, binary_to_list(Name)}};
+            {error, {header_no_value, Name}};
         {more, Len} ->
             case Len > MaxHL of
                 true  -> {error, {max_header_length, MaxHL}};
@@ -197,14 +195,14 @@ parse_hdrname_esc(<<>>, S) ->
 parse_hdrname_esc(<<?CR>>, S) ->
     more(fun(Rest) -> parse_hdrname_esc(<<?CR, Rest/binary>>, S) end);
 parse_hdrname_esc(<<?CR, ?LF, _/binary>>, #ps{acc = Acc}) ->
-    {error, {header_no_value, lists:reverse(Acc)}};
+    {error, {header_no_value, list_to_binary(lists:reverse(Acc))}};
 parse_hdrname_esc(<<?CR, Ch:8, _/binary>>, _) ->
     {error, {unexpected_chars_in_header, [?CR, Ch]}};
 parse_hdrname_esc(<<?LF, _/binary>>, #ps{acc = Acc}) ->
-    {error, {header_no_value, lists:reverse(Acc)}};
+    {error, {header_no_value, list_to_binary(lists:reverse(Acc))}};
 parse_hdrname_esc(<<?COLON, Rest/binary>>, S = #ps{acc = Acc}) ->
     parse_hdrvalue_esc(Rest, S#ps{acc = [], acc_len = 0,
-                                  hdrname = lists:reverse(Acc)});
+                                  hdrname = list_to_binary(lists:reverse(Acc))});
 parse_hdrname_esc(<<?BSL>>, S) ->
     more(fun(Rest) -> parse_hdrname_esc(<<?BSL, Rest/binary>>, S) end);
 parse_hdrname_esc(<<?BSL, Ch:8, Rest/binary>>, S) ->
@@ -241,7 +239,7 @@ parse_hdrvalue_esc(<<Ch:8, Rest/binary>>, S = #ps{acc_len = Len,
     end.
 
 finish_hdr_esc(Rest, #ps{acc = Acc, hdrs = Hdrs, hdrname = HdrName} = S) ->
-    Hdrs1 = insert_header(Hdrs, HdrName, lists:reverse(Acc)),
+    Hdrs1 = insert_header(Hdrs, HdrName, list_to_binary(lists:reverse(Acc))),
     parse_headers(Rest, S#ps{hdrs = Hdrs1}).
 
 %%
@@ -427,18 +425,18 @@ header(F, K, D) -> default_value(header(F, K), D).
 
 boolean_header(#stomp_frame{headers = Headers}, Key) ->
     case lists:keysearch(Key, 1, Headers) of
-        {value, {_, "true"}}  -> {ok, true};
-        {value, {_, "false"}} -> {ok, false};
-        {value, {_, "True"}}  -> {ok, true};
-        {value, {_, "False"}} -> {ok, false};
-        _                     -> not_found
+        {value, {_, <<"true">>}}  -> {ok, true};
+        {value, {_, <<"false">>}} -> {ok, false};
+        {value, {_, <<"True">>}}  -> {ok, true};
+        {value, {_, <<"False">>}} -> {ok, false};
+        _                         -> not_found
     end.
 
 boolean_header(F, K, D) -> default_value(boolean_header(F, K), D).
 
 internal_integer_header(Headers, Key) ->
     case lists:keysearch(Key, 1, Headers) of
-        {value, {_, Str}} -> {ok, list_to_integer(string:strip(Str))};
+        {value, {_, Str}} -> {ok, binary_to_integer(string:trim(Str))};
         _                 -> not_found
     end.
 
@@ -447,11 +445,7 @@ integer_header(#stomp_frame{headers = Headers}, Key) ->
 
 integer_header(F, K, D) -> default_value(integer_header(F, K), D).
 
-binary_header(F, K) ->
-    case header(F, K) of
-        {ok, Str} -> {ok, list_to_binary(Str)};
-        not_found -> not_found
-    end.
+binary_header(F, K) -> header(F, K).
 
 binary_header(F, K, D) -> default_value(binary_header(F, K), D).
 
@@ -495,7 +489,7 @@ serialize(#stomp_frame{command = Command,
      lists:map(fun serialize_header/1,
                lists:keydelete(?HEADER_CONTENT_LENGTH, 1, Headers)),
      if
-         Len > 0 -> [?HEADER_CONTENT_LENGTH ++ ":", integer_to_list(Len), ?LF];
+         Len > 0 -> [?HEADER_CONTENT_LENGTH, ?COLON, integer_to_list(Len), ?LF];
          true    -> []
      end,
      ?LF, case BodyFragments of
@@ -509,17 +503,18 @@ serialize_command(Command) -> Command.
 
 serialize_header({K, V}) when is_integer(V) -> hdr(escape(K), integer_to_list(V));
 serialize_header({K, V}) when is_boolean(V) -> hdr(escape(K), boolean_to_list(V));
-serialize_header({K, V}) when is_list(V)    -> hdr(escape(K), escape(V)).
+serialize_header({K, V}) when is_binary(V)  -> hdr(escape(K), escape(V)).
 
 boolean_to_list(true) -> "true";
 boolean_to_list(_)    -> "false".
 
 hdr(K, V) -> [K, ?COLON, V, ?LF].
 
-escape(Str) -> [escape1(Ch) || Ch <- Str].
+escape(Bin) -> escape(Bin, []).
 
-escape1(?COLON) -> [?BSL, ?COLON_ESC];
-escape1(?BSL)   -> [?BSL, ?BSL_ESC];
-escape1(?LF)    -> [?BSL, ?LF_ESC];
-escape1(?CR)    -> [?BSL, ?CR_ESC];
-escape1(Ch)     -> Ch.
+escape(<<>>, Acc) -> lists:reverse(Acc);
+escape(<<?COLON, Rest/binary>>, Acc) -> escape(Rest, [?COLON_ESC, ?BSL | Acc]);
+escape(<<?BSL, Rest/binary>>, Acc) -> escape(Rest, [?BSL_ESC, ?BSL | Acc]);
+escape(<<?LF, Rest/binary>>, Acc) -> escape(Rest, [?LF_ESC, ?BSL | Acc]);
+escape(<<?CR, Rest/binary>>, Acc) -> escape(Rest, [?CR_ESC, ?BSL | Acc]);
+escape(<<Ch:8, Rest/binary>>, Acc) -> escape(Rest, [Ch | Acc]).