Merge pull request #63 from emqx/sync/emqx-OTP-26.2.5-5

sync: merge emqx-OTP-26.2.5-5 changes

OTP_VERSION

@@ -1 +1 @@
-26.2.5.2-1
+26.2.5.2-2

erts/doc/src/erlang.xml

@@ -1554,7 +1554,7 @@ Z = erlang:crc32_combine(X,Y,iolist_size(Data2)).</code>
               byte is included in the returned packet unless the line
               was truncated according to option <c>line_length</c>.</p>
           </item>
-          <tag><c>asn1 | cdr | sunrm | fcgi | tpkt</c></tag>
+          <tag><c>asn1 | cdr | sunrm | fcgi | tpkt | mqtt</c></tag>
           <item>
             <p>The header is <em>not</em> stripped off.</p>
             <p>The meanings of the packet types are as follows:</p>
@@ -1564,6 +1564,7 @@ Z = erlang:crc32_combine(X,Y,iolist_size(Data2)).</code>
               <tag><c>cdr</c> - CORBA (GIOP 1.1)</tag><item></item>
               <tag><c>fcgi</c> - Fast CGI</tag><item></item>
               <tag><c>tpkt</c> - TPKT format [RFC1006]</tag><item></item>
+              <tag><c>mqtt</c> - MQTT packet [mqtt-v5.0] / [mqtt-v3.1.1]</tag><item></item>
             </taglist>
           </item>
           <tag><c>http | httph | http_bin | httph_bin</c></tag>

erts/emulator/beam/atom.names

@@ -457,6 +457,7 @@ atom monitors
 atom monotonic
 atom monotonic_timestamp
 atom more
+atom mqtt
 atom multi_scheduling
 atom multiline
 atom nano_seconds

erts/emulator/beam/erl_bif_port.c

@@ -1570,6 +1570,7 @@ BIF_RETTYPE decode_packet_3(BIF_ALIST_3)
     case am_httph: type = TCP_PB_HTTPH; break;
     case am_http_bin: type = TCP_PB_HTTP_BIN; break;
     case am_httph_bin: type = TCP_PB_HTTPH_BIN; break;
+    case am_mqtt: type = TCP_PB_MQTT; break;
     case am_ssl_tls: type = TCP_PB_SSL_TLS; break;
     default:
         BIF_P->fvalue = am_badopt;

erts/emulator/beam/packet_parser.c

@@ -471,6 +471,40 @@ int packet_get_length(enum PacketParseType htype,
             plen = get_int16(&ptr[3]);
         }
         goto remain;
+
+    case TCP_PB_MQTT: {
+        /* Byte 1: MQTT Control Packet fixed header
+         * Bytes 2-2/3/4/5: Remaining Length (variable byte integer)
+         */
+        byte vb, ptype;
+        hlen = 2;
+        plen = 0;
+        if (n < 1) goto more;
+        /* Bits 4-8: Packet type */
+        ptype = ptr[0] >> 4;
+        /* ERROR: Type 0 is reserved, forbidden */
+        if (ptype == 0)
+            goto error;
+        while (hlen <= 1 + 4) {
+            if (hlen > n) goto more;
+            vb = ptr[hlen - 1] & 0x7F;
+            plen |= vb << (7 * (hlen - 2));
+            if (ptr[hlen - 1] & 0x80) {
+                hlen = hlen + 1;
+            }
+            else {
+                /* NOTE: Tolerate minumum-number-of-bytes rule violation
+                 * [MQTT-1.5.5-1]
+                 */
+                goto packet;
+            }
+        }
+        /* ERROR: variable byte integer >4 bytes long */
+        goto error;
+    packet:
+        /* No special parsing for now */
+        goto remain;
+    }
 
     default:
         DEBUGF((" => case error\r\n"));

erts/emulator/beam/packet_parser.h

@@ -44,7 +44,8 @@ enum PacketParseType {
     TCP_PB_HTTPH    = 11,
     TCP_PB_SSL_TLS  = 12,
     TCP_PB_HTTP_BIN = 13,
-    TCP_PB_HTTPH_BIN = 14
+    TCP_PB_HTTPH_BIN = 14,
+    TCP_PB_MQTT     = 15
 };
 
 typedef struct http_atom {

erts/emulator/test/decode_packet_SUITE.erl

@@ -26,15 +26,15 @@
 
 -export([all/0, suite/0,groups/0,
          init_per_testcase/2,end_per_testcase/2,
-         basic/1, ipv6/1, packet_size/1, neg/1, http/1, line/1, ssl/1, otp_8536/1,
-         otp_9389/1, otp_9389_line/1]).
+         basic/1, ipv6/1, packet_size/1, neg/1, http/1, line/1, mqtt/1, ssl/1,
+         otp_8536/1, otp_9389/1, otp_9389_line/1]).
 
 suite() ->
     [{ct_hooks,[ts_install_cth]},
      {timetrap, {minutes, 1}}].
 
 all() -> 
-    [basic, packet_size, neg, http, line, ssl, otp_8536,
+    [basic, packet_size, neg, http, line, mqtt, ssl, otp_8536,
      otp_9389, otp_9389_line, ipv6].
 
 groups() -> 
@@ -62,7 +62,7 @@ basic(Config) when is_list(Config) ->
     {more, undefined} = decode_pkt(2,<<0>>),
     {more, undefined} = decode_pkt(4,<<0,0,0>>),
 
-    Types = [1,2,4,asn1,sunrm,cdr,fcgi,tpkt,ssl_tls],
+    Types = [1,2,4,asn1,sunrm,cdr,fcgi,tpkt,mqtt,ssl_tls],
 
     %% Run tests for different header types and bit offsets.
 
@@ -183,6 +183,11 @@ pack(tpkt,Bin) ->
     Size = byte_size(Bin) + 4,
     Res = <<Ver:8,Reserv:8,Size:16,Bin/binary>>,
     {Res, Res};
+pack(mqtt,Bin) ->
+    Type = 3, % PUBLISH
+    Size = pack_mqtt_vbi(byte_size(Bin)),
+    Res = <<Type:4,0:4,Size/binary,Bin/binary>>,
+    {Res, Res};
 pack(ssl_tls,Bin) ->
     Content = case (rand:uniform(256) - 1) of
                   C when C<128 -> C;
@@ -206,6 +211,11 @@ pack_ssl(Content, Major, Minor, Body) ->
     end,
     {Res, {ssl_tls,[],C,{Major,Minor}, Data}}.
 
+pack_mqtt_vbi(N) when N =< 2#01111111 ->
+    <<0:1, N:7>>;
+pack_mqtt_vbi(N) ->
+    <<1:1, (N rem 2#10000000):7, (pack_mqtt_vbi(N div 2#10000000))/binary>>.
+
 ipv6(Config) when is_list(Config) ->
     %% Test with port
     Packet = <<"GET http://[FEDC:BA98:7654:3210:FEDC:BA98:7654:3210]:4000/echo_components HTTP/1.1\r\nhost: orange\r\n\r\n">>,
@@ -240,7 +250,7 @@ packet_size(Config) when is_list(Config) ->
                         ok
                 end
         end,
-    lists:foreach(F, [{T,D} || T<-[1,2,4,asn1,sunrm,cdr,fcgi,tpkt,ssl_tls],
+    lists:foreach(F, [{T,D} || T<-[1,2,4,asn1,sunrm,cdr,fcgi,tpkt,mqtt,ssl_tls],
                                D<-lists:seq(0, byte_size(Packet)*2)]),
 
     %% Test OTP-8102, "negative" 4-byte sizes.
@@ -301,6 +311,20 @@ neg(Config) when is_list(Config) ->
     ok.
 
 
+mqtt(_Config) ->
+    Type = 1, % CONNECT
+    {more, undefined} = decode_pkt(mqtt,<<>>),
+    {error, invalid} = decode_pkt(mqtt,<<0>>),
+    {more, undefined} = decode_pkt(mqtt,<<Type:4,0:4>>),
+    {more, 2 + 10} = decode_pkt(mqtt,<<Type:4,0:4, 10:8>>),
+    {more, undefined} = decode_pkt(mqtt,<<Type:4,0:4, 1:1,10:7>>),
+    {more, 3 + 138} = decode_pkt(mqtt,<<Type:4,0:4, 1:1,10:7,1:8>>),
+    {more, 5 + 13*128*128*128 + 12*128*128 + 11*128 + 10} =
+        decode_pkt(mqtt,<<Type:4,0:4, 1:1,10:7,1:1,11:7,1:1,12:7,0:1,13:7>>),
+    {error, invalid} =
+        decode_pkt(mqtt,<<Type:4,0:4, 1:1,10:7,1:1,11:7,1:1,12:7,1:1,13:7>>).
+
+
 http(Config) when is_list(Config) ->
     <<"foo">> = http_do(http_request("foo")),
     <<" bar">> = http_do(http_request(" bar")),

erts/preloaded/src/erlang.erl

@@ -866,7 +866,8 @@ date() ->
                                   {more, Length} |
                                   {error, Reason} when
       Type :: 'raw' | 0 | 1 | 2 | 4 | 'asn1' | 'cdr' | 'sunrm' | 'fcgi'
-            | 'tpkt' | 'line' | 'http' | 'http_bin' | 'httph' | 'httph_bin',
+            | 'tpkt' | 'line' | 'http' | 'http_bin' | 'httph' | 'httph_bin'
+            | 'mqtt',
       Bin :: binary(),
       Options :: [Opt],
       Opt :: {packet_size, non_neg_integer()}

erts/preloaded/src/prim_inet.erl

@@ -1738,6 +1738,7 @@ type_opt_1(packet) ->
 	   {httph,?TCP_PB_HTTPH},
 	   {http_bin, ?TCP_PB_HTTP_BIN},
 	   {httph_bin,?TCP_PB_HTTPH_BIN},
+	   {mqtt, ?TCP_PB_MQTT},
 	   {ssl, ?TCP_PB_SSL_TLS}, % obsolete
 	   {ssl_tls, ?TCP_PB_SSL_TLS}]};
 type_opt_1(line_delimiter)  -> int;

lib/kernel/doc/src/inet.xml

@@ -1433,7 +1433,7 @@ setcap cap_sys_admin,cap_sys_ptrace,cap_dac_read_search+epi beam.smp</code>
                   is stripped off on each receive operation.</p>
                 <p>The 4-byte header is limited to 2Gb.</p>
               </item>
-              <tag><c>asn1 | cdr | sunrm | fcgi | tpkt | line</c></tag>
+              <tag><c>asn1 | cdr | sunrm | fcgi | tpkt | mqtt | line</c></tag>
               <item>
                 <p>These packet types only have effect on receiving.
                   When sending a packet, it is the responsibility of
@@ -1450,6 +1450,7 @@ setcap cap_sys_admin,cap_sys_ptrace,cap_dac_read_search+epi beam.smp</code>
 		  <item><c>cdr</c> - CORBA (GIOP 1.1)</item>
 		  <item><c>fcgi</c> - Fast CGI</item>
 		  <item><c>tpkt</c> - TPKT format [RFC1006]</item>
+		  <item><c>mqtt</c> - MQTT packet [mqtt-v5.0] / [mqtt-v3.1.1]</item>
 		  <item><c>line</c> - Line mode, a packet is a line-terminated
 		    with newline, lines longer than the receive buffer are
 		    truncated</item>

lib/kernel/src/gen_tcp.erl

@@ -56,7 +56,8 @@
         {nodelay,         boolean()} |
         {packet,
          0 | 1 | 2 | 4 | raw | sunrm |  asn1 |
-         cdr | fcgi | line | tpkt | http | httph | http_bin | httph_bin } |
+         cdr | fcgi | line | tpkt | http | httph | http_bin | httph_bin |
+         mqtt } |
         {packet_size,     non_neg_integer()} |
         {priority,        non_neg_integer()} |
         {raw,

lib/kernel/src/gen_tcp_socket.erl

@@ -1637,7 +1637,8 @@ module_socket(#params{socket = Socket}) ->
 
 %% -type packet_option_value() ::
 %%         0 | 1 | 2 | 4 | raw | sunrm |  asn1 |
-%%         cdr | fcgi | line | tpkt | http | httph | http_bin | httph_bin.
+%%         cdr | fcgi | line | tpkt | mqtt | http | httph | http_bin |
+%%         httph_bin.
 
 -compile({inline, [is_packet_option_value/1]}).
 is_packet_option_value(Value) ->
@@ -1650,6 +1651,7 @@ is_packet_option_value(Value) ->
         fcgi -> true;
         line -> true;
         tpkt -> true;
+        mqtt -> true;
         http -> true;
         httph -> true;
         http_bin -> true;

lib/kernel/src/inet_int.hrl

@@ -212,6 +212,7 @@
 -define(TCP_PB_SSL_TLS, 12).
 -define(TCP_PB_HTTP_BIN,13).
 -define(TCP_PB_HTTPH_BIN,14).
+-define(TCP_PB_MQTT,    15).
 
 
 %% getstat, INET_REQ_GETSTAT