C++ Third-party library RabbitMq example detailed explanation

# When the installation is completed, there is a user guest by default, but the permissions are not enough. You must create an administrator user to log in and publish subscription messages as remote login.#Add usersudo rabbitmqctl add_user root <PASSWORD>
#Set user tagssudo rabbitmqctl set_user_tags root administrator 
#Set user permissionssudo rabbitmqctl set_permissions -p / root "." "." ".*" 
# RabbitMQ comes with a web management interface, execute the following command to enable it, the default port 15672sudo rabbitmq-plugins enable rabbitmq_management 

sudo apt install libev-dev #libev Network Library Componentsgit clone /CopernicaMarketingSoftware/
cd AMQP-CPP/
make
make install

/usr/include/openssl/:147:4: error: #error 
"OPENSSL_API_COMPAT expresses an impossible API compatibility 
level" 
  147 | #  error "OPENSSL_API_COMPAT expresses an impossible API 
compatibility level" 
      |    ^~~~~ 
In file included from /usr/include/openssl/:18, 
                 from linux_tcp/:20, 
                 from linux_tcp/:12: 
/usr/include/openssl/:687:1: error: expected constructor, 
destructor, or type conversion before ‘DEPRECATEDIN_1_1_0' 
  687 | DEPRECATEDIN_1_1_0(int BIO_get_port(const char *str, 
unsigned short *port_ptr))

dpkg -l | grep ssl
sudo dpkg -P --force-all libevent-openssl-2.1-7
sudo dpkg -P --force-all openssl
sudo dpkg -P --force-all libssl-dev
sudo apt --fix-broken install

namespace AMQP 
{ 
    /** 
     *  Generic callbacks that are used by many deferred objects 
     */ 
    using SuccessCallback = std::function<void()>; 
    using ErrorCallback = std::function<void(const char *message)>; 
    using FinalizeCallback = std::function<void()>;
    /** 
     *  Declaring and deleting a queue 
     */ 
    using QueueCallback = std::function<void(const std::string &name, 
                                                   uint32_t messagecount, 
                                                   uint32_t consumercount)>;
    using DeleteCallback = std::function<void(uint32_t deletedmessages)>; 
    using MessageCallback = std::function<void(const Message &message, 
                                                   uint64_t deliveryTag, 
                                                   bool redelivered)>; 
    // When using publisher acknowledgement, AckCallback will be called when the server acknowledgement message has been received and processed.    using AckCallback = std::function<void(uint64_t deliveryTag, bool multiple)>;
    // When using the confirmation package channel, these callbacks will be called when the message is ack/nacked    using PublishAckCallback = std::function<void()>; 
    using PublishNackCallback = std::function<void()>; 
    using PublishLostCallback = std::function<void()>; 
	// Channel class    class Channel 
    { 
        Channel(Connection *connection); 
        bool connected();
        /**
             *Declare the switch
             *If an empty name is provided, the server will assign a name. 
             *The following flags can be used for switches:
             *
             *-durable persistence, the switch is still valid after restart
             *-autodelete After deleting all connected queues, automatically delete the exchange
             *-passive Only passive checks whether the switch exists
             *-internal Create internal exchange
             *
             *@param name The name of the switch
             *@param-type exchange type
                 enum ExchangeType
                 {
                     fanout, broadcast exchange, bound queues can get messages
                     direct, directly exchange, only handing messages to queues with consistent routingkey
                     topic, topic exchange, handing messages to a queue that complies with bindingkey rules
                     headers,
                     consistent_hash,
                     message_deduplication
                 };
             *@param flags switch flag
             *@param argumentsOther parameters
             *
             *This function returns a delay handler.  Callbacks can be installed
             using onSuccess(), onError() and onFinalize() methods.
         */ 
        Deferred &declareExchange(const std::string_view &name,
                                  ExchangeType type,
                                  int flags,
                                  const Table &arguments);
        /**
             *Declare queue
             *If a name is not provided, the server will be assigned a name. 
             *flags can be a combination of the following values:
             *
             *-durable persistent queue remains valid after proxy restart
             *-autodelete Automatically delete the queue when all connected consumers leave
             *-passive Only passive checks whether the queue exists
             *-exclusive queue exists only on this connection and is automatically deleted when the connection is disconnected
             *
             *@param name The name of the queue
             *@param flags combination
             *@param arguments Optional parameters
             *
             *This function returns a delay handler.  Callbacks can be installed
             *Use onSuccess(), onError() and onFinalize() methods. 
             *
             Deferred &onError(const char *message)
             *
             *The onSuccess() callback that can be installed should have the following signature:
             void myCallback(const std::string &name,
                 uint32_t messageCount,
                 uint32_t consumerCount);
             For example: 
             ("myqueue").onSuccess(
                 [](const std::string &name,
                     uint32_t messageCount,
                     uint32_t consumerCount) {
                        std::cout << "Queue '" << name << "' ";
                        std::cout << "has been declared with ";
                        std::cout << messageCount;
                        std::cout << " messages and ";
                        std::cout << consumerCount;
                        std::cout << " consumers" << std::endl;
          * });
         */ 
        DeferredQueue &amp;declareQueue(const std::string_view &amp;name,
                                    int flags,
                                    const Table &amp;arguments);
        /**
             *Bind the queue to the switch
             *
             *@param exchange source switch
             *@param queue Target queue
             *@param routingkey
             *@param arguments Other binding parameters
             *
             *This function returns a delay handler.  Callbacks can be installed
             *Use onSuccess(), onError() and onFinalize() methods. 
         */ 
        Deferred &amp;bindQueue(const std::string_view &amp;exchange,
                            const std::string_view &amp;queue,
                            const std::string_view &amp;routingkey,
                            const Table &amp;arguments);
        /**
             * Publish messages to exchange
             *You must provide the name and routing key of the switch. 
             RabbitMQ will then attempt to send the message to one or more queues. 
             Using the optional flags parameter, you can specify what should happen if the message cannot be routed to the queue.
             By default, unchangeable messages are silently discarded. 
             *
             *If the 'mandatory' or 'immediate' flags are set,
             Then unprocessed messages will be returned to the application. 
             Before you start publishing, make sure you have called the recall()- method,
             And set up all appropriate handlers to handle these returned messages. 
             *
             *The following flags can be provided:
             *
             *-mandatory If set, the server will return a message not sent to the queue
             *-immediate If set, the server will return a message that cannot be forwarded to the consumer immediately. 
             *@param exchange to be published to
             *@param routingkey routing key
             *@param envelope full envelope to send
             *@param message to send
             *@param size message size
             *@param flags optional logo
         */ 
        bool publish(const std::string_view &amp;exchange,
                     const std::string_view &amp;routingKey,
                     const std::string &amp;message,
                     int flags = 0);
        /**
             *Tell the RabbitMQ server that is ready to use messages - that is, subscribe to queue messages
             *
             *After calling this method, RabbitMQ starts delivering messages to the client application. 
             consumer tag is a string identifier.
             If you want to stop it later via channel::cancel() call,
             You can use it to identify the user. 
             *If you do not specify a user tag, the server will assign one to you. 
             *
             *Support the following flags:
             *
             *-nolocal If set, messages posted on this channel will not be consumed at the same time
             *-noack If set, no confirmation of consumed messages is not necessary
             *-exclusive request exclusive access, only this user can access the queue
             *
             *@param queue The queue you want to use
             *@param tag The consumer tag associated with this consumption operation
             *@param flags Other tags
             *@param argumentsOther parameters
             *
             *This function returns a delay handler. 
             You can use the onSuccess(), onError() and onFinalize() methods to install the callback
             The onSuccess() callback that can be installed should have the following format:
                 void myCallback(const std::string_view&tag);
             Sample:
             ("myqueue").onSuccess(
                 [](const std::string_view& tag) {
                     std::cout << "Started consume under tag ";
                     std::cout << tag << std::endl;
             });
         */ 
        DeferredConsumer &amp;consume(const std::string_view &amp;queue,
                                  const std::string_view &amp;tag,
                                  int flags,
                                  const Table &amp;arguments);
        /**
             *Confirm the received message
             *
             *The consumer client confirms and answers the received message
             *
             *When a message is received in the DeferredConsumer::onReceived() method,
             The message must be confirmed.
             So that RabbitMQ will remove it from the queue (unless consumed with the noack option)
             *
             *Support the following flags:
             *
             *-Multiple acknowledgement multiple messages: All previously unconfirmed messages will also be acknowledged
             *
             *@param deliveryTag The only delivery tag for the message
             *@param flags Optional logo
             *@return bool
         */ 
        bool ack(uint64_t deliveryTag, int flags=0);
    };
    class DeferredConsumer 
    { 
        /*
             Register a callback function that is called when the consumer starts
             void onSuccess(const std::string &consumertag)
         */ 
        DeferredConsumer &amp;onSuccess(const ConsumeCallback&amp; callback);
        /*
             Register a callback function, used to be called when a complete message is received
             void MessageCallback(const AMQP::Message &message,
                 uint64_t deliveryTag, bool redelivered)
         */ 
        DeferredConsumer &amp;onReceived(const MessageCallback&amp; callback);
        /* Alias for onReceived() */ 
        DeferredConsumer &amp;onMessage(const MessageCallback&amp; callback);
        /*
             Register the function to be called when the server cancels the consumer
             void CancelCallback(const std::string &tag)
         */ 
        DeferredConsumer &amp;onCancelled(const CancelCallback&amp; callback);
    };
    class Message : public Envelope
    { 
        const std::string &amp;exchange();
        const std::string &amp;routingkey();
    };
    class Envelope : public MetaData
    { 
        const char *body();  // Get the message body        uint64_t bodySize(); // Get the message body size    };
}

typedef struct ev_async 
{ 
    EV_WATCHER (ev_async);
    EV_ATOMIC_T sent; /* private */ 
}ev_async; 
//break type 
enum 
{ 
    EVBREAK_CANCEL = 0, /* undo unloop */ 
    EVBREAK_ONE    = 1, /* unloop once */ 
    EVBREAK_ALL    = 2  /* unloop all loops */ 
}; 
// Instantiate and get the IO event monitoring interface handlestruct ev_loop *ev_default_loop (unsigned int flags EV_CPP (= 0));
# define EV_DEFAULT  ev_default_loop (0) 
// Start running IO event monitoring, this is a blocking interfaceint  ev_run (struct ev_loop *loop);
/* break out of the loop */
// End IO monitoring// If you perform ev_run() on the main thread, you can call it directly,// If ev_run() is performed in other threads, it needs to be done through asynchronous notificationvoid ev_break (struct ev_loop *loop, int32_t break_type) ;  
void (*callback)(struct ev_loop *loop, ev_async *watcher, int32_t revents);
// Initialize the asynchronous event structure and set the callback functionvoid ev_async_init(ev_async *w, callback cb);
// Start asynchronous task processing in event monitoring loopvoid ev_async_start(struct ev_loop *loop, ev_async *w); 
// Send the current asynchronous event to execute in the asynchronous threadvoid ev_async_send(struct ev_loop *loop, ev_async *w);

#include <>
#include <>
#include <amqpcpp/>
#include <openssl/>
#include <openssl/>
int main()
{
    // 1. Instantiate the IO event monitoring handle of the underlying network communication framework    auto *loop = EV_DEFAULT;
    // 2. Instantiate the libEvHandler handle -> Associate the AMQP framework with event monitoring    AMQP::LibEvHandler handler(loop);
    // 3. Instantiate the connection object    AMQP::Address address("amqp://root:[email protected]:5672/");
    AMQP::TcpConnection connection(&handler, address);
    // 4. Instantiate the channel object    AMQP::TcpChannel channel(&connection);
    // 5. Declare the switch    ("test-exchange", AMQP::ExchangeType::direct)
        .onError([](const char *message)
                { std::cout << "Declare the switch failed: " << message << std::endl; })
        .onSuccess([]()
                { std::cout << "test-exchange switch creation successfully" << std::endl; });
    // 6. Declare the queue    ("test-queue")
        .onError([](const char *message)
                 { std::cout << "Declaration queue failed: " << message << std::endl; })
        .onSuccess([]()
                 { std::cout << "test-queue queue creation successfully" << std::endl; });
    // 7. Bind the switch and queue    ("test-exchange", "test-queue", "test-queue-key")
        .onError([](const char *message)
                 { std::cout << "test-exchange - test-queue binding failed: " \
                 << message << std::endl; })
        .onSuccess([]()
                 { std::cout << "test-exchange - test-queue binding succeeded" 
                 << std::endl; });
    // 8. Publish a message to the switch    for (int i = 0; i < 5; ++i)
    {
        std::string msg = "Hello SnowK-" + std::to_string(i);
        if(("test-exchange", "test-queue-key", msg) == false)
        {
            std::cout << "Publish failed" << std::endl;
        }
    }
    // 9. Start the underlying network communication framework -> Turn on IO    ev_run(loop, 0);
    return 0;
}

#include <>
#include <>
#include <amqpcpp/>
#include <openssl/>
#include <openssl/>
void MessageCB(AMQP::TcpChannel* channel, const AMQP::Message& message, 
               uint64_t deliveryTag, bool redelivered)
{
    std::string msg;
    ((), ());
    // This cannot be used, AMQP::Message does not store '\0' after '    // std::cout << message << std::endl 
    std::cout << msg << std::endl;
    channel->ack(deliveryTag);
}
int main()
{
    // 1. Instantiate the IO event monitoring handle of the underlying network communication framework    auto *loop = EV_DEFAULT;
    // 2. Instantiate the libEvHandler handle -> Associate the AMQP framework with event monitoring    AMQP::LibEvHandler handler(loop);
    // 3. Instantiate the connection object    AMQP::Address address("amqp://root:[email protected]:5672/");
    AMQP::TcpConnection connection(&handler, address);
    // 4. Instantiate the channel object    AMQP::TcpChannel channel(&connection);
    // 5. Declare the switch    ("test-exchange", AMQP::ExchangeType::direct)
        .onError([](const char *message)
                 { std::cout << "Declare the switch failed: " << message << std::endl; })
        .onSuccess([]()
                 { std::cout << "test-exchange switch creation successfully" << std::endl; });
    // 6. Declare the queue    ("test-queue")
        .onError([](const char *message)
                 { std::cout << "Declaration queue failed: " << message << std::endl; })
        .onSuccess([]()
                 { std::cout << "test-queue queue creation successfully" << std::endl; });
    // 7. Bind the switch and queue    ("test-exchange", "test-queue", "test-queue-key")
        .onError([](const char *message)
                 { std::cout << "test-exchange - test-queue binding failed: " \
				 << message << std::endl; })
        .onSuccess([]()
                 { std::cout << "test-exchange - test-queue binding succeeded"; });
    // 8. Subscribe to message confrontation -> Set message processing callback function    auto callback = std::bind(MessageCB, &channel, std::placeholders::_1, 
                              std::placeholders::_2, std::placeholders::_3);
    ("test-queue", "consume-tag")
        .onReceived(callback)
        .onError([](const char *message)
        { 
            std::cout << "Subscribe to test-queue queue message failed: " << message << std::endl;
            exit(0); 
        });
    // 9. Start the underlying network communication framework -> Turn on IO    ev_run(loop, 0);
    return 0;
}

all: publish consume
publish: 
	g++ -o $@ $^ -lamqpcpp -lev -std=c++17
consume: 
	g++ -o $@ $^ -lamqpcpp -lev -std=c++17
.PHONY:clean
clean:
	rm publish consume