Use AMQ::Protocol::Stream for frame size limitations

carlhoerberg · carlhoerberg · commit 49f935fefc89 · 2025-11-19T22:31:37.000+01:00
diff --git a/shard.lock b/shard.lock
@@ -6,7 +6,7 @@ shards:
 
   amq-protocol:
     git: https://github.com/cloudamqp/amq-protocol.cr.git
-    version: 1.1.16
+    version: 1.1.19
 
   amqp-client:
     git: https://github.com/cloudamqp/amqp-client.cr.git
@@ -18,7 +18,7 @@ shards:
 
   mqtt-protocol:
     git: https://github.com/84codes/mqtt-protocol.cr.git
-    version: 0.2.0+git.commit.5cf504b32313b3b63eecde22e2a136b9b9935b8b
+    version: 0.3.0
 
   systemd:
     git: https://github.com/84codes/systemd.cr.git
diff --git a/spec/connection_spec.cr b/spec/connection_spec.cr
@@ -143,7 +143,7 @@ describe LavinMQ::Server do
         ch = conn.channel
         frame_max = LavinMQ::Config.instance.frame_max
         bytes = frame_max + 1
-        expect_raises(AMQP::Client::Connection::ClosedException, /frame size #{bytes} exceeded max #{frame_max}/) do
+        expect_raises(AMQP::Client::Connection::ClosedException, /Frame size #{bytes} exceeds max frame size #{frame_max}/) do
           conn.unsafe_write AMQ::Protocol::Frame::Basic::Publish.new(ch.id, 0_u16, "amq.direct", "test", false, false)
           conn.unsafe_write AMQ::Protocol::Frame::Header.new(ch.id, 60_u16, 0_u16, bytes.to_u64, AMQ::Protocol::Properties.new)
           conn.unsafe_write AMQ::Protocol::Frame::BytesBody.new(ch.id, bytes, Slice.new(bytes.to_i32, 0_u8))
diff --git a/src/lavinmq/amqp/client.cr b/src/lavinmq/amqp/client.cr
@@ -117,9 +117,9 @@ module LavinMQ
 
       private def read_loop
         received_bytes = 0_u32
-        socket = @socket
+        stream = AMQ::Protocol::Stream.new(@socket, @max_frame_size)
         loop do
-          AMQP::Frame.from_io(socket) do |frame|
+          stream.next_frame do |frame|
             {% unless flag?(:release) %}
               @log.trace { "Received #{frame.inspect}" }
             {% end %}
@@ -155,6 +155,10 @@ module LavinMQ
           end
         rescue IO::TimeoutError
           send_heartbeat || break
+        rescue ex : AMQ::Protocol::Error::FrameSizeError
+          @log.error { ex.inspect }
+          send_frame_error(ex.message)
+          break
         rescue ex : AMQ::Protocol::Error::NotImplemented
           @log.error { ex.inspect }
           send_not_implemented(ex)
diff --git a/src/lavinmq/amqp/connection_factory.cr b/src/lavinmq/amqp/connection_factory.cr
@@ -20,18 +20,19 @@ module LavinMQ
         metadata = ::Log::Metadata.build({address: connection_info.remote_address.to_s})
         logger = Logger.new(Log, metadata)
         if confirm_header(socket, logger)
-          if start_ok = start(socket, logger)
-            if user = authenticate(socket, connection_info.remote_address, start_ok, logger)
-              if tune_ok = tune(socket, logger)
-                if vhost = open(socket, user, logger)
+          stream = AMQ::Protocol::Stream.new(socket)
+          if start_ok = start(stream, logger)
+            if user = authenticate(stream, connection_info.remote_address, start_ok, logger)
+              if tune_ok = tune(stream, logger)
+                if vhost = open(stream, user, logger)
                   socket.read_timeout = heartbeat_timeout(tune_ok)
                   return LavinMQ::AMQP::Client.new(socket, connection_info, vhost, user, tune_ok, start_ok)
                 end
               end
             end
           end
         end
-      rescue ex : IO::TimeoutError | IO::Error | OpenSSL::SSL::Error | AMQ::Protocol::Error::FrameDecode
+      rescue ex : IO::TimeoutError | IO::Error | OpenSSL::SSL::Error | AMQ::Protocol::Error
         Log.warn { "#{ex} when #{connection_info.remote_address} tried to establish connection" }
         nil
       rescue ex
@@ -81,7 +82,7 @@ module LavinMQ
         start = AMQP::Frame::Connection::Start.new(server_properties: SERVER_PROPERTIES)
         socket.write_bytes start, ::IO::ByteFormat::NetworkEndian
         socket.flush
-        start_ok = AMQP::Frame.from_io(socket) { |f| f.as(AMQP::Frame::Connection::StartOk) }
+        start_ok = socket.next_frame.as(AMQP::Frame::Connection::StartOk)
         if start_ok.bytesize > 4096
           log.warn { "StartOk frame was #{start_ok.bytesize} bytes, max allowed is 4096 bytes" }
           return
@@ -137,35 +138,33 @@ module LavinMQ
           frame_max: frame_max,
           heartbeat: Config.instance.heartbeat), IO::ByteFormat::NetworkEndian
         socket.flush
-        tune_ok = AMQP::Frame.from_io(socket) do |frame|
-          case frame
-          when AMQP::Frame::Connection::TuneOk
-            if LOW_FRAME_MAX_RANGE.includes? frame.frame_max
-              log.warn { "Suggested Frame max (#{frame.frame_max}) too low, closing connection" }
-              reply_text = "failed to negotiate connection parameters: negotiated frame_max = #{frame.frame_max} is lower than the minimum allowed value (#{MIN_FRAME_MAX})"
-              return close_connection(socket, ConnectionReplyCode::NOT_ALLOWED, reply_text, frame)
-            end
-            if too_high?(frame.frame_max, Config.instance.frame_max)
-              log.warn { "Suggested Frame max (#{frame.frame_max}) too high, closing connection" }
-              reply_text = "failed to negotiate connection parameters: negotiated frame_max = #{frame.frame_max} is higher than the maximum allowed value (#{Config.instance.frame_max})"
-              return close_connection(socket, ConnectionReplyCode::NOT_ALLOWED, reply_text, frame)
-            end
-            if too_high?(frame.channel_max, Config.instance.channel_max)
-              log.warn { "Suggested Channel max (#{frame.channel_max}) too high, closing connection" }
-              reply_text = "failed to negotiate connection parameters: negotiated channel_max = #{frame.channel_max} is higher than the maximum allowed value (#{Config.instance.channel_max})"
-              return close_connection(socket, ConnectionReplyCode::NOT_ALLOWED, reply_text, frame)
-            end
-            frame
-          else
-            log.warn { "Expected TuneOk Frame got #{frame.inspect}" }
-            return
+        frame = socket.next_frame
+        case frame
+        when AMQP::Frame::Connection::TuneOk
+          if LOW_FRAME_MAX_RANGE.includes? frame.frame_max
+            log.warn { "Suggested Frame max (#{frame.frame_max}) too low, closing connection" }
+            reply_text = "failed to negotiate connection parameters: negotiated frame_max = #{frame.frame_max} is lower than the minimum allowed value (#{MIN_FRAME_MAX})"
+            return close_connection(socket, ConnectionReplyCode::NOT_ALLOWED, reply_text, frame)
           end
+          if too_high?(frame.frame_max, Config.instance.frame_max)
+            log.warn { "Suggested Frame max (#{frame.frame_max}) too high, closing connection" }
+            reply_text = "failed to negotiate connection parameters: negotiated frame_max = #{frame.frame_max} is higher than the maximum allowed value (#{Config.instance.frame_max})"
+            return close_connection(socket, ConnectionReplyCode::NOT_ALLOWED, reply_text, frame)
+          end
+          if too_high?(frame.channel_max, Config.instance.channel_max)
+            log.warn { "Suggested Channel max (#{frame.channel_max}) too high, closing connection" }
+            reply_text = "failed to negotiate connection parameters: negotiated channel_max = #{frame.channel_max} is higher than the maximum allowed value (#{Config.instance.channel_max})"
+            return close_connection(socket, ConnectionReplyCode::NOT_ALLOWED, reply_text, frame)
+          end
+          frame
+        else
+          log.warn { "Expected TuneOk Frame got #{frame.inspect}" }
+          return
         end
-        tune_ok
       end
 
       def open(socket, user, log)
-        open = AMQP::Frame.from_io(socket) { |f| f.as(AMQP::Frame::Connection::Open) }
+        open = socket.next_frame.as(AMQP::Frame::Connection::Open)
         vhost_name = open.vhost.empty? ? "/" : open.vhost
         if vhost = @vhosts[vhost_name]?
           if user.permissions[vhost_name]?
diff --git a/src/lavinmq/definitions_generator.cr b/src/lavinmq/definitions_generator.cr
@@ -187,9 +187,9 @@ class LavinMQCtl
       queues = Array(Frame::Method::Queue::Declare).new
       File.open(File.join(vhost_dir, "definitions.amqp")) do |defs|
         _schema = defs.read_bytes Int32
+        stream = AMQ::Protocol::Stream.new(defs, format: IO::ByteFormat::SystemEndian)
         loop do
-          frame = Frame.from_io(defs, IO::ByteFormat::SystemEndian) { |f| f }
-          case frame
+          case frame = stream.next_frame
           when Frame::Method::Queue::Declare
             queues << frame
           when Frame::Method::Queue::Delete
@@ -206,9 +206,9 @@ class LavinMQCtl
       exchanges = Array(Frame::Method::Exchange::Declare).new
       File.open(File.join(vhost_dir, "definitions.amqp")) do |defs|
         _schema = defs.read_bytes Int32
+        stream = AMQ::Protocol::Stream.new(defs, format: IO::ByteFormat::SystemEndian)
         loop do
-          frame = Frame.from_io(defs, IO::ByteFormat::SystemEndian) { |f| f }
-          case frame
+          case frame = stream.next_frame
           when Frame::Method::Exchange::Declare
             exchanges << frame
           when Frame::Method::Exchange::Delete
@@ -225,9 +225,9 @@ class LavinMQCtl
       bindings = Array(Frame::Method::Queue::Bind).new
       File.open(File.join(vhost_dir, "definitions.amqp")) do |defs|
         _schema = defs.read_bytes Int32
+        stream = AMQ::Protocol::Stream.new(defs, format: IO::ByteFormat::SystemEndian)
         loop do
-          frame = Frame.from_io(defs, IO::ByteFormat::SystemEndian) { |f| f }
-          case frame
+          case frame = stream.next_frame
           when Frame::Method::Queue::Bind
             bindings << frame
           when Frame::Method::Queue::Unbind
@@ -244,9 +244,9 @@ class LavinMQCtl
       bindings = Array(Frame::Method::Exchange::Bind).new
       File.open(File.join(vhost_dir, "definitions.amqp")) do |defs|
         _schema = defs.read_bytes Int32
+        stream = AMQ::Protocol::Stream.new(defs, format: IO::ByteFormat::SystemEndian)
         loop do
-          frame = Frame.from_io(defs, IO::ByteFormat::SystemEndian) { |f| f }
-          case frame
+          case frame = stream.next_frame
           when Frame::Method::Exchange::Bind
             bindings << frame
           when Frame::Method::Exchange::Unbind
diff --git a/src/lavinmq/schema.cr b/src/lavinmq/schema.cr
@@ -109,14 +109,13 @@ module LavinMQ
             version = io.read_bytes Int32
             raise UnsupportedSchemaVersion.new(version, io.path) unless version == 1
 
+            stream = AMQ::Protocol::Stream.new(io, format: IO::ByteFormat::SystemEndian)
             loop do
-              AMQP::Frame.from_io(io, IO::ByteFormat::SystemEndian) do |frame|
-                case frame
-                when AMQP::Frame::Queue::Declare
-                  queues.push frame.queue_name
-                when AMQP::Frame::Queue::Delete
-                  queues.delete frame.queue_name
-                end
+              case frame = stream.next_frame
+              when AMQP::Frame::Queue::Declare
+                queues.push frame.queue_name
+              when AMQP::Frame::Queue::Delete
+                queues.delete frame.queue_name
               end
             rescue IO::EOFError
               break
diff --git a/src/lavinmq/vhost.cr b/src/lavinmq/vhost.cr
@@ -495,44 +495,44 @@ module LavinMQ
       @definitions_lock.synchronize do
         @log.debug { "Verifying schema" }
         SchemaVersion.verify(io, :definition)
+        stream = AMQ::Protocol::Stream.new(io, format: IO::ByteFormat::SystemEndian)
         loop do
-          AMQP::Frame.from_io(io, IO::ByteFormat::SystemEndian) do |f|
-            @log.trace { "Reading frame #{f.inspect}" }
-            case f
-            when AMQP::Frame::Exchange::Declare
-              exchanges[f.exchange_name] = f
-            when AMQP::Frame::Exchange::Delete
-              exchanges.delete f.exchange_name
-              exchange_bindings.delete f.exchange_name
-              should_compact = true
-            when AMQP::Frame::Exchange::Bind
-              exchange_bindings[f.destination] << f
-            when AMQP::Frame::Exchange::Unbind
-              exchange_bindings[f.destination].reject! do |b|
-                b.source == f.source &&
-                  b.routing_key == f.routing_key &&
-                  b.arguments == f.arguments
-              end
-              should_compact = true
-            when AMQP::Frame::Queue::Declare
-              queues[f.queue_name] = f
-            when AMQP::Frame::Queue::Delete
-              queues.delete f.queue_name
-              queue_bindings.delete f.queue_name
-              should_compact = true
-            when AMQP::Frame::Queue::Bind
-              queue_bindings[f.queue_name] << f
-            when AMQP::Frame::Queue::Unbind
-              queue_bindings[f.queue_name].reject! do |b|
-                b.exchange_name == f.exchange_name &&
-                  b.routing_key == f.routing_key &&
-                  b.arguments == f.arguments
-              end
-              should_compact = true
-            else
-              raise "Cannot apply frame #{f.class} in vhost #{@name}"
+          f = stream.next_frame
+          @log.trace { "Reading frame #{f.inspect}" }
+          case f
+          when AMQP::Frame::Exchange::Declare
+            exchanges[f.exchange_name] = f
+          when AMQP::Frame::Exchange::Delete
+            exchanges.delete f.exchange_name
+            exchange_bindings.delete f.exchange_name
+            should_compact = true
+          when AMQP::Frame::Exchange::Bind
+            exchange_bindings[f.destination] << f
+          when AMQP::Frame::Exchange::Unbind
+            exchange_bindings[f.destination].reject! do |b|
+              b.source == f.source &&
+                b.routing_key == f.routing_key &&
+                b.arguments == f.arguments
             end
-          end # Frame.from_io
+            should_compact = true
+          when AMQP::Frame::Queue::Declare
+            queues[f.queue_name] = f
+          when AMQP::Frame::Queue::Delete
+            queues.delete f.queue_name
+            queue_bindings.delete f.queue_name
+            should_compact = true
+          when AMQP::Frame::Queue::Bind
+            queue_bindings[f.queue_name] << f
+          when AMQP::Frame::Queue::Unbind
+            queue_bindings[f.queue_name].reject! do |b|
+              b.exchange_name == f.exchange_name &&
+                b.routing_key == f.routing_key &&
+                b.arguments == f.arguments
+            end
+            should_compact = true
+          else
+            raise "Cannot apply frame #{f.class} in vhost #{@name}"
+          end
         rescue ex : IO::EOFError
           break
         end # loop
