AMQP-CPP - не работает цикл событий RabbitMQ - PullRequest
Новая
       15

AMQP-CPP - не работает цикл событий RabbitMQ

0 голосов
/ 30 мая 2018

Я не получаю цикл событий с библиотекой RabbitMQ CPP (CopernicaMarketingSoftware / AMQP-CPP) из https://github.com/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, и я никогдаперейти к обратному вызову onConnected в обработчике TCP.Я просто прыгаю в обратный вызов монитора один раз.Как ни странно, файловый дескриптор в методе монитора имеет номер 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, как среди примеров, но я хотел бы понять, как это работает без этого.

Как мне заставить его работать?

1 Ответ

0 голосов
/ 30 мая 2018

Следующая строка 

FD_SET(myHandler.m_fd, &rfds);

должна быть добавлена ​​в цикл обработки событий.

...