c++ - AMQP-CPP - RabbitMQ 事件循环不工作

Question

我没有从 https://github.com/CopernicaMarketingSoftware/AMQP-CPP 获得带有 RabbitMQ CPP (CopernicaMarketingSoftware/AMQP-CPP) 库的事件循环正常运行。我已经设置了一个 rabbitmq 服务器，它工作正常。

我的代码是这样的:

#include <iostream>
#include <sys/select.h>
#include <cop_amqpcpp.h>
#include <cop_amqpcpp/linux_tcp.h>


using namespace std;

fd_set rfds;

class MyTcpHandler : public AMQP::TcpHandler
{
private:
    /**
    * Method that is called when the connection succeeded
    * @param socket Pointer to the socket
    */
    virtual void onConnected(AMQP::TcpConnection* connection)
    {
        std::cout << "connected" << std::endl;
    }

    /**
     *  When the connection ends up in an error state this method is called.
     *  This happens when data comes in that does not match the AMQP protocol
     *
     *  After this method is called, the connection no longer is in a valid
     *  state and can be used. In normal circumstances this method is not called.
     *
     *  @param  connection      The connection that entered the error state
     *  @param  message         Error message
     */
    virtual void onError(AMQP::TcpConnection* connection, const char* message)
    {
        // report error
        std::cout << "AMQP TCPConnection error: " << message << std::endl;
    }

    /**
     *  Method that is called when the connection was closed.
     *  @param  connection      The connection that was closed and that is now unusable
     */
    virtual void onClosed(AMQP::TcpConnection* connection)
    {
        std::cout << "closed" << std::endl;
    }


    /**
     *  Method that is called by AMQP-CPP to register a filedescriptor for readability or writability
     *  @param  connection  The TCP connection object that is reporting
     *  @param  fd          The filedescriptor to be monitored
     *  @param  flags       Should the object be monitored for readability or writability?
     */
    virtual void monitor(AMQP::TcpConnection* connection, int fd, int flags)
    {
        // we did not yet have this watcher - but that is ok if no filedescriptor was registered
        if (flags == 0) return;
        cout << "Fd " << fd << " Flags " << flags << endl;
        if (flags & AMQP::readable)
        {
            FD_SET(fd, &rfds);
            m_fd = fd;
            m_flags = flags;
        }
    }

public:
    int m_fd = -1;
    int m_flags = 0;
};


int main(int argc, char* argv[])
{
    MyTcpHandler myHandler;
    int maxfd = 1;
    int result = 0;

    struct timeval tv
    {
        1, 0
    };


    // address of the server
    AMQP::Address address(AMQP::Address("amqp://test:test@localhost/"));

    // create a AMQP connection object
    AMQP::TcpConnection connection(&myHandler, address);

    // and create a channel
    AMQP::TcpChannel channel(&connection);

    channel.onError([](const char* message)
    {
        cout << "channel error: " << message << endl;
    });
    channel.onReady([]()
    {
        cout << "channel ready " << endl;
    });

    // use the channel object to call the AMQP method you like

    channel.declareExchange("my-exchange", AMQP::fanout);
    channel.declareQueue("my-queue");
    channel.bindQueue("my-exchange", "my-queue", "my-routing-key");

    do
    {
        FD_ZERO(&rfds);
        FD_SET(0, &rfds);
        cout << myHandler.m_fd << endl;
        if (myHandler.m_fd != -1)
        {
            maxfd = myHandler.m_fd + 1;
        }


        result = select(maxfd, &rfds, NULL, NULL, &tv);
        if ((result == -1) && errno == EINTR)
        {
            cout << "Error in socket" << endl;
        }
        else if (result > 0)
        {
            if (myHandler.m_flags & AMQP::readable)

                cout << "Got something" << endl;
            if (FD_ISSET(myHandler.m_fd, &rfds))
            {
                connection.process(maxfd, myHandler.m_flags);
            }
        }
    }
    while (1);
}

不幸的是，select 只返回 0，而且我从不跳转到 TCP 处理程序中的 onConnected 回调。我只是跳入监视器回调一次。奇怪的是，monitor 方法中的文件描述符是数字 3，但是如果我查看 /proc 文件夹中的文件描述符，我看到套接字的 fd 数字是 5。

这是我在/proc 文件夹中的打印输出:

lrwx------ 1 pi pi 64 May 29 06:29 0 -> /dev/pts/1
lrwx------ 1 pi pi 64 May 29 06:29 1 -> /dev/pts/1
lrwx------ 1 pi pi 64 May 29 06:29 2 -> /dev/pts/1
lr-x------ 1 pi pi 64 May 29 06:29 3 -> pipe:[680897]
l-wx------ 1 pi pi 64 May 29 06:29 4 -> pipe:[680897]
lrwx------ 1 pi pi 64 May 29 06:30 5 -> socket:[678884]

我让它与示例中的 libuv 和 libevent 等现成的事件循环一起工作，但我想了解没有它它是如何工作的。

我如何让它工作？

Answer 1

下面一行

FD_SET(myHandler.m_fd, &rfds);

需要添加到事件循环中。