C++/QT 使用RabbitMQ

#include <QTimer> #include <QTextStream> #include <QStringList> #include <QSslSocket> #include <QtEndian> #include "qamqpglobal.h" #include "qamqpexchange.h" #include "qamqpexchange_p.h" #include "qamqpqueue.h" #include "qamqpqueue_p.h" #include "qamqpauthenticator.h" #include "qamqptable.h" #include "qamqpclient_p.h" #include "qamqpclient.h" QAmqpClientPrivate::QAmqpClientPrivate(QAmqpClient *q) : port(AMQP_PORT), host(AMQP_HOST), virtualHost(AMQP_VHOST), autoReconnect(false), reconnectFixedTimeout(false), timeout(0), connecting(false), useSsl(false), socket(0), closed(false), connected(false), channelMax(0), heartbeatDelay(0), frameMax(AMQP_FRAME_MAX), error(QAMQP::NoError), q_ptr(q) { qRegisterMetaType<QAmqpMessage::PropertyHash>(); } QAmqpClientPrivate::~QAmqpClientPrivate() { } void QAmqpClientPrivate::init() { Q_Q(QAmqpClient); initSocket(); heartbeatTimer = new QTimer(q); QObject::connect(heartbeatTimer, SIGNAL(timeout()), q, SLOT(_q_heartbeat())); reconnectTimer = new QTimer(q); reconnectTimer->setSingleShot(true); QObject::connect(reconnectTimer, SIGNAL(timeout()), q, SLOT(_q_connect())); authenticator = QSharedPointer<QAmqpAuthenticator>( new QAmqpPlainAuthenticator(QString::fromLatin1(AMQP_LOGIN), QString::fromLatin1(AMQP_PSWD))); } void QAmqpClientPrivate::initSocket() { Q_Q(QAmqpClient); socket = new QSslSocket(q); socket->setSocketOption(QAbstractSocket::LowDelayOption, 1); socket->setSocketOption(QAbstractSocket::KeepAliveOption, 1); QObject::connect(socket, SIGNAL(connected()), q, SLOT(_q_socketConnected())); QObject::connect(socket, SIGNAL(disconnected()), q, SLOT(_q_socketDisconnected())); QObject::connect(socket, SIGNAL(readyRead()), q, SLOT(_q_readyRead())); QObject::connect(socket, SIGNAL(error(QAbstractSocket::SocketError)), q, SLOT(_q_socketError(QAbstractSocket::SocketError))); QObject::connect(socket, SIGNAL(error(QAbstractSocket::SocketError)), q, SIGNAL(socketError(QAbstractSocket::SocketError))); QObject::connect(socket, SIGNAL(stateChanged(QAbstractSocket::SocketState)), q, SIGNAL(socketStateChanged(QAbstractSocket::SocketState))); QObject::connect(socket, SIGNAL(sslErrors(QList<QSslError>)), q, SIGNAL(sslErrors(QList<QSslError>))); } void QAmqpClientPrivate::resetChannelState() { foreach (QString exchangeName, exchanges.channels()) { QAmqpExchange *exchange = qobject_cast<QAmqpExchange*>(exchanges.get(exchangeName)); if (exchange) exchange->d_ptr->resetInternalState(); } foreach (QString queueName, queues.channels()) { QAmqpQueue *queue = qobject_cast<QAmqpQueue*>(queues.get(queueName)); if (queue) queue->d_ptr->resetInternalState(); } } void QAmqpClientPrivate::setUsername(const QString &username) { QAmqpAuthenticator *auth = authenticator.data(); if (auth && auth->type() == QLatin1String("AMQPLAIN")) { QAmqpPlainAuthenticator *a = static_cast<QAmqpPlainAuthenticator*>(auth); a->setLogin(username); } } void QAmqpClientPrivate::setPassword(const QString &password) { QAmqpAuthenticator *auth = authenticator.data(); if (auth && auth->type() == QLatin1String("AMQPLAIN")) { QAmqpPlainAuthenticator *a = static_cast<QAmqpPlainAuthenticator*>(auth); a->setPassword(password); } } void QAmqpClientPrivate::parseConnectionString(const QString &uri) { #if QT_VERSION > 0x040801 QUrl connectionString = QUrl::fromUserInput(uri); #else QUrl connectionString(uri, QUrl::TolerantMode); #endif if (connectionString.scheme() != AMQP_SCHEME && connectionString.scheme() != AMQP_SSL_SCHEME) { qAmqpDebug() << Q_FUNC_INFO << "invalid scheme: " << connectionString.scheme(); return; } useSsl = (connectionString.scheme() == AMQP_SSL_SCHEME); port = connectionString.port((useSsl ? AMQP_SSL_PORT : AMQP_PORT)); host = connectionString.host(); QString vhost = connectionString.path(); if (vhost.startsWith("/") && vhost.size() > 1) vhost = vhost.mid(1); #if QT_VERSION <= 0x050200 virtualHost = QUrl::fromPercentEncoding(vhost.toUtf8()); setPassword(QUrl::fromPercentEncoding(connectionString.password().toUtf8())); setUsername(QUrl::fromPercentEncoding(connectionString.userName().toUtf8())); #else virtualHost = vhost; setPassword(connectionString.password()); setUsername(connectionString.userName()); #endif } void QAmqpClientPrivate::_q_connect() { if (reconnectTimer) reconnectTimer->stop(); if (socket->state() != QAbstractSocket::UnconnectedState) { qAmqpDebug() << Q_FUNC_INFO << "socket already connected, disconnecting.."; _q_disconnect(); // We need to explicitly close connection here because either way it will not be closed until we receive closeOk closeConnection(); } qAmqpDebug() << "connecting to host: " << host << ", port: " << port; if (useSsl) socket->connectToHostEncrypted(host, port); else socket->connectToHost(host, port); } void QAmqpClientPrivate::_q_disconnect() { if (reconnectTimer) reconnectTimer->stop(); if (socket->state() == QAbstractSocket::UnconnectedState) { qAmqpDebug() << Q_FUNC_INFO << "already disconnected"; return; } buffer.clear(); close(200, "client disconnect"); } // private slots void QAmqpClientPrivate::_q_socketConnected() { if (reconnectTimer) reconnectTimer->stop(); if(reconnectFixedTimeout == false) timeout = 0; char header[8] = {'A', 'M', 'Q', 'P', 0, 0, 9, 1}; socket->write(header, 8); } void QAmqpClientPrivate::_q_socketDisconnected() { Q_Q(QAmqpClient); buffer.clear(); resetChannelState(); if (connected) connected = false; Q_EMIT q->disconnected(); } void QAmqpClientPrivate::_q_heartbeat() { QAmqpHeartbeatFrame frame; sendFrame(frame); } void QAmqpClientPrivate::_q_socketError(QAbstractSocket::SocketError error) { if(reconnectFixedTimeout == false) { if (timeout <= 0) { timeout = 1000; } else { if (timeout < 120000) timeout *= 5; } } switch (error) { case QAbstractSocket::ConnectionRefusedError: case QAbstractSocket::RemoteHostClosedError: case QAbstractSocket::SocketTimeoutError: case QAbstractSocket::NetworkError: case QAbstractSocket::ProxyConnectionClosedError: case QAbstractSocket::ProxyConnectionRefusedError: case QAbstractSocket::ProxyConnectionTimeoutError: default: qAmqpDebug() << "socket error: " << socket->errorString(); break; } // per spec, on any error we need to close the socket immediately // and send no more data. only try to send the close message if we // are actively connected if (socket->state() == QAbstractSocket::ConnectedState || socket->state() == QAbstractSocket::ConnectingState) { socket->abort(); } errorString = socket->errorString(); if (autoReconnect && reconnectTimer) { qAmqpDebug() << "trying to reconnect after: " << timeout << "ms"; reconnectTimer->start(timeout); } } void QAmqpClientPrivate::_q_readyRead() { Q_Q(QAmqpClient); while (socket->bytesAvailable() >= QAmqpFrame::HEADER_SIZE) { unsigned char headerData[QAmqpFrame::HEADER_SIZE]; socket->peek((char*)headerData, QAmqpFrame::HEADER_SIZE); const quint32 payloadSiz