Make MQTT max_packet_size configurable

Default to 256 MiB as before.

Co-authored-by: Jon Börjesson <[email protected]>

Author: 2 people

src/lavinmq/config.cr

@@ -77,6 +77,7 @@ module LavinMQ
     property default_user : String = ENV.fetch("LAVINMQ_DEFAULT_USER", "guest")
     property default_password : String = ENV.fetch("LAVINMQ_DEFAULT_PASSWORD", DEFAULT_PASSWORD_HASH) # Hashed password for default user
     property max_consumers_per_channel = 0
+    property mqtt_max_packet_size = 268_435_455_u32 # bytes
     @@instance : Config = self.new
 
     def self.instance : LavinMQ::Config
@@ -367,6 +368,7 @@ module LavinMQ
         when "max_inflight_messages"    then @max_inflight_messages = v.to_u16
         when "default_vhost"            then @default_mqtt_vhost = v
         when "permission_check_enabled" then @mqtt_permission_check_enabled = true?(v)
+        when "max_packet_size"          then @mqtt_max_packet_size = v.to_u32
         else
           STDERR.puts "WARNING: Unrecognized configuration 'mqtt/#{config}'"
         end

src/lavinmq/mqtt/broker.cr

@@ -37,11 +37,11 @@ module LavinMQ
         true
       end
 
-      def add_client(socket, connection_info, user, packet)
+      def add_client(io, connection_info, user, packet)
         if prev_client = @clients[packet.client_id]?
           prev_client.close("New client #{connection_info.remote_address} (username=#{packet.username}) connected as #{packet.client_id}")
         end
-        client = MQTT::Client.new(socket,
+        client = MQTT::Client.new(io,
           connection_info,
           user,
           self,

src/lavinmq/mqtt/client.cr

@@ -26,15 +26,14 @@ module LavinMQ
         @broker.vhost
       end
 
-      def initialize(@socket : ::IO,
+      def initialize(@io : MQTT::IO,
                      @connection_info : ConnectionInfo,
                      @user : Auth::User,
                      @broker : MQTT::Broker,
                      @client_id : String,
                      @clean_session : Bool = false,
                      @keepalive : UInt16 = 30,
                      @will : MQTT::Will? = nil)
-        @io = MQTT::IO.new(@socket)
         @lock = Mutex.new
         @waitgroup = WaitGroup.new(1)
         @name = "#{@connection_info.remote_address} -> #{@connection_info.local_address}"
@@ -50,7 +49,7 @@ module LavinMQ
 
       private def read_loop
         received_bytes = 0_u32
-        socket = @socket
+        socket = @io.io
         if socket.responds_to?(:"read_timeout=")
           # 50% grace period according to [MQTT-3.1.2-24]
           socket.read_timeout = @keepalive.zero? ? nil : (@keepalive * 1.5).seconds
@@ -88,7 +87,7 @@ module LavinMQ
       end
 
       def read_and_handle_packet
-        packet : MQTT::Packet = MQTT::Packet.from_io(@io)
+        packet = @io.read_packet
         @log.trace { "Received packet:  #{packet.inspect}" }
         @recv_oct_count.add(packet.bytesize, :relaxed)
 
@@ -106,8 +105,8 @@ module LavinMQ
 
       def send(packet)
         @lock.synchronize do
-          packet.to_io(@io)
-          @socket.flush
+          @io.write_packet(packet)
+          @io.flush
           @send_oct_count.add(packet.bytesize, :relaxed)
         end
         case packet
@@ -227,7 +226,7 @@ module LavinMQ
       end
 
       private def close_socket
-        socket = @socket
+        socket = @io
         if socket.responds_to?(:"write_timeout=")
           socket.write_timeout = 1.seconds
         end

src/lavinmq/mqtt/connection_factory.cr

@@ -20,15 +20,15 @@ module LavinMQ
         metadata = ::Log::Metadata.build({address: connection_info.remote_address.to_s})
         logger = Logger.new(Log, metadata)
         begin
-          io = MQTT::IO.new(socket)
-          if packet = Packet.from_io(socket).as?(Connect)
+          io = MQTT::IO.new(socket, @config.mqtt_max_packet_size)
+          if packet = io.read_packet.as?(Connect)
             logger.trace { "recv #{packet.inspect}" }
             if user_and_broker = authenticate(io, packet)
               user, broker = user_and_broker
               packet = assign_client_id(packet) if packet.client_id.empty?
               session_present = broker.session_present?(packet.client_id, packet.clean_session?)
               connack io, session_present, Connack::ReturnCode::Accepted
-              return broker.add_client(socket, connection_info, user, packet)
+              return broker.add_client(io, connection_info, user, packet)
             else
               logger.warn { "Authentication failure for user \"#{packet.username}\"" }
               connack io, false, Connack::ReturnCode::NotAuthorized