Merge pull request #64 from keynslug/sync/emqx-OTP-26.2.5.2-2

sync(27.1): merge OTP-26.2.5.2-2 changes

OTP_VERSION

@@ -1 +1 @@
-27.1-1
+27.1-2

erts/emulator/beam/atom.names

@@ -465,6 +465,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

@@ -1502,6 +1502,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.
 
@@ -185,6 +185,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;
@@ -208,6 +213,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">>,
@@ -242,7 +252,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.
@@ -303,6 +313,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

@@ -1723,7 +1723,7 @@ returned.
   latin-1 newline character. The delimiter byte is included in the returned
   packet unless the line was truncated according to option `line_length`.
 
-- **`asn1 | cdr | sunrm | fcgi | tpkt`** - The header is _not_ stripped off.
+- **`asn1 | cdr | sunrm | fcgi | tpkt | mqtt`** - The header is _not_ stripped off.
 
   The meanings of the packet types are as follows:
 
@@ -1737,6 +1737,8 @@ returned.
 
   - **`tpkt` \- TPKT format \[RFC1006]**
 
+  - **`mqtt` \- MQTT packet \[mqtt-v5.0\] / \[mqtt-v3.1.1\]
+
 - **`http | httph | http_bin | httph_bin`** - The Hypertext Transfer Protocol.
   The packets are returned with the format according to `HttpPacket` described
   earlier. A packet is either a request, a response, a header, or an end of
@@ -1791,7 +1793,8 @@ Examples:
                                   {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

@@ -1740,6 +1740,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/src/gen_tcp.erl

@@ -278,7 +278,8 @@ way, option `send_timeout` comes in handy.
         {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

@@ -1314,7 +1314,8 @@ nopush_or_cork() ->
 
 %% -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) ->
@@ -1327,6 +1328,7 @@ is_packet_option_value(Value) ->
         fcgi -> true;
         line -> true;
         tpkt -> true;
+        mqtt -> true;
         http -> true;
         httph -> true;
         http_bin -> true;
@@ -2507,6 +2509,7 @@ packet_header_length(PacketType) ->
         sunrm   -> 4;
         fcgi    -> 8;
         tpkt    -> 4;
+        mqtt    -> 2;
         ssl     -> 5;
         ssl_tls -> 5;
         asn1    -> 2;

lib/kernel/src/inet.erl

@@ -1255,12 +1255,12 @@ The following options are available:
 
     The 4-byte header is limited to 2Gb.
 
-  - **`asn1 | cdr | sunrm | fcgi | tpkt | line`** - These packet types only have
-    effect on receiving. When sending a packet, it is the responsibility of the
-    application to supply a correct header. On receiving, however, one message
-    is sent to the controlling process for each complete packet received, and,
-    similarly, each call to `gen_tcp:recv/2,3` returns one complete packet. The
-    header is _not_ stripped off.
+  - **`asn1 | cdr | sunrm | fcgi | tpkt | mqtt | line`** - These packet types only
+    have effect on receiving. When sending a packet, it is the responsibility
+    of the application to supply a correct header. On receiving, however, one
+    message is sent to the controlling process for each complete packet received,
+    and, similarly, each call to `gen_tcp:recv/2,3` returns one complete packet.
+    The header is _not_ stripped off.
 
     The meanings of the packet types are as follows:
 
@@ -1269,6 +1269,7 @@ The following options are available:
     - `cdr` - CORBA (GIOP 1.1)
     - `fcgi` - Fast CGI
     - `tpkt` - TPKT format \[RFC1006]
+    - `mqtt` - MQTT packet \[mqtt-v5.0\] / \[mqtt-v3.1.1\]
     - `line` - Line mode, a packet is a line-terminated with newline, lines
       longer than the receive buffer are truncated
 

lib/kernel/src/inet_int.hrl

@@ -213,6 +213,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