Merge pull request #502 from CopernicaMarketingSoftware/connection-blocked-notification

Implement connection.blocked notification handling

include/amqpcpp/connectionhandler.h

@@ -192,6 +192,39 @@ public:
         // make sure compilers dont complain about unused parameters
         (void) connection; 
     }
+
+    /**
+     *  Method that is called when the AMQP connection was blocked.
+     * 
+     *  This method is called, when the server connection gets blocked for the first
+     *  time due to the broker running low on a resource (memory or disk). For
+     *  example, when a RabbitMQ node detects that it is low on RAM, it sends a
+     *  notification to all connected publishing clients supporting this feature.
+     *  If before the connections are unblocked the node also starts running low on
+     *  disk space, another notification will not be sent.
+     *
+     *  @param  connection      The connection that was blocked
+     *  @param  reason          Why was the connection blocked
+     */
+    virtual void onBlocked(Connection *connection, const char *reason) 
+    {
+        // make sure compilers dont complain about unused parameters
+        (void) connection;
+    }
+
+    /**
+     *  Method that is called when the AMQP connection is no longer blocked.
+     * 
+     *  This method is called when all resource alarms have cleared and the
+     *  connection is fully unblocked.
+     *
+     *  @param  connection      The connection that is no longer blocked
+     */
+    virtual void onUnblocked(Connection *connection) 
+    { 
+        // make sure compilers dont complain about unused parameters
+        (void) connection;
+    }
 };
 
 /**

include/amqpcpp/connectionimpl.h

@@ -66,7 +66,7 @@ protected:
         state_handshake,            // busy with the handshake to open the connection
         state_connected,            // connection is set up and ready for communication
         state_closing,              // connection is busy closing (we have sent the close frame)
-        state_closed                // connection is closed
+        state_closed,               // connection is closed
     } _state = state_protocol;
 
     /**
@@ -428,6 +428,25 @@ public:
         _handler->onClosed(_parent);
     }
 
+    /**
+     *  Report that the connection is blocked
+     *  @param  reason
+     */
+    void reportBlocked(const char *reason)
+    {
+        // inform the handler
+        _handler->onBlocked(_parent, reason);
+    }
+
+    /**
+     *  Report that the connection is unblocked
+     */
+    void reportUnblocked()
+    {
+        // inform the handler
+        _handler->onUnblocked(_parent);
+    }
+
     /**
      *  Retrieve the amount of channels this connection has
      *  @return std::size_t

include/amqpcpp/linux_tcp/tcpconnection.h

@@ -117,6 +117,27 @@ private:
      */
     virtual void onClosed(Connection *connection) override;
 
+    /**
+     *  Method that is called when the AMQP connection was blocked.
+     *  @param  connection      The connection that was blocked
+     *  @param  reason          Why was the connection blocked
+     */
+    virtual void onBlocked(Connection *connection, const char *reason) override
+    {
+        // pass to user space
+        if (_handler) _handler->onBlocked(this, reason);
+    }
+
+    /**
+     *  Method that is called when the AMQP connection is no longer blocked.
+     *  @param  connection      The connection that is no longer blocked
+     */
+    virtual void onUnblocked(Connection *connection) 
+    { 
+        // pass to user space
+        if (_handler) _handler->onUnblocked(this);
+    }
+
     /**
      *  Method that is called when the tcp connection has been established
      *  @param  state

include/amqpcpp/linux_tcp/tcphandler.h

@@ -191,6 +191,39 @@ public:
         (void) connection;
     }
 
+    /**
+     *  Method that is called when the AMQP connection was blocked.
+     * 
+     *  This method is called, when the server connection gets blocked for the first
+     *  time due to the broker running low on a resource (memory or disk). For
+     *  example, when a RabbitMQ node detects that it is low on RAM, it sends a
+     *  notification to all connected publishing clients supporting this feature.
+     *  If before the connections are unblocked the node also starts running low on
+     *  disk space, another notification will not be sent.
+     *
+     *  @param  connection      The connection that was blocked
+     *  @param  reason          Why was the connection blocked
+     */
+    virtual void onBlocked(TcpConnection *connection, const char *reason)
+    {
+        // make sure compilers dont complain about unused parameters
+        (void) connection;
+    }
+
+    /**
+     *  Method that is called when the AMQP connection is no longer blocked.
+     * 
+     *  This method is called when all resource alarms have cleared and the
+     *  connection is fully unblocked.
+     *
+     *  @param  connection      The connection that is no longer blocked
+     */
+    virtual void onUnblocked(TcpConnection *connection) 
+    { 
+        // make sure compilers dont complain about unused parameters
+        (void) connection;
+    }
+
     /**
      *  Method that is called when the TCP connection is lost or closed. This
      *  is always called if you have also received a call to onConnected().

src/connectionblockframe.h

@@ -0,0 +1,108 @@
+/**
+ *  Class describing a connection blocked frame
+ * 
+ *  This frame is sent by the server to the client, when their connection gets
+ *  blocked for the first time due to the broker running low on a resource
+ *  (memory or disk). For example, when a RabbitMQ node detects that it is low
+ *  on RAM, it sends a notification to all connected publishing clients
+ *  supporting this feature. If before the connections are unblocked the node
+ *  also starts running low on disk space, another notification will not be sent.
+ * 
+ *  @copyright 2023 Copernica BV
+ */
+
+/**
+ *  Set up namespace
+ */
+namespace AMQP {
+
+/**
+ *  Class implementation
+ */
+class ConnectionBlockFrame : public ConnectionFrame
+{
+private:
+    /**
+     *  The reason for blocking
+     *  @var ShortString
+     */
+    ShortString _reason;
+
+protected:
+    /**
+     *  Encode a frame on a string buffer
+     *
+     *  @param  buffer  buffer to write frame to
+     */
+    virtual void fill(OutBuffer& buffer) const override
+    {
+        // call base
+        ConnectionFrame::fill(buffer);
+
+        // encode the field
+        _reason.fill(buffer);
+    }
+
+public:
+    /**
+     *  Construct a connection blocked frame from a received frame
+     *
+     *  @param frame    received frame
+     */
+    ConnectionBlockFrame(ReceivedFrame &frame) :
+        ConnectionFrame(frame),
+        _reason(frame)
+    {}
+
+    /**
+     *  Construct a connection blocked frame
+     *
+     *  @param  reason          the reason for blocking
+     */
+    ConnectionBlockFrame(uint16_t code, std::string reason) :
+        ConnectionFrame((uint32_t)(reason.length() + 1)), // 1 for extra string byte
+        _reason(std::move(reason))
+    {}
+
+    /**
+     *  Destructor
+     */
+    virtual ~ConnectionBlockFrame() {}
+
+    /**
+     *  Method id
+     *  @return uint16_t
+     */
+    virtual uint16_t methodID() const override
+    {
+        return 60;
+    }
+
+    /**
+     *  Get the reason for blocking
+     *  @return string
+     */
+    const std::string& reason() const
+    {
+        return _reason;
+    }
+    
+    /**
+     *  Process the frame
+     *  @param  connection
+     */
+    virtual bool process(ConnectionImpl *connection) override
+    {
+        // report that it is blocked
+        connection->reportBlocked(this->reason().c_str());
+        
+        // done
+        return true;
+    }
+};
+
+/**
+ *  end namespace
+ */
+}
+

src/connectionstartframe.h

@@ -219,6 +219,9 @@ public:
         // queue lives dies, we want to receive a notification that the consumer is no longer alive)
         capabilities["consumer_cancel_notify"] = true;
         
+        // when the rabbitmq server reaches its max capacity, it can send a notification to us, we want them
+        capabilities["connection.blocked"] = true;
+        
         // fill the peer properties
         if (!properties.contains("version")) properties["version"] = "AMQP-CPP " VERSION_NAME;
         if (!properties.contains("copyright")) properties["copyright"] = "Copernica AMQP-CPP library :: Copyright 2015-2023 Copernica BV";

src/connectionunblockframe.h

@@ -0,0 +1,81 @@
+/**
+ *  Class describing a connection unblocked frame
+ * 
+ *  This frame is sent by the server to the client, when all resource alarms
+ *  have cleared and the connection is fully unblocked.
+ * 
+ *  @copyright 2023 Copernica BV
+ */
+
+/**
+ *  Set up namespace
+ */
+namespace AMQP {
+
+/**
+ *  Class implementation
+ */
+class ConnectionUnblockFrame : public ConnectionFrame
+{
+protected:
+    /**
+     *  Encode a frame on a string buffer
+     *
+     *  @param  buffer  buffer to write frame to
+     */
+    virtual void fill(OutBuffer& buffer) const override
+    {
+        // call base
+        ConnectionFrame::fill(buffer);
+    }
+
+public:
+    /**
+     *  Construct a connection unblocked frame from a received frame
+     *
+     *  @param frame    received frame
+     */
+    ConnectionUnblockFrame(ReceivedFrame &frame) :
+        ConnectionFrame(frame)
+    {}
+
+    /**
+     *  Construct a connection unblocked frame
+     */
+    ConnectionUnblockFrame(uint16_t code, std::string reason) :
+        ConnectionFrame(0)
+    {}
+
+    /**
+     *  Destructor
+     */
+    virtual ~ConnectionUnblockFrame() {}
+
+    /**
+     *  Method id
+     *  @return uint16_t
+     */
+    virtual uint16_t methodID() const override
+    {
+        return 61;
+    }
+    
+    /**
+     *  Process the frame
+     *  @param  connection
+     */
+    virtual bool process(ConnectionImpl *connection) override
+    {
+        // report that it is no longer blocked
+        connection->reportUnblocked();
+        
+        // done
+        return true;
+    }
+};
+
+/**
+ *  end namespace
+ */
+}
+

src/receivedframe.cpp

@@ -19,6 +19,8 @@
 #include "connectiontuneframe.h"
 #include "connectioncloseokframe.h"
 #include "connectioncloseframe.h"
+#include "connectionblockframe.h"
+#include "connectionunblockframe.h"
 #include "channelopenframe.h"
 #include "channelopenokframe.h"
 #include "channelflowframe.h"
@@ -195,6 +197,8 @@ bool ReceivedFrame::processConnectionFrame(ConnectionImpl *connection)
         case 41:    return ConnectionOpenOKFrame(*this).process(connection);
         case 50:    return ConnectionCloseFrame(*this).process(connection);
         case 51:    return ConnectionCloseOKFrame(*this).process(connection);
+        case 60:    return ConnectionBlockFrame(*this).process(connection);
+        case 61:    return ConnectionUnblockFrame(*this).process(connection);
     }
 
     // this is a problem