AMQP-CPP-RabbitMQ事件循环不起作用

本文介绍了AMQP-CPP-RabbitMQ事件循环不起作用的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

我没有从 https://github.com/CopernicaMarketingSoftware/AMQP-CPP 正常运行.我已经设置了rabbitmq服务器，并且工作正常.

I dont get the Event Loop with the RabbitMQ CPP (CopernicaMarketingSoftware/AMQP-CPP) Library from https://github.com/CopernicaMarketingSoftware/AMQP-CPP running properly. I have set up a rabbitmq server, and it works fine.

我的代码如下:

#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 回调.我只在监视器回调中跳过一次.奇怪的是，监视方法中的文件描述符为3，但是如果查看/proc 文件夹中的文件描述符，我会看到套接字的fd为5.

Unfortunately select only ever returns 0, and I never jump to the onConnected callback in the TCP Handler. I just jump in the monitor callback once. Strangely enough, the file descriptor in the monitor method is number 3, but if I look at the file descriptors in the /proc folder, I see that the socket has the fd number 5.

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

Thats my printout of in the /proc folder:

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之类的就绪事件循环一起使用，但是我想了解没有它的工作循环.

I made it work with ready event loops like libuv and libevent like among the examples, but I would like to understand how it works without this.

我如何使它工作?

CopernicaMarketingSoftware

问题描述

推荐答案