Pass CloseOk from upstream down to client

cloudamqp · May 11, 2024 · a451c5e · a451c5e
1 parent 1da1a2c
commit a451c5e
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 7 deletions.
diff --git a/src/amqproxy/client.cr b/src/amqproxy/client.cr
@@ -134,8 +134,14 @@ module AMQProxy
       while frame = @outgoing_frames.receive?
         socket.write_bytes frame, IO::ByteFormat::NetworkEndian
         socket.flush unless expect_more_publish_frames?(frame)
-
-        break if frame.is_a? AMQ::Protocol::Frame::Connection::CloseOk
+        case frame
+        when AMQ::Protocol::Frame::Channel::Close
+          @channel_map[frame.channel] = nil
+        when AMQ::Protocol::Frame::Channel::CloseOk
+          @channel_map.delete(frame.channel)
+        when AMQ::Protocol::Frame::Connection::CloseOk
+          break
+        end
       end
     rescue ex : IO::Error
       # Client closed connection, suppress error
@@ -146,10 +152,6 @@ module AMQProxy
 
     # Send frame to client, channel id should already be remapped by the caller
     def write(frame : AMQ::Protocol::Frame)
-      case frame
-      when AMQ::Protocol::Frame::Channel::Close
-        @channel_map[frame.channel] = nil
-      end
       @outgoing_frames.send frame
     rescue Channel::ClosedError
       # do nothing

diff --git a/src/amqproxy/upstream.cr b/src/amqproxy/upstream.cr
@@ -82,7 +82,9 @@ module AMQProxy
             downstream_channel.write frame
           end
         when AMQ::Protocol::Frame::Channel::CloseOk # when client requested channel close
-          @channels_lock.synchronize { @channels.delete(frame.channel) }
+          if downstream_channel = @channels_lock.synchronize { @channels.delete(frame.channel) }
+            downstream_channel.write(frame)
+          end
         else
           if downstream_channel = @channels_lock.synchronize { @channels[frame.channel]? }
             downstream_channel.write(frame)