jessequ
/
qamqp
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

use Q_Q/Q_D from Qt rather than P_Q/P_D, style cleanups for Qt coding style 

delete exception-related code, as Qt does not officially support them anymore
This commit is contained in:
Matt Broadstone 2014-05-28 12:25:28 -04:00
parent 387af4038e
commit c069b063d3
25 changed files with 4134 additions and 4301 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

376
src/qamqp/amqp.cpp
View File

@ -7,6 +7,7 @@
#include "amqp_exchange_p.h" #include "amqp_exchange_p.h"
#include "amqp_queue.h" #include "amqp_queue.h"
#include "amqp_queue_p.h" #include "amqp_queue_p.h"
#include "amqp_connection_p.h"
#include "amqp_authenticator.h" #include "amqp_authenticator.h"
using namespace QAMQP; using namespace QAMQP;
@ -15,372 +16,365 @@ namespace QAMQP
{ {
struct ClientExceptionCleaner struct ClientExceptionCleaner
{ {
/* this cleans up when the constructor throws an exception */ /* this cleans up when the constructor throws an exception */
static inline void cleanup(Client *that, ClientPrivate *d) static inline void cleanup(Client *that, ClientPrivate *d)
{ {
#ifdef QT_NO_EXCEPTIONS #ifdef QT_NO_EXCEPTIONS
Q_UNUSED(that); Q_UNUSED(that);
Q_UNUSED(d); Q_UNUSED(d);
#else #else
Q_UNUSED(that); Q_UNUSED(that);
Q_UNUSED(d); Q_UNUSED(d);
#endif #endif
} }
}; };
} }
////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////
ClientPrivate::ClientPrivate( Client * q ) : ClientPrivate::ClientPrivate(Client * q)
port(AMQPPORT) : port(AMQPPORT),
, host(QString::fromLatin1(AMQPHOST)) host(QString::fromLatin1(AMQPHOST)),
, virtualHost(QString::fromLatin1(AMQPVHOST)) virtualHost(QString::fromLatin1(AMQPVHOST)),
, pq_ptr(q) q_ptr(q)
{ {
} }
ClientPrivate::~ClientPrivate() ClientPrivate::~ClientPrivate()
{ {
} }
void ClientPrivate::init(QObject * parent) void ClientPrivate::init(QObject *parent)
{ {
pq_func()->setParent(parent); Q_Q(QAMQP::Client);
if(!network_) q->setParent(parent);
{ if (!network_)
network_ = new QAMQP::Network(pq_func()); network_ = new QAMQP::Network(q);
}
if(!connection_) if (!connection_)
{ connection_ = new QAMQP::Connection(q);
connection_ = new QAMQP::Connection(pq_func()); network_->setMethodHandlerConnection(connection_);
}
network_->setMethodHandlerConnection(connection_); setAuth(new AMQPlainAuthenticator(QString::fromLatin1(AMQPLOGIN), QString::fromLatin1(AMQPPSWD)));
setAuth(new AMQPlainAuthenticator(QString::fromLatin1(AMQPLOGIN), QString::fromLatin1(AMQPPSWD))); QObject::connect(connection_, SIGNAL(connected()), q, SIGNAL(connected()));
QObject::connect(connection_, SIGNAL(disconnected()), q, SIGNAL(disconnected()));
QObject::connect(connection_, SIGNAL(connected()), pq_func(), SIGNAL(connected()));
QObject::connect(connection_, SIGNAL(disconnected()), pq_func(), SIGNAL(disconnected()));
} }
void ClientPrivate::init(QObject * parent, const QUrl & con) void ClientPrivate::init(QObject *parent, const QUrl &connectionString)
{ {
init(parent); init(parent);
parseCnnString(con); parseConnectionString(connectionString);
ClientPrivate::connect(); connect();
} }
void ClientPrivate::setAuth(Authenticator *auth)
void ClientPrivate::setAuth( Authenticator* auth )
{ {
auth_ = QSharedPointer<Authenticator>(auth); auth_ = QSharedPointer<Authenticator>(auth);
} }
void ClientPrivate::printConnect() const void ClientPrivate::printConnect() const
{ {
QTextStream stream(stdout); QTextStream stream(stdout);
stream << "port = " << port << endl; stream << "port = " << port << endl;
stream << "host = " << host << endl; stream << "host = " << host << endl;
stream << "vhost = " << virtualHost << endl; stream << "vhost = " << virtualHost << endl;
if(auth_ && auth_->type() == "AMQPLAIN")
{
QSharedPointer<AMQPlainAuthenticator> a = auth_.staticCast<AMQPlainAuthenticator>();
stream << "user = " << a->login() << endl;
stream << "passw = " << a->password() << endl;
}
if (auth_ && auth_->type() == QLatin1String("AMQPLAIN")) {
QSharedPointer<AMQPlainAuthenticator> a = auth_.staticCast<AMQPlainAuthenticator>();
stream << "user = " << a->login() << endl;
stream << "passw = " << a->password() << endl;
}
} }
void ClientPrivate::connect() void ClientPrivate::connect()
{ {
ClientPrivate::sockConnect(); sockConnect();
ClientPrivate::login(); login();
} }
void ClientPrivate::parseCnnString( const QUrl & con ) void ClientPrivate::parseConnectionString(const QUrl &connectionString)
{ {
P_Q(QAMQP::Client); Q_Q(QAMQP::Client);
if(con.scheme() == AMQPSCHEME || con.scheme() == AMQPSSCHEME ) if (connectionString.scheme() != AMQPSCHEME &&
{ connectionString.scheme() != AMQPSSCHEME) {
q->setSsl(con.scheme() == AMQPSSCHEME); qDebug() << Q_FUNC_INFO << "invalid scheme: " << connectionString.scheme();
q->setPassword(con.password()); return;
q->setUser(con.userName()); }
q->setPort(con.port(AMQPPORT));
q->setHost(con.host()); q->setSsl(connectionString.scheme() == AMQPSSCHEME);
q->setVirtualHost(con.path()); q->setPassword(connectionString.password());
} q->setUser(connectionString.userName());
q->setPort(connectionString.port(AMQPPORT));
q->setHost(connectionString.host());
q->setVirtualHost(connectionString.path());
} }
void ClientPrivate::sockConnect() void ClientPrivate::sockConnect()
{ {
if(network_->state() != QAbstractSocket::UnconnectedState ) if (network_->state() != QAbstractSocket::UnconnectedState)
{ disconnect();
disconnect(); network_->connectTo(host, port);
}
network_->connectTo(host, port);
} }
void ClientPrivate::login() void ClientPrivate::login()
{ {
} }
Exchange * ClientPrivate::createExchange(int channelNumber, const QString &name ) Exchange *ClientPrivate::createExchange(int channelNumber, const QString &name)
{ {
Exchange * exchange_ = new Exchange(channelNumber, pq_func()); Q_Q(QAMQP::Client);
Exchange * exchange_ = new Exchange(channelNumber, q);
network_->addMethodHandlerForChannel(exchange_->channelNumber(), exchange_); network_->addMethodHandlerForChannel(exchange_->channelNumber(), exchange_);
QObject::connect(connection_, SIGNAL(connected()), exchange_, SLOT(_q_open())); QObject::connect(connection_, SIGNAL(connected()), exchange_, SLOT(_q_open()));
exchange_->pd_func()->open(); exchange_->d_func()->open();
QObject::connect(pq_func(), SIGNAL(disconnected()), exchange_, SLOT(_q_disconnected())); QObject::connect(q, SIGNAL(disconnected()), exchange_, SLOT(_q_disconnected()));
exchange_->setName(name); exchange_->setName(name);
return exchange_; return exchange_;
} }
Queue * ClientPrivate::createQueue(int channelNumber, const QString &name ) Queue *ClientPrivate::createQueue(int channelNumber, const QString &name )
{ {
Queue * queue_ = new Queue(channelNumber, pq_func()); Q_Q(QAMQP::Client);
Queue *queue_ = new Queue(channelNumber, q);
network_->addMethodHandlerForChannel(queue_->channelNumber(), queue_); network_->addMethodHandlerForChannel(queue_->channelNumber(), queue_);
network_->addContentHandlerForChannel(queue_->channelNumber(), queue_); network_->addContentHandlerForChannel(queue_->channelNumber(), queue_);
network_->addContentBodyHandlerForChannel(queue_->channelNumber(), queue_); network_->addContentBodyHandlerForChannel(queue_->channelNumber(), queue_);
QObject::connect(connection_, SIGNAL(connected()), queue_, SLOT(_q_open())); QObject::connect(connection_, SIGNAL(connected()), queue_, SLOT(_q_open()));
queue_->pd_func()->open(); queue_->d_func()->open();
QObject::connect(pq_func(), SIGNAL(disconnected()), queue_, SLOT(_q_disconnected())); QObject::connect(q, SIGNAL(disconnected()), queue_, SLOT(_q_disconnected()));
queue_->setName(name); queue_->setName(name);
return queue_; return queue_;
} }
void ClientPrivate::disconnect() void ClientPrivate::disconnect()
{ {
P_Q(QAMQP::Client); Q_Q(QAMQP::Client);
Q_UNUSED(q); if (network_->state() != QAbstractSocket::UnconnectedState) {
if(network_->state() != QAbstractSocket::UnconnectedState) network_->disconnect();
{ connection_->d_func()->connected = false;
network_->QAMQP::Network::disconnect(); Q_EMIT q->disconnected();
connection_->pd_func()->connected = false; }
emit pq_func()->disconnected();
}
} }
////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////
QAMQP::Client::Client(QObject *parent)
QAMQP::Client::Client( QObject * parent /*= 0*/ ) : QObject(parent),
: pd_ptr(new ClientPrivate(this)) d_ptr(new ClientPrivate(this))
{ {
QT_TRY { d_ptr->init(parent);
pd_func()->init(parent);
} QT_CATCH(...) {
ClientExceptionCleaner::cleanup(this, pd_func());
QT_RETHROW;
}
} }
QAMQP::Client::Client( const QUrl & connectionString, QObject * parent /*= 0*/ ) QAMQP::Client::Client(const QUrl & connectionString, QObject * parent)
: pd_ptr(new ClientPrivate(this)) : d_ptr(new ClientPrivate(this))
{ {
QT_TRY { d_ptr->init(parent, connectionString);
pd_func()->init(parent, connectionString);
} QT_CATCH(...) {
ClientExceptionCleaner::cleanup(this, pd_func());
QT_RETHROW;
}
} }
QAMQP::Client::~Client() QAMQP::Client::~Client()
{ {
QT_TRY {
QEvent e(QEvent::Destroy);
QCoreApplication::sendEvent(this, &e);
} QT_CATCH(const std::exception&) {
// if this fails we can't do anything about it but at least we are not allowed to throw.
}
} }
quint32 QAMQP::Client::port() const quint16 QAMQP::Client::port() const
{ {
return pd_func()->port; Q_D(const QAMQP::Client);
return d->port;
} }
void QAMQP::Client::setPort( quint32 port ) void QAMQP::Client::setPort(quint16 port)
{ {
pd_func()->port = port; Q_D(QAMQP::Client);
d->port = port;
} }
QString QAMQP::Client::host() const QString QAMQP::Client::host() const
{ {
return pd_func()->host; Q_D(const QAMQP::Client);
return d->host;
} }
void QAMQP::Client::setHost( const QString & host ) void QAMQP::Client::setHost( const QString & host )
{ {
pd_func()->host = host; Q_D(QAMQP::Client);
d->host = host;
} }
QString QAMQP::Client::virtualHost() const QString QAMQP::Client::virtualHost() const
{ {
return pd_func()->virtualHost; Q_D(const QAMQP::Client);
return d->virtualHost;
} }
void QAMQP::Client::setVirtualHost( const QString & virtualHost ) void QAMQP::Client::setVirtualHost(const QString &virtualHost)
{ {
pd_func()->virtualHost = virtualHost; Q_D(QAMQP::Client);
d->virtualHost = virtualHost;
} }
QString QAMQP::Client::user() const QString QAMQP::Client::user() const
{ {
const Authenticator * auth = pd_func()->auth_.data(); Q_D(const QAMQP::Client);
const Authenticator * auth = d->auth_.data();
if (auth && auth->type() == QLatin1String("AMQPLAIN")) {
const AMQPlainAuthenticator * a = static_cast<const AMQPlainAuthenticator *>(auth);
return a->login();
}
if(auth && auth->type() == "AMQPLAIN") return QString();
{
const AMQPlainAuthenticator * a = static_cast<const AMQPlainAuthenticator *>(auth);
return a->login();
}
return QString();
} }
void QAMQP::Client::setUser( const QString & user ) void QAMQP::Client::setUser(const QString &user)
{ {
Authenticator * auth = pd_func()->auth_.data(); Q_D(const QAMQP::Client);
Authenticator * auth = d->auth_.data();
if(auth && auth->type() == "AMQPLAIN") if (auth && auth->type() == QLatin1String("AMQPLAIN")) {
{ AMQPlainAuthenticator * a = static_cast<AMQPlainAuthenticator *>(auth);
AMQPlainAuthenticator * a = static_cast<AMQPlainAuthenticator *>(auth); a->setLogin(user);
a->setLogin(user); }
}
} }
QString QAMQP::Client::password() const QString QAMQP::Client::password() const
{ {
const Authenticator * auth = pd_func()->auth_.data(); Q_D(const QAMQP::Client);
const Authenticator * auth = d->auth_.data();
if (auth && auth->type() == "AMQPLAIN") {
const AMQPlainAuthenticator * a = static_cast<const AMQPlainAuthenticator *>(auth);
return a->password();
}
if(auth && auth->type() == "AMQPLAIN") return QString();
{
const AMQPlainAuthenticator * a = static_cast<const AMQPlainAuthenticator *>(auth);
return a->password();
}
return QString();
} }
void QAMQP::Client::setPassword( const QString & password ) void QAMQP::Client::setPassword(const QString &password)
{ {
Authenticator * auth = pd_func()->auth_.data(); Q_D(QAMQP::Client);
Authenticator *auth = d->auth_.data();
if(auth && auth->type() == "AMQPLAIN") if (auth && auth->type() == QLatin1String("AMQPLAIN")) {
{ AMQPlainAuthenticator * a = static_cast<AMQPlainAuthenticator *>(auth);
AMQPlainAuthenticator * a = static_cast<AMQPlainAuthenticator *>(auth); a->setPassword(password);
a->setPassword(password); }
}
} }
void QAMQP::Client::printConnect() const void QAMQP::Client::printConnect() const
{ {
#ifdef _DEBUG #ifdef _DEBUG
pd_func()->printConnect(); Q_D(const QAMQP::Client);
d->printConnect();
#endif // _DEBUG #endif // _DEBUG
} }
void QAMQP::Client::closeChannel() void QAMQP::Client::closeChannel()
{ {
} }
Exchange * QAMQP::Client::createExchange(int channelNumber) Exchange *QAMQP::Client::createExchange(int channelNumber)
{ {
return pd_func()->createExchange(channelNumber, QString()); Q_D(QAMQP::Client);
return d->createExchange(channelNumber, QString());
} }
Exchange * QAMQP::Client::createExchange( const QString &name, int channelNumber ) Exchange *QAMQP::Client::createExchange( const QString &name, int channelNumber )
{ {
return pd_func()->createExchange(channelNumber, name); Q_D(QAMQP::Client);
return d->createExchange(channelNumber, name);
} }
Queue * QAMQP::Client::createQueue(int channelNumber) Queue *QAMQP::Client::createQueue(int channelNumber)
{ {
return pd_func()->createQueue(channelNumber, QString()); Q_D(QAMQP::Client);
return d->createQueue(channelNumber, QString());
} }
Queue * QAMQP::Client::createQueue( const QString &name, int channelNumber ) Queue *QAMQP::Client::createQueue( const QString &name, int channelNumber )
{ {
return pd_func()->createQueue(channelNumber, name); Q_D(QAMQP::Client);
return d->createQueue(channelNumber, name);
} }
void QAMQP::Client::open() void QAMQP::Client::open()
{ {
return pd_func()->connect(); Q_D(QAMQP::Client);
return d->connect();
} }
void QAMQP::Client::open( const QUrl & connectionString ) void QAMQP::Client::open(const QUrl &connectionString)
{ {
pd_func()->parseCnnString(connectionString); Q_D(QAMQP::Client);
open(); d->parseConnectionString(connectionString);
open();
} }
void QAMQP::Client::close() void QAMQP::Client::close()
{ {
return pd_func()->disconnect(); Q_D(QAMQP::Client);
return d->disconnect();
} }
void QAMQP::Client::reopen() void QAMQP::Client::reopen()
{ {
pd_func()->disconnect(); Q_D(QAMQP::Client);
pd_func()->connect(); d->disconnect();
d->connect();
} }
void QAMQP::Client::setAuth( Authenticator * auth ) void QAMQP::Client::setAuth(Authenticator *auth)
{ {
pd_func()->setAuth(auth); Q_D(QAMQP::Client);
d->setAuth(auth);
} }
Authenticator * QAMQP::Client::auth() const Authenticator *QAMQP::Client::auth() const
{ {
return pd_func()->auth_.data(); Q_D(const QAMQP::Client);
return d->auth_.data();
} }
bool QAMQP::Client::isSsl() const bool QAMQP::Client::isSsl() const
{ {
return pd_func()->network_->isSsl(); Q_D(const QAMQP::Client);
return d->network_->isSsl();
} }
void QAMQP::Client::setSsl( bool value ) void QAMQP::Client::setSsl(bool value)
{ {
pd_func()->network_->setSsl(value); Q_D(QAMQP::Client);
d->network_->setSsl(value);
} }
bool QAMQP::Client::autoReconnect() const bool QAMQP::Client::autoReconnect() const
{ {
return pd_func()->network_->autoReconnect(); Q_D(const QAMQP::Client);
return d->network_->autoReconnect();
} }
void QAMQP::Client::setAutoReconnect( bool value ) void QAMQP::Client::setAutoReconnect(bool value)
{ {
pd_func()->network_->setAutoReconnect(value); Q_D(QAMQP::Client);
d->network_->setAutoReconnect(value);
} }
bool QAMQP::Client::isConnected() const bool QAMQP::Client::isConnected() const
{ {
return pd_func()->connection_->isConnected(); Q_D(const QAMQP::Client);
return d->connection_->isConnected();
} }
void QAMQP::Client::addCustomProperty( const QString & name, const QString & value ) void QAMQP::Client::addCustomProperty(const QString &name, const QString &value)
{ {
return pd_func()->connection_->addCustomProperty(name, value); Q_D(QAMQP::Client);
return d->connection_->addCustomProperty(name, value);
} }
QString QAMQP::Client::customProperty( const QString & name ) const QString QAMQP::Client::customProperty(const QString &name) const
{ {
return pd_func()->connection_->customProperty(name); Q_D(const QAMQP::Client);
return d->connection_->customProperty(name);
} }

134
src/qamqp/amqp.h
View File

@ -1,97 +1,93 @@
#ifndef qamqp_amqp_h__ #ifndef QAMQP_H
#define qamqp_amqp_h__ #define QAMQP_H
#include <QObject> #include <QObject>
#include <QUrl> #include <QUrl>
#include "amqp_global.h" #include "amqp_global.h"
namespace QAMQP namespace QAMQP
{ {
class Exchange;
class Queue;
class ClientPrivate;
class Authenticator;
class ConnectionPrivate;
class Client : public QObject
{
Q_OBJECT
Q_PROPERTY(quint32 port READ port WRITE setPort); class Exchange;
Q_PROPERTY(QString host READ host WRITE setHost); class Queue;
Q_PROPERTY(QString virtualHost READ virtualHost WRITE setVirtualHost); class ClientPrivate;
Q_PROPERTY(QString user READ user WRITE setUser); class Authenticator;
Q_PROPERTY(QString password READ password WRITE setPassword); class ConnectionPrivate;
Q_PROPERTY(bool ssl READ isSsl WRITE setSsl); class Client : public QObject
Q_PROPERTY(bool autoReconnect READ autoReconnect WRITE setAutoReconnect); {
Q_PROPERTY(bool connected READ isConnected ); Q_OBJECT
Q_PROPERTY(quint32 port READ port WRITE setPort)
Q_PROPERTY(QString host READ host WRITE setHost)
Q_PROPERTY(QString virtualHost READ virtualHost WRITE setVirtualHost)
Q_PROPERTY(QString user READ user WRITE setUser)
Q_PROPERTY(QString password READ password WRITE setPassword)
Q_PROPERTY(bool ssl READ isSsl WRITE setSsl)
Q_PROPERTY(bool autoReconnect READ autoReconnect WRITE setAutoReconnect)
Q_PROPERTY(bool connected READ isConnected )
Q_DISABLE_COPY(Client) public:
Client(QObject *parent = 0);
Client(const QUrl &connectionString, QObject *parent = 0);
~Client();
P_DECLARE_PRIVATE(QAMQP::Client) void printConnect() const;
void closeChannel();
friend class ConnectionPrivate; void addCustomProperty(const QString &name, const QString &value);
friend class ChannelPrivate; QString customProperty(const QString &name) const;
public: Exchange *createExchange(int channelNumber = -1);
Client(QObject * parent = 0); Exchange *createExchange(const QString &name, int channelNumber = -1);
Client(const QUrl & connectionString, QObject * parent = 0);
~Client();
void printConnect() const; Queue *createQueue(int channelNumber = -1);
void closeChannel(); Queue *createQueue(const QString &name, int channelNumber = -1);
void addCustomProperty(const QString & name, const QString & value); quint16 port() const;
QString customProperty(const QString & name) const; void setPort(quint16 port);
Exchange * createExchange(int channelNumber = -1); QString host() const;
Exchange * createExchange(const QString &name, int channelNumber = -1); void setHost(const QString &host);
Queue * createQueue(int channelNumber = -1); QString virtualHost() const;
Queue * createQueue(const QString &name, int channelNumber = -1); void setVirtualHost(const QString &virtualHost);
quint32 port() const; QString user() const;
void setPort(quint32 port); void setUser(const QString &user);
QString host() const; QString password() const;
void setHost(const QString & host); void setPassword(const QString &password);
QString virtualHost() const; void setAuth(Authenticator *auth);
void setVirtualHost(const QString & virtualHost); Authenticator *auth() const;
QString user() const; void open();
void setUser(const QString & user); void open(const QUrl &connectionString);
void close();
void reopen();
QString password() const; bool isSsl() const;
void setPassword(const QString & password); void setSsl(bool value);
void setAuth(Authenticator * auth); bool autoReconnect() const;
Authenticator * auth() const; void setAutoReconnect(bool value);
void open();
void open(const QUrl & connectionString);
void close();
void reopen();
bool isSsl() const; bool isConnected() const;
void setSsl(bool value);
bool autoReconnect() const; signals:
void setAutoReconnect(bool value); void connected();
void disconnected();
bool isConnected() const; private:
Q_DISABLE_COPY(Client)
Q_DECLARE_PRIVATE(QAMQP::Client)
QScopedPointer<ClientPrivate> d_ptr;
signals: friend class ConnectionPrivate;
void connected(); friend class ChannelPrivate;
void disconnected(); friend struct ClientExceptionCleaner;
};
} // namespace QAMQP
protected: #endif // QAMQP
ClientPrivate * const pd_ptr;
private:
friend struct ClientExceptionCleaner;
//void chanalConnect();
};
}
#endif // qamqp_amqp_h__

43
src/qamqp/amqp_authenticator.cpp
View File

@ -1,47 +1,48 @@
#include "amqp_authenticator.h" #include "amqp_authenticator.h"
#include "amqp_frame.h" #include "amqp_frame.h"
QString QAMQP::AMQPlainAuthenticator::login() const using namespace QAMQP;
AMQPlainAuthenticator::AMQPlainAuthenticator(const QString &l, const QString &p)
: login_(l),
password_(p)
{ {
return login_;
} }
QString QAMQP::AMQPlainAuthenticator::password() const AMQPlainAuthenticator::~AMQPlainAuthenticator()
{ {
return password_;
} }
QAMQP::AMQPlainAuthenticator::AMQPlainAuthenticator( const QString & l /*= QString()*/, const QString & p /*= QString()*/ ) QString AMQPlainAuthenticator::login() const
{ {
login_ = l; return login_;
password_ = p;
} }
QAMQP::AMQPlainAuthenticator::~AMQPlainAuthenticator() QString AMQPlainAuthenticator::password() const
{ {
return password_;
} }
QString QAMQP::AMQPlainAuthenticator::type() const QString AMQPlainAuthenticator::type() const
{ {
return "AMQPLAIN"; return "AMQPLAIN";
} }
void QAMQP::AMQPlainAuthenticator::setLogin( const QString& l ) void AMQPlainAuthenticator::setLogin(const QString &l)
{ {
login_ = l; login_ = l;
} }
void QAMQP::AMQPlainAuthenticator::setPassword( const QString &p ) void AMQPlainAuthenticator::setPassword(const QString &p)
{ {
password_ = p; password_ = p;
} }
void QAMQP::AMQPlainAuthenticator::write( QDataStream & out ) void AMQPlainAuthenticator::write(QDataStream &out)
{ {
QAMQP::Frame::writeField('s', out, type()); QAMQP::Frame::writeField('s', out, type());
QAMQP::Frame::TableField response; QAMQP::Frame::TableField response;
response["LOGIN"] = login_; response["LOGIN"] = login_;
response["PASSWORD"] = password_; response["PASSWORD"] = password_;
QAMQP::Frame::serialize(out, response); QAMQP::Frame::serialize(out, response);
} }

51
src/qamqp/amqp_authenticator.h
View File

@ -1,33 +1,42 @@
#ifndef amqp_authenticator_h__ #ifndef amqp_authenticator_h__
#define amqp_authenticator_h__ #define amqp_authenticator_h__
#include "amqp_global.h"
#include <QString> #include <QString>
#include <QDataStream> #include <QDataStream>
#include "amqp_global.h"
namespace QAMQP namespace QAMQP
{ {
class Authenticator
{
public:
virtual ~Authenticator(){};
virtual QString type() const = 0;
virtual void write(QDataStream & out) = 0;
};
class AMQPlainAuthenticator : public Authenticator class Authenticator
{ {
QString login_, password_; public:
public: virtual ~Authenticator() {}
AMQPlainAuthenticator(const QString & login = QString(), const QString & password = QString()); virtual QString type() const = 0;
virtual ~AMQPlainAuthenticator(); virtual void write(QDataStream &out) = 0;
QString login() const; };
void setLogin(const QString& l);
QString password() const; class AMQPlainAuthenticator : public Authenticator
void setPassword(const QString &p); {
virtual QString type() const; public:
virtual void write(QDataStream & out); AMQPlainAuthenticator(const QString &login = QString(), const QString &password = QString());
}; virtual ~AMQPlainAuthenticator();
QString login() const;
void setLogin(const QString &l);
QString password() const;
void setPassword(const QString &p);
virtual QString type() const;
virtual void write(QDataStream &out);
private:
QString login_;
QString password_;
};
} }
#endif // amqp_authenticator_h__ #endif // amqp_authenticator_h__

337
src/qamqp/amqp_channel.cpp
View File

@ -1,5 +1,6 @@
#include "amqp_channel.h" #include "amqp_channel.h"
#include "amqp_channel_p.h" #include "amqp_channel_p.h"
#include "amqp_connection_p.h"
#include "amqp.h" #include "amqp.h"
#include "amqp_p.h" #include "amqp_p.h"
@ -10,311 +11,283 @@
using namespace QAMQP; using namespace QAMQP;
namespace QAMQP
{
int ChannelPrivate::nextChannelNumber_ = 0;
struct ChannelExceptionCleaner
{
/* this cleans up when the constructor throws an exception */
static inline void cleanup(Channel *that, ChannelPrivate *d)
{
#ifdef QT_NO_EXCEPTIONS
Q_UNUSED(that);
Q_UNUSED(d);
#else
Q_UNUSED(that);
Q_UNUSED(d);
#endif
}
};
}
////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////
QAMQP::Channel::Channel(int channelNumber /*= -1*/, Client * parent /*= 0*/ ) QAMQP::Channel::Channel(int channelNumber, Client *parent)
: pd_ptr(new ChannelPrivate(this)) : QObject(parent),
d_ptr(new ChannelPrivate(this))
{ {
QT_TRY { Q_D(QAMQP::Channel);
pd_func()->init(channelNumber, parent); d->init(channelNumber, parent);
} QT_CATCH(...) {
ChannelExceptionCleaner::cleanup(this, pd_func());
QT_RETHROW;
}
} }
QAMQP::Channel::Channel( ChannelPrivate * d ) QAMQP::Channel::Channel(ChannelPrivate *dd, Client *parent)
: pd_ptr(d) : QObject(parent),
d_ptr(dd)
{ {
} }
QAMQP::Channel::~Channel() QAMQP::Channel::~Channel()
{ {
QT_TRY {
QEvent e(QEvent::Destroy);
QCoreApplication::sendEvent(this, &e);
} QT_CATCH(const std::exception&) {
// if this fails we can't do anything about it but at least we are not allowed to throw.
}
} }
void QAMQP::Channel::closeChannel() void QAMQP::Channel::closeChannel()
{ {
P_D(Channel); Q_D(QAMQP::Channel);
d->needOpen = true; d->needOpen = true;
if(d->opened) if (d->opened)
d->close(0, QString(), 0,0); d->close(0, QString(), 0,0);
} }
void QAMQP::Channel::reopen() void QAMQP::Channel::reopen()
{ {
closeChannel(); Q_D(QAMQP::Channel);
pd_func()->open(); closeChannel();
d->open();
} }
QString QAMQP::Channel::name() QString QAMQP::Channel::name() const
{ {
return pd_func()->name; Q_D(const QAMQP::Channel);
return d->name;
} }
int QAMQP::Channel::channelNumber() int QAMQP::Channel::channelNumber() const
{ {
return pd_func()->number; Q_D(const QAMQP::Channel);
return d->number;
} }
void QAMQP::Channel::setName( const QString &name ) void QAMQP::Channel::setName(const QString &name)
{ {
pd_func()->name = name; Q_D(QAMQP::Channel);
d->name = name;
} }
void QAMQP::Channel::stateChanged( int state ) void QAMQP::Channel::stateChanged(int state)
{ {
switch(ChannelPrivate::State(state)) switch(ChannelPrivate::State(state)) {
{ case ChannelPrivate::csOpened:
case ChannelPrivate::csOpened: emit opened();
emit opened(); break;
break; case ChannelPrivate::csClosed:
case ChannelPrivate::csClosed: emit closed();
emit closed(); break;
break; case ChannelPrivate::csIdle:
case ChannelPrivate::csIdle: emit flowChanged(false);
emit flowChanged(false); break;
break; case ChannelPrivate::csRunning:
case ChannelPrivate::csRunning: emit flowChanged(true);
emit flowChanged(true); break;
break; }
}
} }
void QAMQP::Channel::_q_method(const Frame::Method &frame) void QAMQP::Channel::_q_method(const Frame::Method &frame)
{ {
pd_func()->_q_method(frame); Q_D(QAMQP::Channel);
d->_q_method(frame);
} }
bool QAMQP::Channel::isOpened() const bool QAMQP::Channel::isOpened() const
{ {
return pd_func()->opened; Q_D(const QAMQP::Channel);
return d->opened;
} }
void QAMQP::Channel::onOpen() void QAMQP::Channel::onOpen()
{ {
} }
void QAMQP::Channel::onClose() void QAMQP::Channel::onClose()
{ {
} }
void QAMQP::Channel::setQOS( qint32 prefetchSize, quint16 prefetchCount ) void QAMQP::Channel::setQOS(qint32 prefetchSize, quint16 prefetchCount)
{ {
pd_func()->setQOS(prefetchSize, prefetchCount); Q_D(QAMQP::Channel);
d->setQOS(prefetchSize, prefetchCount);
} }
////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////
int ChannelPrivate::nextChannelNumber_ = 0;
ChannelPrivate::ChannelPrivate(Channel * q) ChannelPrivate::ChannelPrivate(Channel * q)
: number(0) : number(0),
, opened(false) opened(false),
, needOpen(true) needOpen(true),
, pq_ptr(q) q_ptr(q)
{ {
} }
ChannelPrivate::~ChannelPrivate() ChannelPrivate::~ChannelPrivate()
{ {
} }
void ChannelPrivate::init(int channelNumber, Client * parent) void ChannelPrivate::init(int channelNumber, Client *parent)
{ {
needOpen = channelNumber == -1 ? true : false; Q_Q(QAMQP::Channel);
number = channelNumber == -1 ? ++nextChannelNumber_ : channelNumber; needOpen = channelNumber == -1 ? true : false;
nextChannelNumber_ = qMax(channelNumber, (nextChannelNumber_ + 1)); number = channelNumber == -1 ? ++nextChannelNumber_ : channelNumber;
pq_func()->setParent(parent); nextChannelNumber_ = qMax(channelNumber, (nextChannelNumber_ + 1));
client_ = parent; q->setParent(parent);
client_ = parent;
} }
bool ChannelPrivate::_q_method( const QAMQP::Frame::Method & frame ) bool ChannelPrivate::_q_method(const QAMQP::Frame::Method &frame)
{ {
Q_ASSERT(frame.channel() == number); Q_ASSERT(frame.channel() == number);
if(frame.channel() != number ) if (frame.channel() != number)
return true; return true;
if(frame.methodClass() != QAMQP::Frame::fcChannel) if (frame.methodClass() != QAMQP::Frame::fcChannel)
return false; return false;
qDebug("Channel#%d:", number); qDebug("Channel#%d:", number);
switch(frame.id()) switch (frame.id()) {
{ case miOpenOk:
case miOpenOk: openOk(frame);
openOk(frame); break;
break; case miFlow:
case miFlow: flow(frame);
flow(frame); break;
break; case miFlowOk:
case miFlowOk: flowOk(frame);
flowOk(frame); break;
break; case miClose:
case miClose: close(frame);
close(frame); break;
break; case miCloseOk:
case miCloseOk: closeOk(frame);
closeOk(frame); break;
break; }
} return true;
return true;
} }
void ChannelPrivate::_q_open() void ChannelPrivate::_q_open()
{ {
open(); open();
} }
void ChannelPrivate::sendFrame( const QAMQP::Frame::Base & frame ) void ChannelPrivate::sendFrame(const QAMQP::Frame::Base &frame)
{ {
if(client_) if (client_) {
{ client_->d_func()->network_->sendFrame(frame);
client_->pd_func()->network_->sendFrame(frame); }
}
} }
void ChannelPrivate::open() void ChannelPrivate::open()
{ {
if(!needOpen || opened) if (!needOpen || opened)
return; return;
if(!client_->pd_func()->connection_->isConnected()) if (!client_->d_func()->connection_->isConnected())
return; return;
qDebug("Open channel #%d", number);
QAMQP::Frame::Method frame(QAMQP::Frame::fcChannel, miOpen); qDebug("Open channel #%d", number);
frame.setChannel(number); QAMQP::Frame::Method frame(QAMQP::Frame::fcChannel, miOpen);
QByteArray arguments_; frame.setChannel(number);
arguments_.resize(1); QByteArray arguments_;
arguments_[0] = 0; arguments_.resize(1);
frame.setArguments(arguments_); arguments_[0] = 0;
sendFrame(frame); frame.setArguments(arguments_);
sendFrame(frame);
} }
void ChannelPrivate::flow() void ChannelPrivate::flow()
{ {
} }
void ChannelPrivate::flow( const QAMQP::Frame::Method & frame ) void ChannelPrivate::flow(const QAMQP::Frame::Method &frame)
{ {
Q_UNUSED(frame); Q_UNUSED(frame);
} }
void ChannelPrivate::flowOk() void ChannelPrivate::flowOk()
{ {
} }
void ChannelPrivate::flowOk( const QAMQP::Frame::Method & frame ) void ChannelPrivate::flowOk(const QAMQP::Frame::Method &frame)
{ {
Q_UNUSED(frame); Q_UNUSED(frame);
} }
void ChannelPrivate::close(int code, const QString & text, int classId, int methodId) void ChannelPrivate::close(int code, const QString &text, int classId, int methodId)
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcChannel, miClose); QAMQP::Frame::Method frame(QAMQP::Frame::fcChannel, miClose);
QByteArray arguments_; QByteArray arguments_;
QDataStream stream(&arguments_, QIODevice::WriteOnly); QDataStream stream(&arguments_, QIODevice::WriteOnly);
QAMQP::Frame::writeField('s',stream, client_->virtualHost()); QAMQP::Frame::writeField('s',stream, client_->virtualHost());
stream << qint16(code); stream << qint16(code);
QAMQP::Frame::writeField('s', stream, text); QAMQP::Frame::writeField('s', stream, text);
stream << qint16(classId); stream << qint16(classId);
stream << qint16(methodId); stream << qint16(methodId);
frame.setArguments(arguments_); frame.setArguments(arguments_);
client_->pd_func()->network_->sendFrame(frame); client_->d_func()->network_->sendFrame(frame);
} }
void ChannelPrivate::close( const QAMQP::Frame::Method & frame ) void ChannelPrivate::close(const QAMQP::Frame::Method &frame)
{ {
pq_func()->stateChanged(csClosed); Q_Q(QAMQP::Channel);
q->stateChanged(csClosed);
qDebug(">> CLOSE"); qDebug(">> CLOSE");
QByteArray data = frame.arguments(); QByteArray data = frame.arguments();
QDataStream stream(&data, QIODevice::ReadOnly); QDataStream stream(&data, QIODevice::ReadOnly);
qint16 code_ = 0, classId, methodId; qint16 code_ = 0, classId, methodId;
stream >> code_; stream >> code_;
QString text(QAMQP::Frame::readField('s', stream).toString()); QString text(QAMQP::Frame::readField('s', stream).toString());
stream >> classId; stream >> classId;
stream >> methodId; stream >> methodId;
qDebug(">> code: %d", code_);
qDebug(">> text: %s", qPrintable(text));
qDebug(">> class-id: %d", classId);
qDebug(">> method-id: %d", methodId);
qDebug(">> code: %d", code_);
qDebug(">> text: %s", qPrintable(text));
qDebug(">> class-id: %d", classId);
qDebug(">> method-id: %d", methodId);
} }
void ChannelPrivate::closeOk() void ChannelPrivate::closeOk()
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcChannel, miCloseOk); QAMQP::Frame::Method frame(QAMQP::Frame::fcChannel, miCloseOk);
sendFrame(frame); sendFrame(frame);
} }
void ChannelPrivate::closeOk( const QAMQP::Frame::Method & frame ) void ChannelPrivate::closeOk(const QAMQP::Frame::Method &frame)
{ {
Q_UNUSED(frame); Q_UNUSED(frame)
P_Q(Channel); Q_Q(QAMQP::Channel);
q->stateChanged(csClosed);
q->onClose(); q->stateChanged(csClosed);
opened = false; q->onClose();
opened = false;
} }
void ChannelPrivate::openOk( const QAMQP::Frame::Method & frame ) void ChannelPrivate::openOk(const QAMQP::Frame::Method &frame)
{ {
Q_UNUSED(frame); Q_UNUSED(frame)
P_Q(Channel); Q_Q(QAMQP::Channel);
qDebug(">> OpenOK");
opened = true;
q->stateChanged(csOpened);
q->onOpen();
qDebug(">> OpenOK");
opened = true;
q->stateChanged(csOpened);
q->onOpen();
} }
void ChannelPrivate::setQOS( qint32 prefetchSize, quint16 prefetchCount ) void ChannelPrivate::setQOS(qint32 prefetchSize, quint16 prefetchCount)
{ {
client_->pd_func()->connection_->pd_func()->setQOS(prefetchSize, prefetchCount, number, false); client_->d_func()->connection_->d_func()->setQOS(prefetchSize, prefetchCount, number, false);
} }
void ChannelPrivate::_q_disconnected() void ChannelPrivate::_q_disconnected()
{ {
nextChannelNumber_ = 0; nextChannelNumber_ = 0;
opened = false; opened = false;
} }
#include "moc_amqp_channel.cpp"

74
src/qamqp/amqp_channel.h
View File

@ -7,54 +7,52 @@
namespace QAMQP namespace QAMQP
{ {
class ChannelPrivate;
class Client;
class Channel : public QObject, public Frame::MethodHandler
{
Q_OBJECT
Q_PROPERTY(int number READ channelNumber); class Client;
Q_PROPERTY(QString name READ name WRITE setName); class ChannelPrivate;
class Channel : public QObject, public Frame::MethodHandler
{
Q_OBJECT
Q_PROPERTY(int number READ channelNumber)
Q_PROPERTY(QString name READ name WRITE setName)
P_DECLARE_PRIVATE(QAMQP::Channel) public:
Q_DISABLE_COPY(Channel) ~Channel();
public:
~Channel();
void closeChannel(); void closeChannel();
void reopen(); void reopen();
QString name(); QString name() const;
int channelNumber(); int channelNumber() const;
void setName(const QString &name); void setName(const QString &name);
void setQOS(qint32 prefetchSize, quint16 prefetchCount); void setQOS(qint32 prefetchSize, quint16 prefetchCount);
bool isOpened() const; bool isOpened() const;
signals: signals:
void opened(); void opened();
void closed(); void closed();
void flowChanged(bool enabled); void flowChanged(bool enabled);
protected: protected:
Channel(int channelNumber = -1, Client * parent = 0); Q_DISABLE_COPY(Channel)
Channel(ChannelPrivate * d); Q_DECLARE_PRIVATE(QAMQP::Channel)
virtual void onOpen();
virtual void onClose();
ChannelPrivate * const pd_ptr; Channel(int channelNumber = -1, Client *parent = 0);
Channel(ChannelPrivate *dd, Client *parent = 0);
QScopedPointer<ChannelPrivate> d_ptr;
private: Q_PRIVATE_SLOT(d_func(), void _q_open())
void stateChanged(int state); Q_PRIVATE_SLOT(d_func(), void _q_disconnected())
friend class ClientPrivate;
void _q_method(const QAMQP::Frame::Method & frame); virtual void onOpen();
virtual void onClose();
void stateChanged(int state);
void _q_method(const QAMQP::Frame::Method &frame);
friend class ClientPrivate;
};
Q_PRIVATE_SLOT(pd_func(), void _q_open())
Q_PRIVATE_SLOT(pd_func(), void _q_disconnected())
};
} }
#ifdef QAMQP_P_INCLUDE
# include "amqp_channel_p.h"
#endif #endif
#endif // amqp_channel_h__

120
src/qamqp/amqp_channel_p.h
View File

@ -8,78 +8,78 @@
namespace QAMQP namespace QAMQP
{ {
class Client; class Client;
class ClientPrivate; class ClientPrivate;
class ChannelPrivate class ChannelPrivate
{ {
P_DECLARE_PUBLIC(QAMQP::Channel) public:
public: enum MethodId {
enum MethodId METHOD_ID_ENUM(miOpen, 10),
{ METHOD_ID_ENUM(miFlow, 20),
METHOD_ID_ENUM(miOpen, 10), METHOD_ID_ENUM(miClose, 40)
METHOD_ID_ENUM(miFlow, 20), };
METHOD_ID_ENUM(miClose, 40)
};
enum State { enum State {
csOpened, csOpened,
csClosed, csClosed,
csIdle, csIdle,
csRunning csRunning
}; };
enum BasicMethod enum BasicMethod
{ {
METHOD_ID_ENUM(bmQos, 10), METHOD_ID_ENUM(bmQos, 10),
METHOD_ID_ENUM(bmConsume, 20), METHOD_ID_ENUM(bmConsume, 20),
METHOD_ID_ENUM(bmCancel, 30), METHOD_ID_ENUM(bmCancel, 30),
bmPublish = 40, bmPublish = 40,
bmReturn = 50, bmReturn = 50,
bmDeliver = 60, bmDeliver = 60,
METHOD_ID_ENUM(bmGet, 70), METHOD_ID_ENUM(bmGet, 70),
bmGetEmpty = 72, bmGetEmpty = 72,
bmAck = 80, bmAck = 80,
bmReject = 90, bmReject = 90,
bmRecoverAsync = 100, bmRecoverAsync = 100,
METHOD_ID_ENUM(bmRecover, 110) METHOD_ID_ENUM(bmRecover, 110)
}; };
ChannelPrivate(Channel * q); ChannelPrivate(Channel * q);
virtual ~ChannelPrivate(); virtual ~ChannelPrivate();
void init(int channelNumber, Client * parent); void init(int channelNumber, Client *parent);
void open(); void open();
void flow(); void flow();
void flowOk(); void flowOk();
void close(int code, const QString & text, int classId, int methodId); void close(int code, const QString &text, int classId, int methodId);
void closeOk(); void closeOk();
////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////
void openOk(const QAMQP::Frame::Method & frame); void openOk(const QAMQP::Frame::Method &frame);
void flow(const QAMQP::Frame::Method & frame); void flow(const QAMQP::Frame::Method &frame);
void flowOk(const QAMQP::Frame::Method & frame); void flowOk(const QAMQP::Frame::Method &frame);
void close(const QAMQP::Frame::Method & frame); void close(const QAMQP::Frame::Method &frame);
void closeOk(const QAMQP::Frame::Method & frame); void closeOk(const QAMQP::Frame::Method &frame);
virtual bool _q_method(const QAMQP::Frame::Method & frame); virtual bool _q_method(const QAMQP::Frame::Method &frame);
virtual void _q_disconnected(); virtual void _q_disconnected();
void _q_open(); void _q_open();
void setQOS(qint32 prefetchSize, quint16 prefetchCount); void setQOS(qint32 prefetchSize, quint16 prefetchCount);
void sendFrame(const QAMQP::Frame::Base & frame); void sendFrame(const QAMQP::Frame::Base &frame);
QPointer<Client> client_; QPointer<Client> client_;
QString name; QString name;
int number; int number;
static int nextChannelNumber_; static int nextChannelNumber_;
bool opened; bool opened;
bool needOpen; bool needOpen;
Q_DECLARE_PUBLIC(QAMQP::Channel)
Channel * const q_ptr;
};
Channel * const pq_ptr;
};
} }
#endif // amqp_channel_p_h__ #endif // amqp_channel_p_h__

454
src/qamqp/amqp_connection.cpp
View File

@ -12,386 +12,358 @@
using namespace QAMQP; using namespace QAMQP;
namespace QAMQP ConnectionPrivate::ConnectionPrivate(Connection * q)
: closed_(false),
connected(false),
q_ptr(q)
{ {
struct ConnectionExceptionCleaner
{
/* this cleans up when the constructor throws an exception */
static inline void cleanup(Connection *that, ConnectionPrivate *d)
{
#ifdef QT_NO_EXCEPTIONS
Q_UNUSED(that);
Q_UNUSED(d);
#else
Q_UNUSED(that);
Q_UNUSED(d);
#endif
}
};
}
//////////////////////////////////////////////////////////////////////////
ConnectionPrivate::ConnectionPrivate( Connection * q)
: closed_(false)
, connected(false)
, pq_ptr(q)
{
} }
ConnectionPrivate::~ConnectionPrivate() ConnectionPrivate::~ConnectionPrivate()
{ {
} }
void ConnectionPrivate::init(Client * parent) void ConnectionPrivate::init(Client * parent)
{ {
pq_func()->setParent(parent); Q_Q(QAMQP::Connection);
client_ = parent; q->setParent(parent);
heartbeatTimer_ = new QTimer(parent); client_ = parent;
QObject::connect(heartbeatTimer_, SIGNAL(timeout()), heartbeatTimer_ = new QTimer(parent);
pq_func(), SLOT(_q_heartbeat())); QObject::connect(heartbeatTimer_, SIGNAL(timeout()), q, SLOT(_q_heartbeat()));
} }
void ConnectionPrivate::startOk() void ConnectionPrivate::startOk()
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miStartOk); QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miStartOk);
QByteArray arguments_; QByteArray arguments_;
QDataStream stream(&arguments_, QIODevice::WriteOnly); QDataStream stream(&arguments_, QIODevice::WriteOnly);
QAMQP::Frame::TableField clientProperties; QAMQP::Frame::TableField clientProperties;
clientProperties["version"] = QString(QAMQP_VERSION); clientProperties["version"] = QString(QAMQP_VERSION);
clientProperties["platform"] = QString("Qt %1").arg(qVersion()); clientProperties["platform"] = QString("Qt %1").arg(qVersion());
clientProperties["product"] = QString("QAMQP"); clientProperties["product"] = QString("QAMQP");
clientProperties.unite(customProperty); clientProperties.unite(customProperty);
QAMQP::Frame::serialize(stream, clientProperties); QAMQP::Frame::serialize(stream, clientProperties);
client_->pd_func()->auth_->write(stream); client_->d_func()->auth_->write(stream);
QAMQP::Frame::writeField('s', stream, "en_US");
QAMQP::Frame::writeField('s', stream, "en_US"); frame.setArguments(arguments_);
client_->d_func()->network_->sendFrame(frame);
frame.setArguments(arguments_);
client_->pd_func()->network_->sendFrame(frame);
} }
void ConnectionPrivate::secureOk() void ConnectionPrivate::secureOk()
{ {
} }
void ConnectionPrivate::tuneOk() void ConnectionPrivate::tuneOk()
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miTuneOk); QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miTuneOk);
QByteArray arguments_; QByteArray arguments_;
QDataStream stream(&arguments_, QIODevice::WriteOnly); QDataStream stream(&arguments_, QIODevice::WriteOnly);
stream << qint16(0); //channel_max stream << qint16(0); //channel_max
stream << qint32(FRAME_MAX); //frame_max stream << qint32(FRAME_MAX); //frame_max
stream << qint16(heartbeatTimer_->interval() / 1000); //heartbeat stream << qint16(heartbeatTimer_->interval() / 1000); //heartbeat
frame.setArguments(arguments_); frame.setArguments(arguments_);
client_->pd_func()->network_->sendFrame(frame); client_->d_func()->network_->sendFrame(frame);
} }
void ConnectionPrivate::open() void ConnectionPrivate::open()
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miOpen); QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miOpen);
QByteArray arguments_; QByteArray arguments_;
QDataStream stream(&arguments_, QIODevice::WriteOnly); QDataStream stream(&arguments_, QIODevice::WriteOnly);
QAMQP::Frame::writeField('s',stream, client_->virtualHost()); QAMQP::Frame::writeField('s',stream, client_->virtualHost());
stream << qint8(0); stream << qint8(0);
stream << qint8(0); stream << qint8(0);
frame.setArguments(arguments_); frame.setArguments(arguments_);
client_->pd_func()->network_->sendFrame(frame); client_->d_func()->network_->sendFrame(frame);
} }
void ConnectionPrivate::start( const QAMQP::Frame::Method & frame ) void ConnectionPrivate::start(const QAMQP::Frame::Method &frame)
{ {
qDebug(">> Start"); qDebug(">> Start");
QByteArray data = frame.arguments(); QByteArray data = frame.arguments();
QDataStream stream(&data, QIODevice::ReadOnly); QDataStream stream(&data, QIODevice::ReadOnly);
quint8 version_major = 0; quint8 version_major = 0;
quint8 version_minor = 0; quint8 version_minor = 0;
stream >> version_major >> version_minor; stream >> version_major >> version_minor;
QAMQP::Frame::TableField table; QAMQP::Frame::TableField table;
QAMQP::Frame::deserialize(stream, table); QAMQP::Frame::deserialize(stream, table);
QString mechanisms = QAMQP::Frame::readField('S', stream).toString(); QString mechanisms = QAMQP::Frame::readField('S', stream).toString();
QString locales = QAMQP::Frame::readField('S', stream).toString(); QString locales = QAMQP::Frame::readField('S', stream).toString();
qDebug(">> version_major: %d", version_major); qDebug(">> version_major: %d", version_major);
qDebug(">> version_minor: %d", version_minor); qDebug(">> version_minor: %d", version_minor);
QAMQP::Frame::print(table); QAMQP::Frame::print(table);
qDebug(">> mechanisms: %s", qPrintable(mechanisms)); qDebug(">> mechanisms: %s", qPrintable(mechanisms));
qDebug(">> locales: %s", qPrintable(locales)); qDebug(">> locales: %s", qPrintable(locales));
startOk(); startOk();
} }
void ConnectionPrivate::secure( const QAMQP::Frame::Method & frame ) void ConnectionPrivate::secure(const QAMQP::Frame::Method &frame)
{ {
Q_UNUSED(frame); Q_UNUSED(frame)
} }
void ConnectionPrivate::tune( const QAMQP::Frame::Method & frame ) void ConnectionPrivate::tune(const QAMQP::Frame::Method &frame)
{ {
qDebug(">> Tune"); qDebug(">> Tune");
QByteArray data = frame.arguments(); QByteArray data = frame.arguments();
QDataStream stream(&data, QIODevice::ReadOnly); QDataStream stream(&data, QIODevice::ReadOnly);
qint16 channel_max = 0, qint16 channel_max = 0,
heartbeat = 0; heartbeat = 0;
qint32 frame_max = 0; qint32 frame_max = 0;
stream >> channel_max; stream >> channel_max;
stream >> frame_max; stream >> frame_max;
stream >> heartbeat; stream >> heartbeat;
qDebug(">> channel_max: %d", channel_max); qDebug(">> channel_max: %d", channel_max);
qDebug(">> frame_max: %d", frame_max); qDebug(">> frame_max: %d", frame_max);
qDebug(">> heartbeat: %d", heartbeat); qDebug(">> heartbeat: %d", heartbeat);
if(heartbeatTimer_) if (heartbeatTimer_) {
{ heartbeatTimer_->setInterval(heartbeat * 1000);
heartbeatTimer_->setInterval(heartbeat * 1000); if (heartbeatTimer_->interval())
if(heartbeatTimer_->interval()) heartbeatTimer_->start();
{ else
heartbeatTimer_->start(); heartbeatTimer_->stop();
} else { }
heartbeatTimer_->stop();
} tuneOk();
} open();
tuneOk();
open();
} }
void ConnectionPrivate::openOk( const QAMQP::Frame::Method & frame ) void ConnectionPrivate::openOk(const QAMQP::Frame::Method &frame)
{ {
Q_UNUSED(frame); Q_UNUSED(frame)
qDebug(">> OpenOK"); Q_Q(QAMQP::Connection);
connected = true;
pq_func()->openOk(); qDebug(">> OpenOK");
connected = true;
q->openOk();
} }
void ConnectionPrivate::close( const QAMQP::Frame::Method & frame ) void ConnectionPrivate::close(const QAMQP::Frame::Method &frame)
{ {
qDebug(">> CLOSE"); Q_Q(QAMQP::Connection);
QByteArray data = frame.arguments();
QDataStream stream(&data, QIODevice::ReadOnly);
qint16 code_ = 0, classId, methodId;
stream >> code_;
QString text(QAMQP::Frame::readField('s', stream).toString());
stream >> classId;
stream >> methodId;
qDebug(">> code: %d", code_); qDebug(">> CLOSE");
qDebug(">> text: %s", qPrintable(text)); QByteArray data = frame.arguments();
qDebug(">> class-id: %d", classId); QDataStream stream(&data, QIODevice::ReadOnly);
qDebug(">> method-id: %d", methodId); qint16 code_ = 0, classId, methodId;
connected = false; stream >> code_;
client_->pd_func()->network_->error(QAbstractSocket::RemoteHostClosedError); QString text(QAMQP::Frame::readField('s', stream).toString());
QMetaObject::invokeMethod(pq_func(), "disconnected"); stream >> classId;
stream >> methodId;
qDebug(">> code: %d", code_);
qDebug(">> text: %s", qPrintable(text));
qDebug(">> class-id: %d", classId);
qDebug(">> method-id: %d", methodId);
connected = false;
client_->d_func()->network_->error(QAbstractSocket::RemoteHostClosedError);
QMetaObject::invokeMethod(q, "disconnected");
} }
void ConnectionPrivate::close(int code, const QString & text, int classId, int methodId) void ConnectionPrivate::close(int code, const QString &text, int classId, int methodId)
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miClose); QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miClose);
QByteArray arguments_; QByteArray arguments_;
QDataStream stream(&arguments_, QIODevice::WriteOnly); QDataStream stream(&arguments_, QIODevice::WriteOnly);
QAMQP::Frame::writeField('s',stream, client_->virtualHost()); QAMQP::Frame::writeField('s',stream, client_->virtualHost());
stream << qint16(code); stream << qint16(code);
QAMQP::Frame::writeField('s', stream, text); QAMQP::Frame::writeField('s', stream, text);
stream << qint16(classId); stream << qint16(classId);
stream << qint16(methodId); stream << qint16(methodId);
frame.setArguments(arguments_); frame.setArguments(arguments_);
client_->pd_func()->network_->sendFrame(frame); client_->d_func()->network_->sendFrame(frame);
} }
void ConnectionPrivate::closeOk() void ConnectionPrivate::closeOk()
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miCloseOk); QAMQP::Frame::Method frame(QAMQP::Frame::fcConnection, miCloseOk);
connected = false; connected = false;
client_->pd_func()->network_->sendFrame(frame); client_->d_func()->network_->sendFrame(frame);
} }
void ConnectionPrivate::closeOk( const QAMQP::Frame::Method & ) void ConnectionPrivate::closeOk(const QAMQP::Frame::Method &frame)
{ {
connected = false; Q_UNUSED(frame)
QMetaObject::invokeMethod(pq_func(), "disconnected"); Q_Q(QAMQP::Connection);
if(heartbeatTimer_)
{ connected = false;
heartbeatTimer_->stop(); QMetaObject::invokeMethod(q, "disconnected");
} if (heartbeatTimer_)
heartbeatTimer_->stop();
} }
void ConnectionPrivate::setQOS( qint32 prefetchSize, quint16 prefetchCount, int channel, bool global ) void ConnectionPrivate::setQOS(qint32 prefetchSize, quint16 prefetchCount, int channel, bool global)
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, 10); QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, 10);
frame.setChannel(channel); frame.setChannel(channel);
QByteArray arguments_; QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly); QDataStream out(&arguments_, QIODevice::WriteOnly);
out << prefetchSize; out << prefetchSize;
out << prefetchCount; out << prefetchCount;
out << qint8(global ? 1 : 0); out << qint8(global ? 1 : 0);
frame.setArguments(arguments_); frame.setArguments(arguments_);
client_->pd_func()->network_->sendFrame(frame); client_->d_func()->network_->sendFrame(frame);
} }
bool ConnectionPrivate::_q_method( const QAMQP::Frame::Method & frame ) bool ConnectionPrivate::_q_method(const QAMQP::Frame::Method &frame)
{ {
Q_ASSERT(frame.methodClass() == QAMQP::Frame::fcConnection); Q_ASSERT(frame.methodClass() == QAMQP::Frame::fcConnection);
if(frame.methodClass() != QAMQP::Frame::fcConnection) if (frame.methodClass() != QAMQP::Frame::fcConnection)
return true; return true;
qDebug() << "Connection:"; qDebug() << "Connection:";
if (closed_) if (closed_) {
{ if (frame.id() == miCloseOk)
if( frame.id() == miCloseOk) closeOk(frame);
closeOk(frame);
return true;
}
switch(MethodId(frame.id())) return true;
{ }
case miStart:
start(frame); switch (MethodId(frame.id())) {
break; case miStart:
case miSecure: start(frame);
secure(frame); break;
break; case miSecure:
case miTune: secure(frame);
tune(frame); break;
break; case miTune:
case miOpenOk: tune(frame);
openOk(frame); break;
break; case miOpenOk:
case miClose: openOk(frame);
close(frame); break;
break; case miClose:
case miCloseOk: close(frame);
closeOk(frame); break;
break; case miCloseOk:
default: closeOk(frame);
qWarning("Unknown method-id %d", frame.id()); break;
return false; default:
} qWarning("Unknown method-id %d", frame.id());
return true; return false;
}
return true;
} }
void ConnectionPrivate::_q_heartbeat() void ConnectionPrivate::_q_heartbeat()
{ {
QAMQP::Frame::Heartbeat frame; QAMQP::Frame::Heartbeat frame;
client_->pd_func()->network_->sendFrame(frame); client_->d_func()->network_->sendFrame(frame);
} }
////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////
Connection::Connection( Client * parent /*= 0*/ ) Connection::Connection(Client *parent)
: pd_ptr(new ConnectionPrivate(this)) : QObject(parent),
d_ptr(new ConnectionPrivate(this))
{ {
QT_TRY { Q_D(QAMQP::Connection);
pd_func()->init(parent); d->init(parent);
} QT_CATCH(...) {
ConnectionExceptionCleaner::cleanup(this, pd_func());
QT_RETHROW;
}
} }
Connection::~Connection() Connection::~Connection()
{ {
QT_TRY {
QEvent e(QEvent::Destroy);
QCoreApplication::sendEvent(this, &e);
} QT_CATCH(const std::exception&) {
// if this fails we can't do anything about it but at least we are not allowed to throw.
}
} }
void Connection::startOk() void Connection::startOk()
{ {
pd_func()->startOk(); Q_D(QAMQP::Connection);
d->startOk();
} }
void Connection::secureOk() void Connection::secureOk()
{ {
pd_func()->secureOk(); Q_D(QAMQP::Connection);
d->secureOk();
} }
void Connection::tuneOk() void Connection::tuneOk()
{ {
pd_func()->tuneOk(); Q_D(QAMQP::Connection);
d->tuneOk();
} }
void Connection::open() void Connection::open()
{ {
pd_func()->open(); Q_D(QAMQP::Connection);
d->open();
} }
void Connection::close(int code, const QString & text, int classId , int methodId) void Connection::close(int code, const QString &text, int classId , int methodId)
{ {
pd_func()->close(code, text, classId, methodId); Q_D(QAMQP::Connection);
d->close(code, text, classId, methodId);
} }
void Connection::closeOk() void Connection::closeOk()
{ {
pd_func()->closeOk(); Q_D(QAMQP::Connection);
emit disconnect(); d->closeOk();
Q_EMIT disconnect();
} }
void Connection::openOk() void Connection::openOk()
{ {
emit connected(); Q_EMIT connected();
} }
void Connection::_q_method(const QAMQP::Frame::Method & frame) void Connection::_q_method(const QAMQP::Frame::Method &frame)
{ {
pd_func()->_q_method(frame); Q_D(QAMQP::Connection);
d->_q_method(frame);
} }
bool Connection::isConnected() const bool Connection::isConnected() const
{ {
return pd_func()->connected; Q_D(const QAMQP::Connection);
return d->connected;
} }
void Connection::setQOS(qint32 prefetchSize, quint16 prefetchCount)
void Connection::setQOS( qint32 prefetchSize, quint16 prefetchCount )
{ {
pd_func()->setQOS(prefetchSize, prefetchCount, 0, true); Q_D(QAMQP::Connection);
d->setQOS(prefetchSize, prefetchCount, 0, true);
} }
void Connection::addCustomProperty(const QString &name, const QString &value)
void Connection::addCustomProperty( const QString & name, const QString & value )
{ {
pd_func()->customProperty[name] = value; Q_D(QAMQP::Connection);
d->customProperty[name] = value;
} }
QString Connection::customProperty( const QString & name ) const QString Connection::customProperty(const QString &name) const
{ {
if(pd_func()->customProperty.contains(name)) Q_D(const QAMQP::Connection);
{ if (d->customProperty.contains(name))
return pd_func()->customProperty.value(name).toString(); return d->customProperty.value(name).toString();
} return QString();
return QString();
} }
#include "moc_amqp_connection.cpp"

75
src/qamqp/amqp_connection.h
View File

@ -7,53 +7,50 @@
namespace QAMQP namespace QAMQP
{ {
class ConnectionPrivate;
class ChannelPrivate;
class ClientPrivate;
class Client;
class Connection : public QObject, public Frame::MethodHandler
{
Q_OBJECT
P_DECLARE_PRIVATE(QAMQP::Connection)
Q_DISABLE_COPY(Connection)
Connection(Client * parent = 0);
public:
~Connection();
class Client;
class ClientPrivate;
class ChannelPrivate;
class ConnectionPrivate;
class Connection : public QObject, public Frame::MethodHandler
{
Q_OBJECT
public:
virtual ~Connection();
void addCustomProperty(const QString & name, const QString & value); void addCustomProperty(const QString &name, const QString &value);
QString customProperty(const QString & name) const; QString customProperty(const QString &name) const;
void startOk(); void startOk();
void secureOk(); void secureOk();
void tuneOk(); void tuneOk();
void open(); void open();
void close(int code, const QString & text, int classId = 0, int methodId = 0); void close(int code, const QString &text, int classId = 0, int methodId = 0);
void closeOk(); void closeOk();
bool isConnected() const; bool isConnected() const;
void setQOS(qint32 prefetchSize, quint16 prefetchCount); void setQOS(qint32 prefetchSize, quint16 prefetchCount);
Q_SIGNALS: Q_SIGNALS:
void disconnected(); void disconnected();
void connected(); void connected();
protected:
ConnectionPrivate * const pd_ptr;
private: private:
void openOk(); Q_DISABLE_COPY(Connection)
friend class ClientPrivate; Q_DECLARE_PRIVATE(Connection)
friend class ChannelPrivate; QScopedPointer<ConnectionPrivate> d_ptr;
Connection(Client * parent = 0);
void openOk();
friend class ClientPrivate;
friend class ChannelPrivate;
void _q_method(const QAMQP::Frame::Method &frame);
Q_PRIVATE_SLOT(d_func(), void _q_heartbeat())
};
void _q_method(const QAMQP::Frame::Method & frame);
Q_PRIVATE_SLOT(pd_func(), void _q_heartbeat());
};
} }
// Include private header so MOC won't complain
#ifdef QAMQP_P_INCLUDE
# include "amqp_connection_p.h"
#endif
#endif // amqp_connection_h__ #endif // amqp_connection_h__

81
src/qamqp/amqp_connection_p.h
View File

@ -9,51 +9,54 @@ class QTimer;
namespace QAMQP namespace QAMQP
{ {
class Client; class Client;
class ClientPrivate; class ClientPrivate;
class ConnectionPrivate class Connection;
{ class ConnectionPrivate
P_DECLARE_PUBLIC(QAMQP::Connection) {
public: public:
enum MethodId enum MethodId {
{ METHOD_ID_ENUM(miStart, 10),
METHOD_ID_ENUM(miStart, 10), METHOD_ID_ENUM(miSecure, 20),
METHOD_ID_ENUM(miSecure, 20), METHOD_ID_ENUM(miTune, 30),
METHOD_ID_ENUM(miTune, 30), METHOD_ID_ENUM(miOpen, 40),
METHOD_ID_ENUM(miOpen, 40), METHOD_ID_ENUM(miClose, 50)
METHOD_ID_ENUM(miClose, 50) };
};
ConnectionPrivate(Connection * q); ConnectionPrivate(Connection *q);
~ConnectionPrivate(); ~ConnectionPrivate();
void init(Client * parent);
void startOk();
void secureOk();
void tuneOk();
void open();
void close(int code, const QString & text, int classId = 0, int methodId = 0);
void closeOk();
void start(const QAMQP::Frame::Method & frame); void init(Client *parent);
void secure(const QAMQP::Frame::Method & frame); void startOk();
void tune(const QAMQP::Frame::Method & frame); void secureOk();
void openOk(const QAMQP::Frame::Method & frame); void tuneOk();
void close(const QAMQP::Frame::Method & frame); void open();
void closeOk(const QAMQP::Frame::Method & frame); void close(int code, const QString & text, int classId = 0, int methodId = 0);
bool _q_method(const QAMQP::Frame::Method & frame); void closeOk();
void _q_heartbeat();
void setQOS(qint32 prefetchSize, quint16 prefetchCount, int channel, bool global); void start(const QAMQP::Frame::Method &frame);
void secure(const QAMQP::Frame::Method &frame);
void tune(const QAMQP::Frame::Method &frame);
void openOk(const QAMQP::Frame::Method &frame);
void close(const QAMQP::Frame::Method &frame);
void closeOk(const QAMQP::Frame::Method &frame);
QPointer<Client> client_; bool _q_method(const QAMQP::Frame::Method &frame);
bool closed_; void _q_heartbeat();
bool connected;
QPointer<QTimer> heartbeatTimer_;
Connection * const pq_ptr; void setQOS(qint32 prefetchSize, quint16 prefetchCount, int channel, bool global);
QAMQP::Frame::TableField customProperty; QPointer<Client> client_;
bool closed_;
bool connected;
QPointer<QTimer> heartbeatTimer_;
QAMQP::Frame::TableField customProperty;
Q_DECLARE_PUBLIC(QAMQP::Connection)
Connection * const q_ptr;
};
};
} }
#endif // amqp_connection_p_h__ #endif // amqp_connection_p_h__

314
src/qamqp/amqp_exchange.cpp
View File

@ -9,268 +9,244 @@ using namespace QAMQP::Frame;
#include <QDebug> #include <QDebug>
#include <QDataStream> #include <QDataStream>
Exchange::Exchange(int channelNumber, Client *parent)
namespace QAMQP : Channel(new ExchangePrivate(this), parent)
{ {
struct ExchangeExceptionCleaner Q_D(QAMQP::Exchange);
{ d->init(channelNumber, parent);
/* this cleans up when the constructor throws an exception */
static inline void cleanup(Exchange *that, ExchangePrivate *d)
{
#ifdef QT_NO_EXCEPTIONS
Q_UNUSED(that);
Q_UNUSED(d);
#else
Q_UNUSED(that);
Q_UNUSED(d);
#endif
}
};
}
Exchange::Exchange(int channelNumber, Client * parent /*= 0*/ )
: Channel(new ExchangePrivate(this))
{
QT_TRY {
pd_func()->init(channelNumber, parent);
} QT_CATCH(...) {
ExchangeExceptionCleaner::cleanup(this, pd_func());
QT_RETHROW;
}
} }
Exchange::~Exchange() Exchange::~Exchange()
{ {
remove(); remove();
} }
void Exchange::onOpen() void Exchange::onOpen()
{ {
P_D(Exchange); Q_D(QAMQP::Exchange);
if(d->delayedDeclare) if (d->delayedDeclare)
{ d->declare();
d->declare();
}
} }
void Exchange::onClose() void Exchange::onClose()
{ {
pd_func()->remove(true, true); Q_D(QAMQP::Exchange);
d->remove(true, true);
} }
Exchange::ExchangeOptions Exchange::option() const Exchange::ExchangeOptions Exchange::option() const
{ {
return pd_func()->options; Q_D(const QAMQP::Exchange);
return d->options;
} }
QString Exchange::type() const QString Exchange::type() const
{ {
return pd_func()->type; Q_D(const QAMQP::Exchange);
return d->type;
} }
void Exchange::declare(const QString &type, ExchangeOptions option , const TableField &arg)
void Exchange::declare(const QString &type, ExchangeOptions option , const TableField & arg)
{ {
P_D(Exchange); Q_D(QAMQP::Exchange);
d->options = option; d->options = option;
d->type = type; d->type = type;
d->arguments = arg; d->arguments = arg;
d->declare(); d->declare();
} }
void Exchange::remove( bool ifUnused /*= true*/, bool noWait /*= true*/ ) void Exchange::remove(bool ifUnused, bool noWait)
{ {
pd_func()->remove(ifUnused, noWait); Q_D(QAMQP::Exchange);
d->remove(ifUnused, noWait);
} }
void Exchange::bind(QAMQP::Queue *queue)
void Exchange::bind( QAMQP::Queue * queue )
{ {
queue->bind(this, pd_func()->name); Q_D(QAMQP::Exchange);
queue->bind(this, d->name);
} }
void Exchange::bind( const QString & queueName ) void Exchange::bind(const QString &queueName)
{ {
Q_UNUSED(queueName); Q_UNUSED(queueName);
qWarning("Not implement"); qWarning("Not implemented");
} }
void Exchange::bind( const QString & queueName, const QString &key ) void Exchange::bind(const QString &queueName, const QString &key)
{ {
Q_UNUSED(queueName); Q_UNUSED(queueName);
Q_UNUSED(key); Q_UNUSED(key);
qWarning("Not implement"); qWarning("Not implemented");
} }
void Exchange::publish( const QString & message, const QString & key, const MessageProperties &prop ) void Exchange::publish(const QString &message, const QString &key, const MessageProperties &prop)
{ {
pd_func()->publish(message.toUtf8(), key, QString::fromLatin1("text.plain"), QVariantHash(), prop); Q_D(QAMQP::Exchange);
d->publish(message.toUtf8(), key, QLatin1String("text.plain"), QVariantHash(), prop);
} }
void Exchange::publish(const QByteArray &message, const QString &key,
void Exchange::publish( const QByteArray & message, const QString & key, const QString &mimeType, const MessageProperties &prop ) const QString &mimeType, const MessageProperties &prop)
{ {
pd_func()->publish(message, key, mimeType, QVariantHash(), prop); Q_D(QAMQP::Exchange);
d->publish(message, key, mimeType, QVariantHash(), prop);
} }
void Exchange::publish( const QByteArray & message, const QString & key, const QVariantHash &headers, const QString &mimeType, const MessageProperties &prop ) void Exchange::publish(const QByteArray &message, const QString &key,
const QVariantHash &headers, const QString &mimeType,
const MessageProperties &prop)
{ {
pd_func()->publish(message, key, mimeType, headers, prop); Q_D(QAMQP::Exchange);
d->publish(message, key, mimeType, headers, prop);
} }
////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////
ExchangePrivate::ExchangePrivate(Exchange * q) ExchangePrivate::ExchangePrivate(Exchange * q)
:ChannelPrivate(q) : ChannelPrivate(q),
, delayedDeclare(false) delayedDeclare(false),
, declared(false) declared(false)
{ {
} }
ExchangePrivate::~ExchangePrivate() ExchangePrivate::~ExchangePrivate()
{ {
} }
bool ExchangePrivate::_q_method(const QAMQP::Frame::Method &frame)
bool ExchangePrivate::_q_method( const QAMQP::Frame::Method & frame )
{ {
if(ChannelPrivate::_q_method(frame)) if (ChannelPrivate::_q_method(frame))
return true; return true;
if(frame.methodClass() != QAMQP::Frame::fcExchange) if (frame.methodClass() != QAMQP::Frame::fcExchange)
return false; return false;
switch(frame.id()) switch(frame.id()) {
{ case miDeclareOk:
case miDeclareOk: declareOk(frame);
declareOk(frame); break;
break; case miDelete:
case miDelete: deleteOk(frame);
deleteOk(frame); break;
break; default:
default: break;
break; }
}
return true; return true;
} }
void ExchangePrivate::declareOk( const QAMQP::Frame::Method & ) void ExchangePrivate::declareOk(const QAMQP::Frame::Method &frame)
{ {
qDebug() << "Declared exchange: " << name; Q_UNUSED(frame)
declared = true; Q_Q(QAMQP::Exchange);
QMetaObject::invokeMethod(pq_func(), "declared"); qDebug() << "Declared exchange: " << name;
declared = true;
QMetaObject::invokeMethod(q, "declared");
} }
void ExchangePrivate::deleteOk( const QAMQP::Frame::Method & ) void ExchangePrivate::deleteOk(const QAMQP::Frame::Method &frame)
{ {
qDebug() << "Deleted exchange: " << name; Q_UNUSED(frame)
declared = false; Q_Q(QAMQP::Exchange);
QMetaObject::invokeMethod(pq_func(), "removed"); qDebug() << "Deleted exchange: " << name;
declared = false;
QMetaObject::invokeMethod(q, "removed");
} }
void ExchangePrivate::declare( ) void ExchangePrivate::declare()
{ {
if(!opened) if (!opened) {
{ delayedDeclare = true;
delayedDeclare = true; return;
return; }
}
if(name.isEmpty()) if (name.isEmpty())
return; return;
QAMQP::Frame::Method frame(QAMQP::Frame::fcExchange, miDeclare); QAMQP::Frame::Method frame(QAMQP::Frame::fcExchange, miDeclare);
frame.setChannel(number); frame.setChannel(number);
QByteArray arguments_; QByteArray arguments_;
QDataStream stream(&arguments_, QIODevice::WriteOnly); QDataStream stream(&arguments_, QIODevice::WriteOnly);
stream << qint16(0); //reserver 1 stream << qint16(0); //reserver 1
writeField('s', stream, name); writeField('s', stream, name);
writeField('s', stream, type); writeField('s', stream, type);
stream << qint8(options); stream << qint8(options);
writeField('F', stream, ExchangePrivate::arguments); writeField('F', stream, ExchangePrivate::arguments);
frame.setArguments(arguments_); frame.setArguments(arguments_);
sendFrame(frame); sendFrame(frame);
delayedDeclare = false; delayedDeclare = false;
} }
void ExchangePrivate::remove( bool ifUnused /*= true*/, bool noWait /*= true*/ ) void ExchangePrivate::remove(bool ifUnused, bool noWait)
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcExchange, miDelete); QAMQP::Frame::Method frame(QAMQP::Frame::fcExchange, miDelete);
frame.setChannel(number); frame.setChannel(number);
QByteArray arguments_; QByteArray arguments_;
QDataStream stream(&arguments_, QIODevice::WriteOnly); QDataStream stream(&arguments_, QIODevice::WriteOnly);
stream << qint16(0); //reserver 1 stream << qint16(0); //reserver 1
writeField('s', stream, name); writeField('s', stream, name);
qint8 flag = 0; qint8 flag = 0;
flag |= (ifUnused ? 0x1 : 0); flag |= (ifUnused ? 0x1 : 0);
flag |= (noWait ? 0x2 : 0); flag |= (noWait ? 0x2 : 0);
stream << flag; //reserver 1 stream << flag; //reserver 1
frame.setArguments(arguments_); frame.setArguments(arguments_);
sendFrame(frame); sendFrame(frame);
} }
void ExchangePrivate::publish( const QByteArray & message, const QString & key, const QString &mimeType /*= QString::fromLatin1("text/plain")*/, const QVariantHash & headers, const Exchange::MessageProperties & prop ) void ExchangePrivate::publish(const QByteArray &message, const QString &key,
const QString &mimeType, const QVariantHash & headers,
const Exchange::MessageProperties &prop)
{ {
QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, bmPublish); QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, bmPublish);
frame.setChannel(number); frame.setChannel(number);
QByteArray arguments_; QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly); QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1 out << qint16(0); //reserver 1
writeField('s', out, name); writeField('s', out, name);
writeField('s', out, key); writeField('s', out, key);
out << qint8(0); out << qint8(0);
frame.setArguments(arguments_); frame.setArguments(arguments_);
sendFrame(frame); sendFrame(frame);
QAMQP::Frame::Content content(QAMQP::Frame::fcBasic);
content.setChannel(number);
content.setProperty(Content::cpContentType, mimeType);
content.setProperty(Content::cpContentEncoding, "utf-8");
content.setProperty(Content::cpHeaders, headers);
content.setProperty(Content::cpMessageId, "0");
QAMQP::Frame::Content content(QAMQP::Frame::fcBasic); Exchange::MessageProperties::ConstIterator i;
content.setChannel(number);
content.setProperty(Content::cpContentType, mimeType);
content.setProperty(Content::cpContentEncoding, "utf-8");
content.setProperty(Content::cpHeaders, headers);
content.setProperty(Content::cpMessageId, "0");
Exchange::MessageProperties::ConstIterator i; for (i = prop.begin(); i != prop.end(); ++i)
content.setProperty(i.key(), i.value());
for(i = prop.begin(); i != prop.end(); ++i) content.setBody(message);
{ sendFrame(content);
content.setProperty(i.key(), i.value());
}
content.setBody(message);
sendFrame(content);
int fullSize = message.size();
for (int sended_ = 0; sended_ < fullSize; sended_+= (FRAME_MAX - 7))
{
QAMQP::Frame::ContentBody body;
QByteArray partition_ = message.mid(sended_, (FRAME_MAX - 7));
body.setChannel(number);
body.setBody(partition_);
sendFrame(body);
}
int fullSize = message.size();
for (int sended_ = 0; sended_ < fullSize; sended_+= (FRAME_MAX - 7)) {
QAMQP::Frame::ContentBody body;
QByteArray partition_ = message.mid(sended_, (FRAME_MAX - 7));
body.setChannel(number);
body.setBody(partition_);
sendFrame(body);
}
} }
void ExchangePrivate::_q_disconnected() void ExchangePrivate::_q_disconnected()
{ {
ChannelPrivate::_q_disconnected(); ChannelPrivate::_q_disconnected();
qDebug() << "Exchange " << name << " disconnected"; qDebug() << "Exchange " << name << " disconnected";
delayedDeclare = false; delayedDeclare = false;
declared = false; declared = false;
} }

98
src/qamqp/amqp_exchange.h
View File

@ -2,63 +2,75 @@
#define amqp_exchange_h__ #define amqp_exchange_h__
#include "amqp_channel.h" #include "amqp_channel.h"
namespace QAMQP namespace QAMQP
{ {
class Client;
class Queue;
class ClientPrivate;
class ExchangePrivate;
using namespace QAMQP::Frame; using namespace QAMQP::Frame;
class Exchange : public Channel
{
Q_OBJECT;
Exchange(int channelNumber = -1, Client * parent = 0);
Q_PROPERTY(QString type READ type); class Client;
Q_PROPERTY(ExchangeOptions option READ option ); class Queue;
class ClientPrivate;
class ExchangePrivate;
class Exchange : public Channel
{
Q_OBJECT
Q_PROPERTY(QString type READ type)
Q_PROPERTY(ExchangeOptions option READ option)
Q_ENUMS(ExchangeOption)
P_DECLARE_PRIVATE(QAMQP::Exchange) public:
Q_DISABLE_COPY(Exchange); enum ExchangeOption {
friend class ClientPrivate; NoOptions = 0x0,
protected: Passive = 0x01,
void onOpen(); Durable = 0x02,
void onClose(); AutoDelete = 0x4,
Internal = 0x8,
NoWait = 0x10
};
Q_DECLARE_FLAGS(ExchangeOptions, ExchangeOption)
public: typedef QHash<QAMQP::Frame::Content::Property, QVariant> MessageProperties;
enum ExchangeOption { virtual ~Exchange();
NoOptions = 0x0,
Passive = 0x01,
Durable = 0x02,
AutoDelete = 0x4,
Internal = 0x8,
NoWait = 0x10
};
Q_DECLARE_FLAGS(ExchangeOptions, ExchangeOption)
typedef QHash<QAMQP::Frame::Content::Property, QVariant> MessageProperties; QString type() const;
ExchangeOptions option() const;
~Exchange(); void declare(const QString &type = QLatin1String("direct"),
ExchangeOptions option = NoOptions,
const TableField &arg = TableField());
void remove(bool ifUnused = true, bool noWait = true);
QString type() const; void bind(QAMQP::Queue *queue);
ExchangeOptions option() const; void bind(const QString &queueName);
void bind(const QString &queueName, const QString &key);
void declare(const QString &type = QString::fromLatin1("direct"), ExchangeOptions option = NoOptions, const TableField & arg = TableField()); void publish(const QString &message, const QString &key,
void remove(bool ifUnused = true, bool noWait = true); const MessageProperties &property = MessageProperties());
void publish(const QByteArray &message, const QString &key,
const QString &mimeType, const MessageProperties &property = MessageProperties());
void publish(const QByteArray &message, const QString &key, const QVariantHash &headers,
const QString &mimeType, const MessageProperties &property = MessageProperties());
void bind(QAMQP::Queue * queue); Q_SIGNALS:
void bind(const QString & queueName); void declared();
void bind(const QString & queueName, const QString &key); void removed();
void publish(const QString & message, const QString & key, const MessageProperties &property = MessageProperties() ); protected:
void publish(const QByteArray & message, const QString & key, const QString &mimeType, const MessageProperties &property = MessageProperties()); void onOpen();
void publish(const QByteArray & message, const QString & key, const QVariantHash &headers, const QString &mimeType, const MessageProperties &property = MessageProperties()); void onClose();
private:
Q_DISABLE_COPY(Exchange)
Q_DECLARE_PRIVATE(QAMQP::Exchange)
Exchange(int channelNumber = -1, Client * parent = 0);
friend class ClientPrivate;
};
Q_SIGNALS:
void declared();
void removed();
};
} }
Q_DECLARE_OPERATORS_FOR_FLAGS(QAMQP::Exchange::ExchangeOptions) Q_DECLARE_OPERATORS_FOR_FLAGS(QAMQP::Exchange::ExchangeOptions)
#endif // amqp_exchange_h__ #endif // amqp_exchange_h__

54
src/qamqp/amqp_exchange_p.h
View File

@ -1,40 +1,44 @@
#include "amqp_channel_p.h" #include "amqp_channel_p.h"
#define METHOD_ID_ENUM(name, id) name = id, name ## Ok #define METHOD_ID_ENUM(name, id) name = id, name ## Ok
namespace QAMQP namespace QAMQP
{ {
using namespace QAMQP::Frame;
class ExchangePrivate: public ChannelPrivate
{
P_DECLARE_PUBLIC(QAMQP::Exchange)
public:
enum MethodId using namespace QAMQP::Frame;
{ class ExchangePrivate: public ChannelPrivate
METHOD_ID_ENUM(miDeclare, 10), {
METHOD_ID_ENUM(miDelete, 20) public:
}; enum MethodId {
METHOD_ID_ENUM(miDeclare, 10),
METHOD_ID_ENUM(miDelete, 20)
};
ExchangePrivate(Exchange * q); ExchangePrivate(Exchange * q);
~ExchangePrivate(); ~ExchangePrivate();
void declare(); void declare();
void remove(bool ifUnused = true, bool noWait = true); void remove(bool ifUnused = true, bool noWait = true);
void declareOk(const QAMQP::Frame::Method & frame); void declareOk(const QAMQP::Frame::Method &frame);
void deleteOk(const QAMQP::Frame::Method & frame); void deleteOk(const QAMQP::Frame::Method &frame);
void publish(const QByteArray & message, const QString & key, const QString &mimeType = QString::fromLatin1("text/plain"), const QVariantHash & headers = QVariantHash(), const Exchange::MessageProperties & properties = Exchange::MessageProperties()); void publish(const QByteArray &message, const QString &key,
const QString &mimeType = QLatin1String("text/plain"),
const QVariantHash &headers = QVariantHash(),
const Exchange::MessageProperties &properties = Exchange::MessageProperties());
QString type; QString type;
Exchange::ExchangeOptions options; Exchange::ExchangeOptions options;
TableField arguments; TableField arguments;
bool _q_method(const QAMQP::Frame::Method & frame); bool _q_method(const QAMQP::Frame::Method &frame);
void _q_disconnected(); void _q_disconnected();
bool delayedDeclare; bool delayedDeclare;
bool declared; bool declared;
Q_DECLARE_PUBLIC(QAMQP::Exchange)
};
};
} }

1043
src/qamqp/amqp_frame.cpp
View File

File diff suppressed because it is too large Load Diff

690
src/qamqp/amqp_frame.h
View File

@ -20,402 +20,402 @@
#define AMQP_BASIC_APP_ID_FLAG (1 << 3) #define AMQP_BASIC_APP_ID_FLAG (1 << 3)
#define AMQP_BASIC_CLUSTER_ID_FLAG (1 << 2) #define AMQP_BASIC_CLUSTER_ID_FLAG (1 << 2)
/*! /**
Library namespace * Library namespace
@namespace QAMQP * @namespace QAMQP
*/ */
namespace QAMQP namespace QAMQP
{ {
class QueuePrivate;
/*!
Frame namespace
@namespace Frame
*/
namespace Frame
{
typedef quint16 channel_t;
/*!
@brief Header size in bytes
*/
static const qint64 HEADER_SIZE = 7;
/*!
@brief Frame end indicator size in bytes
*/
static const qint64 FRAME_END_SIZE = 1;
/*! class QueuePrivate;
@brief Frame end marker
*/
static const quint8 FRAME_END = 0xCE;
/*! namespace Frame
@brief Frame type {
*/ typedef quint16 channel_t;
enum Type
{
ftMethod = 1, /*!< Used define method frame */
ftHeader = 2, /*!< Used define content header frame */
ftBody = 3, /*!< Used define content body frame */
ftHeartbeat = 8 /*!< Used define heartbeat frame */
};
/*! /*
@brief Frame method class * @brief Header size in bytes
@enum MethodClass */
*/ static const qint64 HEADER_SIZE = 7;
enum MethodClass
{
fcConnection = 10, /*!< Define class of methods related to connection */
fcChannel = 20, /*!< Define class of methods related to channel */
fcExchange = 40, /*!< Define class of methods related to exchange */
fcQueue = 50, /*!< Define class of methods related to queue */
fcBasic = 60, /*!< Define class of methods related to basic command */
fcTx = 90,
};
struct decimal /*
{ * @brief Frame end indicator size in bytes
qint8 scale; */
quint32 value; static const qint64 FRAME_END_SIZE = 1;
}; /*
* @brief Frame end marker
*/
static const quint8 FRAME_END = 0xCE;
/*! /*
@brief Definition implementation of TableField type * @brief Frame type
@detailed Define implementation TableField type in builtin Qt types. Key contains field name, value contains field data. */
It can by any type witch support serialization in AMQP types. enum Type
*/ {
typedef QHash<QString, QVariant> TableField; ftMethod = 1, /*!< Used define method frame */
ftHeader = 2, /*!< Used define content header frame */
ftBody = 3, /*!< Used define content body frame */
ftHeartbeat = 8 /*!< Used define heartbeat frame */
};
QDataStream & serialize( QDataStream & stream, const QAMQP::Frame::TableField & f ); /*
QDataStream & deserialize( QDataStream & stream, QAMQP::Frame::TableField & f ); * @brief Frame method class
QVariant readField( qint8 valueType, QDataStream &s ); * @enum MethodClass
void writeField( QDataStream &s, const QVariant & value ); */
void writeField( qint8 valueType, QDataStream &s, const QVariant & value, bool withType = false ); enum MethodClass
void print( const QAMQP::Frame::TableField & f ); {
fcConnection = 10, // Define class of methods related to connection
fcChannel = 20, // Define class of methods related to channel
fcExchange = 40, // Define class of methods related to exchange
fcQueue = 50, // Define class of methods related to queue
fcBasic = 60, // Define class of methods related to basic command
fcTx = 90,
};
/*! struct decimal
@brief Base class for any frames. {
@detailed Implement main methods for serialize and deserialize raw frame data. qint8 scale;
All frames start with a 7-octet header composed of a type field (octet), a channel field (short integer) and a quint32 value;
size field (long integer): };
@code Frame struct
0 1 3 7 size+7 size+8
+------+---------+---------+ +-------------+ +-----------+
| type | channel | size | | payload | | frame-end |
+------+---------+---------+ +-------------+ +-----------+
@endcode
octet short long 'size' octets octet
*/
class Base
{
public:
/*!
Base class constructor.
@detailed Construct frame class for sending.
@param type Define type of constructed frame.
*/
Base(Type type);
/*! /*
Base class constructor. * @brief Definition implementation of TableField type
@detailed Construct frame class from received raw data. * @detailed Define implementation TableField type in builtin Qt types. Key contains field name, value contains field data.
@param raw Data stream for reading source data. * It can by any type witch support serialization in AMQP types.
*/ */
Base(QDataStream& raw); typedef QHash<QString, QVariant> TableField;
/*! QDataStream & serialize( QDataStream & stream, const QAMQP::Frame::TableField & f );
Base class virtual destructor QDataStream & deserialize( QDataStream & stream, QAMQP::Frame::TableField & f );
*/ QVariant readField( qint8 valueType, QDataStream &s );
virtual ~Base(); void writeField( QDataStream &s, const QVariant & value );
void writeField( qint8 valueType, QDataStream &s, const QVariant & value, bool withType = false );
void print( const QAMQP::Frame::TableField & f );
/*! /*
Frame type * @brief Base class for any frames.
@detailed Return type of current frame. * @detailed Implement main methods for serialize and deserialize raw frame data.
*/ * All frames start with a 7-octet header composed of a type field (octet), a channel field (short integer) and a
Type type() const; * size field (long integer):
* @code Frame struct
* 0 1 3 7 size+7 size+8
* +------+---------+---------+ +-------------+ +-----------+
* | type | channel | size | | payload | | frame-end |
* +------+---------+---------+ +-------------+ +-----------+
* @endcode
* octet short long 'size' octets octet
*/
class Base
{
public:
/*
* Base class constructor.
* @detailed Construct frame class for sending.
* @param type Define type of constructed frame.
*/
Base(Type type);
/*! /*
Set number of associated channel. * Base class constructor.
@param channel Number of channel. * @detailed Construct frame class from received raw data.
@sa channel() * @param raw Data stream for reading source data.
*/ */
void setChannel(qint16 channel); Base(QDataStream& raw);
/*! /*
Return number of associated channel. * Base class virtual destructor
@sa setChannel() */
*/ virtual ~Base();
qint16 channel() const;
/*! /*
Return size of frame. * Frame type
*/ * @detailed Return type of current frame.
virtual qint32 size() const; */
Type type() const;
/*! /*
Output frame to stream. * Set number of associated channel.
@param stream Stream for serilize frame. * @param channel Number of channel.
*/ * @sa channel()
void toStream(QDataStream & stream) const; */
void setChannel(qint16 channel);
protected: /*
void writeHeader(QDataStream & stream) const; * Return number of associated channel.
virtual void writePayload(QDataStream & stream) const; * @sa setChannel()
void writeEnd(QDataStream & stream) const; */
qint16 channel() const;
void readHeader(QDataStream & stream); /*
virtual void readPayload(QDataStream & stream); * Return size of frame.
void readEnd(QDataStream & stream); */
virtual qint32 size() const;
qint32 size_; /*
private: * Output frame to stream.
qint8 type_; * @param stream Stream for serilize frame.
*/
void toStream(QDataStream &stream) const;
qint16 channel_; protected:
void writeHeader(QDataStream &stream) const;
virtual void writePayload(QDataStream &stream) const;
void writeEnd(QDataStream &stream) const;
}; void readHeader(QDataStream &stream);
virtual void readPayload(QDataStream &stream);
void readEnd(QDataStream &stream);
/*! qint32 size_;
@brief Class for working with method frames.
@detailed Implement main methods for serialize and deserialize raw method frame data.
Method frame bodies consist of an invariant list of data fields, called "arguments". All method bodies start
with identifier numbers for the class and method:
@code Frame struct
0 2 4
+----------+-----------+-------------- - -
| class-id | method-id | arguments...
+----------+-----------+-------------- - -
short short ...
@endcode
*/
class Method : public Base
{
public:
/*!
Method class constructor.
@detailed Construct frame class for sending.
*/
Method();
/*! private:
Method class constructor. qint8 type_;
@detailed Construct frame class for sending. qint16 channel_;
@param methodClass Define method class id of constructed frame.
@param id Define method id of constructed frame.
*/
Method(MethodClass methodClass, qint16 id);
/*! };
Method class constructor.
@detailed Construct frame class from received raw data.
@param raw Data stream for reading source data.
*/
Method(QDataStream& raw);
/*! /*
Method class type. * @brief Class for working with method frames.
*/ * @detailed Implement main methods for serialize and deserialize raw method frame data.
MethodClass methodClass() const; * Method frame bodies consist of an invariant list of data fields, called "arguments". All method bodies start
* with identifier numbers for the class and method:
* @code Frame struct
* 0 2 4
* +----------+-----------+-------------- - -
* | class-id | method-id | arguments...
* +----------+-----------+-------------- - -
* short short ...
* @endcode
*/
class Method : public Base
{
public:
/*
* Method class constructor.
* @detailed Construct frame class for sending.
*/
Method();
/*! /*
Method id. * Method class constructor.
*/ * @detailed Construct frame class for sending.
qint16 id() const; * @param methodClass Define method class id of constructed frame.
qint32 size() const; * @param id Define method id of constructed frame.
*/
Method(MethodClass methodClass, qint16 id);
/*! /*
Set arguments for method. * Method class constructor.
@param data Serialized method arguments. * @detailed Construct frame class from received raw data.
@sa arguments * @param raw Data stream for reading source data.
*/ */
void setArguments(const QByteArray & data); Method(QDataStream &raw);
/*! /*
Return arguments for method. * Method class type.
@sa setArguments */
*/ MethodClass methodClass() const;
QByteArray arguments() const;
protected: /*
void writePayload(QDataStream & stream) const; * Method id.
void readPayload(QDataStream & stream); */
short methodClass_; qint16 id() const;
qint16 id_; qint32 size() const;
QByteArray arguments_;
}; /*
* Set arguments for method.
* @param data Serialized method arguments.
* @sa arguments
*/
void setArguments(const QByteArray &data);
/*
* Return arguments for method.
* @sa setArguments
*/
QByteArray arguments() const;
protected:
void writePayload(QDataStream & stream) const;
void readPayload(QDataStream & stream);
short methodClass_;
qint16 id_;
QByteArray arguments_;
};
/*! /*
@brief Class for working with content frames. * @brief Class for working with content frames.
@detailed Implement main methods for serialize and deserialize raw content frame data. * @detailed Implement main methods for serialize and deserialize raw content frame data.
A content header payload has this format: * A content header payload has this format:
@code Frame struct * @code Frame struct
+----------+--------+-----------+----------------+------------- - - * +----------+--------+-----------+----------------+------------- - -
| class-id | weight | body size | property flags | property list... * | class-id | weight | body size | property flags | property list...
+----------+--------+-----------+----------------+------------- - - * +----------+--------+-----------+----------------+------------- - -
short short long long short remainder... * short short long long short remainder...
@endcode * @endcode
*
* | Property | Description |
* | ---------- | ----------- |
* |cpContentType | MIME content type |
* | ocpContentEncoding | MIME content encoding |
* | cpHeaders | message header field table |
* | cpDeliveryMode| nonpersistent (1) or persistent (2) |
* | cpPriority | message priority, 0 to 9 |
* | cpCorrelationId | application correlation identifier |
* | cpReplyTo | address to reply to |
* | cpExpiration | message expiration specification |
* | cpMessageId | application message identifier |
* | cpTimestamp | message timestamp |
* | cpType | message type name |
* | cpUserId | creating user id |
* | cpAppId | creating application id |
* | cpClusterID| cluster ID |
*
* Default property:
* @sa setProperty
* @sa property
*/
class Content : public Base
{
friend class QAMQP::QueuePrivate;
| Property | Description | public:
| ---------- | ----------- | /*
|cpContentType | MIME content type | * Default content frame property
| ocpContentEncoding | MIME content encoding | */
| cpHeaders | message header field table | enum Property
| cpDeliveryMode| nonpersistent (1) or persistent (2) | {
| cpPriority | message priority, 0 to 9 | cpContentType = AMQP_BASIC_CONTENT_TYPE_FLAG,
| cpCorrelationId | application correlation identifier | cpContentEncoding = AMQP_BASIC_CONTENT_ENCODING_FLAG,
| cpReplyTo | address to reply to | cpHeaders = AMQP_BASIC_HEADERS_FLAG,
| cpExpiration | message expiration specification | cpDeliveryMode = AMQP_BASIC_DELIVERY_MODE_FLAG,
| cpMessageId | application message identifier | cpPriority = AMQP_BASIC_PRIORITY_FLAG,
| cpTimestamp | message timestamp | cpCorrelationId = AMQP_BASIC_CORRELATION_ID_FLAG,
| cpType | message type name | cpReplyTo = AMQP_BASIC_REPLY_TO_FLAG,
| cpUserId | creating user id | cpExpiration = AMQP_BASIC_EXPIRATION_FLAG,
| cpAppId | creating application id | cpMessageId = AMQP_BASIC_MESSAGE_ID_FLAG,
| cpClusterID| cluster ID | cpTimestamp = AMQP_BASIC_TIMESTAMP_FLAG,
cpType = AMQP_BASIC_TYPE_FLAG,
cpUserId = AMQP_BASIC_USER_ID_FLAG,
cpAppId = AMQP_BASIC_APP_ID_FLAG,
cpClusterID = AMQP_BASIC_CLUSTER_ID_FLAG
};
Q_DECLARE_FLAGS(Properties, Property)
Default property: /*
@sa setProperty * Content class constructor.
@sa property * @detailed Construct frame content header class for sending.
*/
Content();
*/ /*
class Content : public Base * Content class constructor.
{ * @detailed Construct frame content header class for sending.
friend class QAMQP::QueuePrivate; * @param methodClass Define method class id of constructed frame.
public: */
Content(MethodClass methodClass);
/*! Default content frame property /*
* Content class constructor.
* @detailed Construct frame content header class for sending.
* @param raw Data stream for reading source data.
*/
Content(QDataStream &raw);
*/ /*
enum Property * Method class type.
{ */
cpContentType = AMQP_BASIC_CONTENT_TYPE_FLAG, MethodClass methodClass() const;
cpContentEncoding = AMQP_BASIC_CONTENT_ENCODING_FLAG, qint32 size() const;
cpHeaders = AMQP_BASIC_HEADERS_FLAG,
cpDeliveryMode = AMQP_BASIC_DELIVERY_MODE_FLAG,
cpPriority = AMQP_BASIC_PRIORITY_FLAG,
cpCorrelationId = AMQP_BASIC_CORRELATION_ID_FLAG,
cpReplyTo = AMQP_BASIC_REPLY_TO_FLAG,
cpExpiration = AMQP_BASIC_EXPIRATION_FLAG,
cpMessageId = AMQP_BASIC_MESSAGE_ID_FLAG,
cpTimestamp = AMQP_BASIC_TIMESTAMP_FLAG,
cpType = AMQP_BASIC_TYPE_FLAG,
cpUserId = AMQP_BASIC_USER_ID_FLAG,
cpAppId = AMQP_BASIC_APP_ID_FLAG,
cpClusterID = AMQP_BASIC_CLUSTER_ID_FLAG
};
Q_DECLARE_FLAGS(Properties, Property);
/*! /*
Content class constructor. * Set default content header property
@detailed Construct frame content header class for sending. * @param prop Any default content header property
*/ * @param value Associated data
Content(); */
void setProperty(Property prop, const QVariant &value);
/*! /*
Content class constructor. * Return associated with property value
@detailed Construct frame content header class for sending. * @param prop Any default content header property
@param methodClass Define method class id of constructed frame. */
*/ QVariant property(Property prop) const;
Content(MethodClass methodClass);
/*! void setBody(const QByteArray &data);
Content class constructor. QByteArray body() const;
@detailed Construct frame content header class for sending. qlonglong bodySize() const;
@param raw Data stream for reading source data.
*/
Content(QDataStream& raw);
/*! protected:
Method class type. void writePayload(QDataStream &stream) const;
*/ void readPayload(QDataStream &stream);
MethodClass methodClass() const; short methodClass_;
qint32 size() const; qint16 id_;
QByteArray body_;
mutable QByteArray buffer_;
QHash<int, QVariant> properties_;
qlonglong bodySize_;
};
/*! class ContentBody : public Base
Set default content header property {
@param prop Any default content header property public:
@param value Associated data ContentBody();
*/ ContentBody(QDataStream &raw);
void setProperty(Property prop, const QVariant & value);
/*! void setBody(const QByteArray &data);
Return associated with property value QByteArray body() const;
@param prop Any default content header property
*/
QVariant property(Property prop) const;
qint32 size() const;
protected:
void writePayload(QDataStream &stream) const;
void readPayload(QDataStream &stream);
void setBody(const QByteArray & data); private:
QByteArray body() const; QByteArray body_;
qlonglong bodySize() const; };
protected: /*
void writePayload(QDataStream & stream) const; * @brief Class for working with heartbeat frames.
void readPayload(QDataStream & stream); * @detailed Implement frame for heartbeat send.
short methodClass_; */
qint16 id_; class Heartbeat : public Base
QByteArray body_; {
mutable QByteArray buffer_; public:
QHash<int, QVariant> properties_; /*
qlonglong bodySize_; * Heartbeat class constructor.
}; * @detailed Construct frame class for sending.
*/
Heartbeat();
class ContentBody : public Base protected:
{ void writePayload(QDataStream &stream) const;
public: void readPayload(QDataStream &stream);
ContentBody(); };
ContentBody(QDataStream& raw);
void setBody(const QByteArray & data);
QByteArray body() const;
qint32 size() const;
protected:
void writePayload(QDataStream & stream) const;
void readPayload(QDataStream & stream);
private: class MethodHandler
QByteArray body_; {
}; public:
virtual void _q_method(const QAMQP::Frame::Method &frame) = 0;
};
/*! class ContentHandler
@brief Class for working with heartbeat frames. {
@detailed Implement frame for heartbeat send. public:
*/ virtual void _q_content(const QAMQP::Frame::Content & frame) = 0;
class Heartbeat : public Base };
{
public:
/*!
Heartbeat class constructor.
@detailed Construct frame class for sending.
*/
Heartbeat();
protected: class ContentBodyHandler
void writePayload(QDataStream & stream) const; {
void readPayload(QDataStream & stream); public:
}; virtual void _q_body(const QAMQP::Frame::ContentBody & frame) = 0;
};
class MethodHandler }
{
public:
virtual void _q_method(const QAMQP::Frame::Method & frame) = 0;
};
class ContentHandler
{
public:
virtual void _q_content(const QAMQP::Frame::Content & frame) = 0;
};
class ContentBodyHandler
{
public:
virtual void _q_body(const QAMQP::Frame::ContentBody & frame) = 0;
};
}
} }
Q_DECLARE_METATYPE(QAMQP::Frame::decimal); Q_DECLARE_METATYPE(QAMQP::Frame::decimal)
Q_DECLARE_METATYPE(QAMQP::Frame::TableField); Q_DECLARE_METATYPE(QAMQP::Frame::TableField)
#endif // amqp_frame_h__ #endif // amqp_frame_h__

19
src/qamqp/amqp_global.h
View File

@ -1,8 +1,6 @@
#ifndef qamqp_global_h__ #ifndef qamqp_global_h__
#define qamqp_global_h__ #define qamqp_global_h__
#include <QtCore/qglobal.h>
#define QAMQP_P_INCLUDE #define QAMQP_P_INCLUDE
#define AMQPSCHEME "amqp" #define AMQPSCHEME "amqp"
#define AMQPSSCHEME "amqps" #define AMQPSSCHEME "amqps"
@ -15,23 +13,6 @@
#define QAMQP_VERSION "0.2.0" #define QAMQP_VERSION "0.2.0"
#define AMQP_CONNECTION_FORCED 320 #define AMQP_CONNECTION_FORCED 320
#define P_DECLARE_PRIVATE(Class) \
friend class Class##Private; \
inline Class##Private* pd_func() { return reinterpret_cast<Class##Private *>(this->pd_ptr); } \
inline const Class##Private* pd_func() const { return reinterpret_cast<const Class##Private *>(this->pd_ptr); }
#define P_DECLARE_PUBLIC(Class) \
inline Class* pq_func() { return static_cast<Class *>(this->pq_ptr); } \
inline const Class* pq_func() const { return static_cast<const Class *>(this->pq_ptr); } \
friend class Class;
#define P_D(Class) Class##Private * const d = this->pd_func()
#define P_Q(Class) Class * const q = this->pq_func()
#endif // qamqp_global_h__ #endif // qamqp_global_h__

15
src/qamqp/amqp_message.cpp Normal file
View File

@ -0,0 +1,15 @@
#include "amqp_message.h"
using namespace QAMQP;
Message::Message()
{
qDebug("Message create");
leftSize = 0;
deliveryTag = 0;
}
Message::~Message()
{
qDebug("Message release");
}

41
src/qamqp/amqp_message.h
View File

@ -5,29 +5,24 @@
namespace QAMQP namespace QAMQP
{ {
struct Message
{
Message()
{
qDebug("Message create");
leftSize = 0;
deliveryTag = 0;
}
~Message()
{
qDebug("Message release");
}
typedef QAMQP::Frame::Content::Property MessageProperty;
Q_DECLARE_FLAGS(MessageProperties, MessageProperty);
qlonglong deliveryTag; struct Message
QByteArray payload; {
QHash<MessageProperty, QVariant> property; Message();
QAMQP::Frame::TableField headers; virtual ~Message();
QString routeKey;
QString exchangeName; typedef QAMQP::Frame::Content::Property MessageProperty;
int leftSize; Q_DECLARE_FLAGS(MessageProperties, MessageProperty)
};
qlonglong deliveryTag;
QByteArray payload;
QHash<MessageProperty, QVariant> property;
QAMQP::Frame::TableField headers;
QString routeKey;
QString exchangeName;
int leftSize;
};
typedef QSharedPointer<QAMQP::Message> MessagePtr;
typedef QSharedPointer<QAMQP::Message> MessagePtr;
} }

351
src/qamqp/amqp_network.cpp
View File

@ -3,274 +3,241 @@
#include <QTimer> #include <QTimer>
#include <QtEndian> #include <QtEndian>
QAMQP::Network::Network( QObject * parent /*= 0*/ ):QObject(parent) QAMQP::Network::Network( QObject * parent)
: QObject(parent)
{ {
qRegisterMetaType<QAMQP::Frame::Method>("QAMQP::Frame::Method"); qRegisterMetaType<QAMQP::Frame::Method>("QAMQP::Frame::Method");
buffer_.reserve(Frame::HEADER_SIZE); buffer_.reserve(Frame::HEADER_SIZE);
timeOut_ = 1000; timeOut_ = 1000;
connect_ = false; connect_ = false;
initSocket(false); initSocket(false);
} }
QAMQP::Network::~Network() QAMQP::Network::~Network()
{ {
disconnect(); disconnect();
} }
void QAMQP::Network::connectTo( const QString & host, quint32 port ) void QAMQP::Network::connectTo(const QString & host, quint16 port)
{ {
if(!socket_) if(!socket_) {
{ qWarning("AMQP: Socket didn't create.");
qWarning("AMQP: Socket didn't create."); return;
return; }
}
QString h(host);
int p(port);
connect_ = true;
if(host.isEmpty())
h = lastHost_ ;
if(port == 0)
p = lastPort_;
if (isSsl()) QString h(host);
{ int p(port);
connect_ = true;
if (host.isEmpty())
h = lastHost_ ;
if (port == 0)
p = lastPort_;
if (isSsl()) {
#ifndef QT_NO_SSL #ifndef QT_NO_SSL
static_cast<QSslSocket *>(socket_.data())->connectToHostEncrypted(h, p); static_cast<QSslSocket *>(socket_.data())->connectToHostEncrypted(h, p);
#else #else
qWarning("AMQP: You library has builded with QT_NO_SSL option."); qWarning("AMQP: You library has builded with QT_NO_SSL option.");
#endif #endif
} else { } else {
socket_->connectToHost(h, p); socket_->connectToHost(h, p);
} }
lastHost_ = h; lastHost_ = h;
lastPort_ = p; lastPort_ = p;
} }
void QAMQP::Network::disconnect() void QAMQP::Network::disconnect()
{ {
connect_ = false; connect_ = false;
if(socket_) if (socket_)
socket_->close(); socket_->close();
} }
void QAMQP::Network::error( QAbstractSocket::SocketError socketError ) void QAMQP::Network::error(QAbstractSocket::SocketError socketError)
{ {
if(timeOut_ == 0) if (timeOut_ == 0) {
{ timeOut_ = 1000;
timeOut_ = 1000; } else {
} else { if (timeOut_ < 120000)
if(timeOut_ < 120000) timeOut_ *= 5;
{ }
timeOut_ *= 5;
}
}
Q_UNUSED(socketError); switch(socketError) {
switch(socketError) case QAbstractSocket::ConnectionRefusedError:
{ case QAbstractSocket::RemoteHostClosedError:
case QAbstractSocket::ConnectionRefusedError: case QAbstractSocket::SocketTimeoutError:
case QAbstractSocket::RemoteHostClosedError: case QAbstractSocket::NetworkError:
case QAbstractSocket::SocketTimeoutError: case QAbstractSocket::ProxyConnectionClosedError:
case QAbstractSocket::NetworkError: case QAbstractSocket::ProxyConnectionRefusedError:
case QAbstractSocket::ProxyConnectionClosedError: case QAbstractSocket::ProxyConnectionTimeoutError:
case QAbstractSocket::ProxyConnectionRefusedError:
case QAbstractSocket::ProxyConnectionTimeoutError:
default: default:
qWarning() << "AMQP: Socket Error: " << socket_->errorString(); qWarning() << "AMQP: Socket Error: " << socket_->errorString();
break; break;
} }
if( autoReconnect_ && connect_ )
{
QTimer::singleShot(timeOut_, this, SLOT(connectTo()));
}
if (autoReconnect_ && connect_)
QTimer::singleShot(timeOut_, this, SLOT(connectTo()));
} }
void QAMQP::Network::readyRead() void QAMQP::Network::readyRead()
{ {
while(socket_->bytesAvailable() >= Frame::HEADER_SIZE) while (socket_->bytesAvailable() >= Frame::HEADER_SIZE) {
{ char *headerData = buffer_.data();
char* headerData = buffer_.data(); socket_->peek(headerData, Frame::HEADER_SIZE);
socket_->peek(headerData, Frame::HEADER_SIZE); const quint32 payloadSize = qFromBigEndian<quint32>(*(quint32*)&headerData[3]);
const quint32 payloadSize = qFromBigEndian<quint32>(*(quint32*)&headerData[3]); const qint64 readSize = Frame::HEADER_SIZE+payloadSize+Frame::FRAME_END_SIZE;
const qint64 readSize = Frame::HEADER_SIZE+payloadSize+Frame::FRAME_END_SIZE;
if(socket_->bytesAvailable() >= readSize)
{
buffer_.resize(readSize);
socket_->read(buffer_.data(), readSize);
const char* bufferData = buffer_.constData();
const quint8 type = *(quint8*)&bufferData[0];
const quint8 magic = *(quint8*)&bufferData[Frame::HEADER_SIZE+payloadSize];
if(magic != QAMQP::Frame::FRAME_END)
{
qWarning() << "Wrong end frame";
}
QDataStream streamB(&buffer_, QIODevice::ReadOnly); if (socket_->bytesAvailable() >= readSize) {
switch(QAMQP::Frame::Type(type)) buffer_.resize(readSize);
{ socket_->read(buffer_.data(), readSize);
case QAMQP::Frame::ftMethod: const char* bufferData = buffer_.constData();
{ const quint8 type = *(quint8*)&bufferData[0];
QAMQP::Frame::Method frame(streamB); const quint8 magic = *(quint8*)&bufferData[Frame::HEADER_SIZE+payloadSize];
if(frame.methodClass() == QAMQP::Frame::fcConnection) if (magic != QAMQP::Frame::FRAME_END)
{ qWarning() << "Wrong end frame";
m_pMethodHandlerConnection->_q_method(frame);
} QDataStream streamB(&buffer_, QIODevice::ReadOnly);
else switch(QAMQP::Frame::Type(type)) {
{ case QAMQP::Frame::ftMethod:
foreach(Frame::MethodHandler* pMethodHandler, m_methodHandlersByChannel[frame.channel()]) {
{ QAMQP::Frame::Method frame(streamB);
pMethodHandler->_q_method(frame); if (frame.methodClass() == QAMQP::Frame::fcConnection) {
} m_pMethodHandlerConnection->_q_method(frame);
} } else {
} foreach(Frame::MethodHandler* pMethodHandler, m_methodHandlersByChannel[frame.channel()])
break; pMethodHandler->_q_method(frame);
case QAMQP::Frame::ftHeader: }
{ }
QAMQP::Frame::Content frame(streamB); break;
foreach(Frame::ContentHandler* pMethodHandler, m_contentHandlerByChannel[frame.channel()]) case QAMQP::Frame::ftHeader:
{ {
pMethodHandler->_q_content(frame); QAMQP::Frame::Content frame(streamB);
} foreach(Frame::ContentHandler* pMethodHandler, m_contentHandlerByChannel[frame.channel()])
} pMethodHandler->_q_content(frame);
break; }
case QAMQP::Frame::ftBody: break;
{ case QAMQP::Frame::ftBody:
QAMQP::Frame::ContentBody frame(streamB); {
foreach(Frame::ContentBodyHandler* pMethodHandler, m_bodyHandlersByChannel[frame.channel()]) QAMQP::Frame::ContentBody frame(streamB);
{ foreach(Frame::ContentBodyHandler* pMethodHandler, m_bodyHandlersByChannel[frame.channel()])
pMethodHandler->_q_body(frame); pMethodHandler->_q_body(frame);
} }
} break;
break; case QAMQP::Frame::ftHeartbeat:
case QAMQP::Frame::ftHeartbeat: qDebug("AMQP: Heartbeat");
{ break;
qDebug("AMQP: Heartbeat"); default:
} qWarning() << "AMQP: Unknown frame type: " << type;
break; }
default: } else {
qWarning() << "AMQP: Unknown frame type: " << type; break;
} }
} }
else
{
break;
}
}
} }
void QAMQP::Network::sendFrame( const QAMQP::Frame::Base & frame ) void QAMQP::Network::sendFrame(const QAMQP::Frame::Base &frame)
{ {
if(socket_->state() == QAbstractSocket::ConnectedState) if (socket_->state() == QAbstractSocket::ConnectedState) {
{ QDataStream stream(socket_);
QDataStream stream(socket_); frame.toStream(stream);
frame.toStream(stream); }
}
} }
bool QAMQP::Network::isSsl() const bool QAMQP::Network::isSsl() const
{ {
if(socket_) if (socket_)
{ return QString(socket_->metaObject()->className()).compare("QSslSocket", Qt::CaseInsensitive) == 0;
return QString(socket_->metaObject()->className()).compare( "QSslSocket", Qt::CaseInsensitive) == 0; return false;
}
return false;
} }
void QAMQP::Network::setSsl( bool value ) void QAMQP::Network::setSsl(bool value)
{ {
initSocket(value); initSocket(value);
} }
void QAMQP::Network::initSocket( bool ssl /*= false*/ ) void QAMQP::Network::initSocket(bool ssl)
{ {
if(socket_) if (socket_) {
delete socket_; socket_->deleteLater();
socket_ = 0;
}
if(ssl) if (ssl) {
{
#ifndef QT_NO_SSL #ifndef QT_NO_SSL
socket_ = new QSslSocket(this); socket_ = new QSslSocket(this);
QSslSocket * ssl_= static_cast<QSslSocket*> (socket_.data()); QSslSocket * ssl_= static_cast<QSslSocket*> (socket_.data());
ssl_->setProtocol(QSsl::AnyProtocol); ssl_->setProtocol(QSsl::AnyProtocol);
connect(socket_, SIGNAL(sslErrors(const QList<QSslError> &)), this, SLOT(sslErrors())); connect(socket_, SIGNAL(sslErrors(const QList<QSslError> &)), this, SLOT(sslErrors()));
connect(socket_, SIGNAL(connected()), this, SLOT(conectionReady()));
connect(socket_, SIGNAL(connected()), this, SLOT(conectionReady()));
#else #else
qWarning("AMQP: You library has builded with QT_NO_SSL option."); qWarning("AMQP: You library has builded with QT_NO_SSL option.");
#endif #endif
} else { } else {
socket_ = new QTcpSocket(this); socket_ = new QTcpSocket(this);
connect(socket_, SIGNAL(connected()), this, SLOT(conectionReady())); connect(socket_, SIGNAL(connected()), this, SLOT(conectionReady()));
} }
if(socket_) if (socket_) {
{ connect(socket_, SIGNAL(disconnected()), this, SIGNAL(disconnected()));
connect(socket_, SIGNAL(disconnected()), this, SIGNAL(disconnected())); connect(socket_, SIGNAL(readyRead()), this, SLOT(readyRead()));
connect(socket_, SIGNAL(readyRead()), this, SLOT(readyRead())); connect(socket_, SIGNAL(error(QAbstractSocket::SocketError)),
connect(socket_, SIGNAL(error(QAbstractSocket::SocketError)), this, SLOT(error(QAbstractSocket::SocketError))); this, SLOT(error(QAbstractSocket::SocketError)));
} }
} }
void QAMQP::Network::sslErrors()
void QAMQP::Network::sslErrors( )
{ {
#ifndef QT_NO_SSL #ifndef QT_NO_SSL
static_cast<QSslSocket*>(socket_.data())->ignoreSslErrors(); static_cast<QSslSocket*>(socket_.data())->ignoreSslErrors();
#endif #endif
} }
void QAMQP::Network::conectionReady() void QAMQP::Network::conectionReady()
{ {
emit connected(); Q_EMIT connected();
timeOut_ = 0; timeOut_ = 0;
char header_[8] = {'A', 'M', 'Q', 'P', 0,0,9,1};
socket_->write(header_, 8); char header_[8] = {'A', 'M', 'Q', 'P', 0,0,9,1};
socket_->write(header_, 8);
} }
bool QAMQP::Network::autoReconnect() const bool QAMQP::Network::autoReconnect() const
{ {
return autoReconnect_; return autoReconnect_;
} }
void QAMQP::Network::setAutoReconnect( bool value ) void QAMQP::Network::setAutoReconnect(bool value)
{ {
autoReconnect_ = value; autoReconnect_ = value;
} }
QAbstractSocket::SocketState QAMQP::Network::state() const QAbstractSocket::SocketState QAMQP::Network::state() const
{ {
if(socket_) if (socket_)
{ return socket_->state();
return socket_->state(); return QAbstractSocket::UnconnectedState;
} else {
return QAbstractSocket::UnconnectedState;
}
} }
void QAMQP::Network::setMethodHandlerConnection(Frame::MethodHandler* pMethodHandlerConnection) void QAMQP::Network::setMethodHandlerConnection(Frame::MethodHandler *pMethodHandlerConnection)
{ {
m_pMethodHandlerConnection = pMethodHandlerConnection; m_pMethodHandlerConnection = pMethodHandlerConnection;
} }
void QAMQP::Network::addMethodHandlerForChannel(Channel channel, Frame::MethodHandler* pHandler) void QAMQP::Network::addMethodHandlerForChannel(Channel channel, Frame::MethodHandler *pHandler)
{ {
m_methodHandlersByChannel[channel].append(pHandler); m_methodHandlersByChannel[channel].append(pHandler);
} }
void QAMQP::Network::addContentHandlerForChannel(Channel channel, Frame::ContentHandler* pHandler) void QAMQP::Network::addContentHandlerForChannel(Channel channel, Frame::ContentHandler *pHandler)
{ {
m_contentHandlerByChannel[channel].append(pHandler); m_contentHandlerByChannel[channel].append(pHandler);
} }
void QAMQP::Network::addContentBodyHandlerForChannel(Channel channel, Frame::ContentBodyHandler* pHandler) void QAMQP::Network::addContentBodyHandlerForChannel(Channel channel, Frame::ContentBodyHandler *pHandler)
{ {
m_bodyHandlersByChannel[channel].append(pHandler); m_bodyHandlersByChannel[channel].append(pHandler);
} }

91
src/qamqp/amqp_network.h
View File

@ -4,7 +4,7 @@
#include <QObject> #include <QObject>
#include <QTcpSocket> #include <QTcpSocket>
#ifndef QT_NO_SSL #ifndef QT_NO_SSL
#include <QSslSocket> # include <QSslSocket>
#endif #endif
#include <QPointer> #include <QPointer>
#include <QBuffer> #include <QBuffer>
@ -13,65 +13,64 @@
namespace QAMQP namespace QAMQP
{ {
class Network : public QObject
{
Q_OBJECT
Q_DISABLE_COPY(Network)
public:
typedef qint16 Channel;
Network(QObject * parent = 0); class Network : public QObject
~Network(); {
Q_OBJECT
public:
Network(QObject *parent = 0);
~Network();
void disconnect(); void disconnect();
void sendFrame(); void sendFrame();
void sendFrame(const QAMQP::Frame::Base & frame); void sendFrame(const QAMQP::Frame::Base &frame);
bool isSsl() const; bool isSsl() const;
void setSsl(bool value); void setSsl(bool value);
bool autoReconnect() const; bool autoReconnect() const;
void setAutoReconnect(bool value); void setAutoReconnect(bool value);
QAbstractSocket::SocketState state() const; QAbstractSocket::SocketState state() const;
void setMethodHandlerConnection(Frame::MethodHandler* pMethodHandlerConnection); typedef qint16 Channel;
void addMethodHandlerForChannel(Channel channel, Frame::MethodHandler* pHandler); void setMethodHandlerConnection(Frame::MethodHandler *pMethodHandlerConnection);
void addContentHandlerForChannel(Channel channel, Frame::ContentHandler* pHandler); void addMethodHandlerForChannel(Channel channel, Frame::MethodHandler *pHandler);
void addContentBodyHandlerForChannel(Channel channel, Frame::ContentBodyHandler* pHandler); void addContentHandlerForChannel(Channel channel, Frame::ContentHandler *pHandler);
void addContentBodyHandlerForChannel(Channel channel, Frame::ContentBodyHandler *pHandler);
public slots: public slots:
void connectTo(const QString & host = QString(), quint32 port = 0); void connectTo(const QString &host = QString(), quint16 port = 0);
void error( QAbstractSocket::SocketError socketError ); void error(QAbstractSocket::SocketError socketError);
signals: Q_SIGNALS:
void connected(); void connected();
void disconnected(); void disconnected();
private slots: private Q_SLOTS:
void readyRead(); void readyRead();
void sslErrors();
void conectionReady();
void sslErrors ( ); private:
Q_DISABLE_COPY(Network)
void initSocket(bool ssl = false);
QPointer<QTcpSocket> socket_;
QByteArray buffer_;
QString lastHost_;
int lastPort_;
bool autoReconnect_;
int timeOut_;
bool connect_;
void conectionReady(); Frame::MethodHandler *m_pMethodHandlerConnection;
private: QHash<Channel, QList<Frame::MethodHandler*> > m_methodHandlersByChannel;
void initSocket(bool ssl = false); QHash<Channel, QList<Frame::ContentHandler*> > m_contentHandlerByChannel;
QPointer<QTcpSocket> socket_; QHash<Channel, QList<Frame::ContentBodyHandler*> > m_bodyHandlersByChannel;
QByteArray buffer_; };
QString lastHost_;
int lastPort_;
bool autoReconnect_;
int timeOut_;
bool connect_;
Frame::MethodHandler* m_pMethodHandlerConnection;
QHash<Channel, QList<Frame::MethodHandler*> > m_methodHandlersByChannel;
QHash<Channel, QList<Frame::ContentHandler*> > m_contentHandlerByChannel;
QHash<Channel, QList<Frame::ContentBodyHandler*> > m_bodyHandlersByChannel;
};
} }
#endif // amqp_network_h__ #endif // amqp_network_h__

60
src/qamqp/amqp_p.h
View File

@ -1,5 +1,5 @@
#ifndef qamqp_amqp_p_h__ #ifndef QAMQP_P_H
#define qamqp_amqp_p_h__ #define QAMQP_P_H
#include <QSharedPointer> #include <QSharedPointer>
@ -10,42 +10,42 @@
namespace QAMQP namespace QAMQP
{ {
class ClientPrivate
{
P_DECLARE_PUBLIC(QAMQP::Client)
public: class Queue;
ClientPrivate(Client * q ) ; class Exchange;
~ClientPrivate(); class ClientPrivate
{
public:
ClientPrivate(Client *q);
~ClientPrivate();
void init(QObject *parent);
void init(QObject *parent, const QUrl &connectionString);
void init(QObject * parent); void printConnect() const;
void init(QObject * parent, const QUrl & connectionString); void connect();
void printConnect() const; void disconnect();
void connect(); void parseConnectionString( const QUrl &connectionString);
void disconnect(); void sockConnect();
void parseCnnString( const QUrl & connectionString); void login();
void sockConnect(); void setAuth(Authenticator* auth);
void login(); Exchange *createExchange(int channelNumber, const QString &name);
void setAuth(Authenticator* auth); Queue *createQueue(int channelNumber, const QString &name);
Exchange * createExchange(int channelNumber, const QString &name);
Queue * createQueue(int channelNumber, const QString &name);
quint32 port; quint32 port;
QString host; QString host;
QString virtualHost; QString virtualHost;
QPointer<QAMQP::Network> network_; QPointer<QAMQP::Network> network_;
QPointer<QAMQP::Connection> connection_; QPointer<QAMQP::Connection> connection_;
QSharedPointer<Authenticator> auth_; QSharedPointer<Authenticator> auth_;
bool isSSl() const; bool isSSl() const;
Client * const q_ptr;
Q_DECLARE_PUBLIC(QAMQP::Client)
};
Client * const pq_ptr;
};
} }
#endif // amqp_p_h__ #endif // amqp_p_h__

708
src/qamqp/amqp_queue.cpp
View File

@ -10,569 +10,525 @@ using namespace QAMQP::Frame;
#include <QDataStream> #include <QDataStream>
#include <QFile> #include <QFile>
namespace QAMQP Queue::Queue(int channelNumber, Client *parent)
: Channel(new QueuePrivate(this), parent)
{ {
struct QueueExceptionCleaner Q_D(QAMQP::Queue);
{ d->init(channelNumber, parent);
/* this cleans up when the constructor throws an exception */
static inline void cleanup(Queue *that, QueuePrivate *d)
{
#ifdef QT_NO_EXCEPTIONS
Q_UNUSED(that);
Q_UNUSED(d);
#else
Q_UNUSED(that);
Q_UNUSED(d);
#endif
}
};
}
Queue::Queue( int channelNumber, Client * parent /*= 0*/ )
: Channel(new QueuePrivate(this))
{
QT_TRY {
pd_func()->init(channelNumber, parent);
} QT_CATCH(...) {
QueueExceptionCleaner::cleanup(this, pd_func());
QT_RETHROW;
}
} }
Queue::~Queue() Queue::~Queue()
{ {
remove(); remove();
} }
void Queue::onOpen() void Queue::onOpen()
{ {
P_D(Queue); Q_D(QAMQP::Queue);
if(d->delayedDeclare) if (d->delayedDeclare)
{ d->declare();
d->declare();
} if (!d->delayedBindings.isEmpty()) {
if(!d->delayedBindings.isEmpty()) typedef QPair<QString, QString> BindingPair;
{ foreach(BindingPair binding, d->delayedBindings)
typedef QPair<QString, QString> BindingPair; d->bind(binding.first, binding.second);
foreach(BindingPair binding, d->delayedBindings) d->delayedBindings.clear();
{ }
d->bind(binding.first, binding.second);
}
d->delayedBindings.clear();
}
} }
void Queue::onClose() void Queue::onClose()
{ {
pd_func()->remove(true, true); Q_D(QAMQP::Queue);
d->remove(true, true);
} }
Queue::QueueOptions Queue::option() const Queue::QueueOptions Queue::option() const
{ {
return pd_func()->options; Q_D(const QAMQP::Queue);
return d->options;
} }
void Queue::setNoAck( bool noAck ) void Queue::setNoAck(bool noAck)
{ {
pd_func()->noAck = noAck; Q_D(QAMQP::Queue);
d->noAck = noAck;
} }
bool Queue::noAck() const bool Queue::noAck() const
{ {
return pd_func()->noAck; Q_D(const QAMQP::Queue);
return d->noAck;
} }
void Queue::declare() void Queue::declare()
{ {
P_D(Queue); Q_D(QAMQP::Queue);
declare(d->name, QueueOptions(Durable | AutoDelete)); declare(d->name, QueueOptions(Durable | AutoDelete));
} }
void Queue::declare( const QString &name, QueueOptions options ) void Queue::declare(const QString &name, QueueOptions options)
{ {
P_D(Queue); Q_D(QAMQP::Queue);
setName(name); setName(name);
d->options = options; d->options = options;
d->declare(); d->declare();
} }
void Queue::remove( bool ifUnused /*= true*/, bool ifEmpty /*= true*/, bool noWait /*= true*/ ) void Queue::remove(bool ifUnused, bool ifEmpty, bool noWait)
{ {
pd_func()->remove(ifUnused, ifEmpty, noWait); Q_D(QAMQP::Queue);
d->remove(ifUnused, ifEmpty, noWait);
} }
void Queue::purge() void Queue::purge()
{ {
pd_func()->purge(); Q_D(QAMQP::Queue);
d->purge();
} }
void Queue::bind( const QString & exchangeName, const QString & key ) void Queue::bind(const QString &exchangeName, const QString &key)
{ {
pd_func()->bind(exchangeName, key); Q_D(QAMQP::Queue);
d->bind(exchangeName, key);
} }
void Queue::bind( Exchange * exchange, const QString & key ) void Queue::bind(Exchange *exchange, const QString &key)
{ {
if(exchange) Q_D(QAMQP::Queue);
pd_func()->bind(exchange->name(), key); if (exchange)
d->bind(exchange->name(), key);
} }
void Queue::unbind( const QString & exchangeName, const QString & key ) void Queue::unbind(const QString &exchangeName, const QString &key)
{ {
pd_func()->unbind(exchangeName, key); Q_D(QAMQP::Queue);
d->unbind(exchangeName, key);
} }
void Queue::unbind( Exchange * exchange, const QString & key ) void Queue::unbind(Exchange *exchange, const QString &key)
{ {
if(exchange) Q_D(QAMQP::Queue);
pd_func()->unbind(exchange->name(), key); if (exchange)
d->unbind(exchange->name(), key);
} }
void Queue::_q_content(const Content &frame) void Queue::_q_content(const Content &frame)
{ {
pd_func()->_q_content(frame); Q_D(QAMQP::Queue);
d->_q_content(frame);
} }
void Queue::_q_body(const ContentBody &frame) void Queue::_q_body(const ContentBody &frame)
{ {
pd_func()->_q_body(frame); Q_D(QAMQP::Queue);
d->_q_body(frame);
} }
QAMQP::MessagePtr Queue::getMessage() QAMQP::MessagePtr Queue::getMessage()
{ {
return pd_func()->messages_.dequeue(); Q_D(QAMQP::Queue);
return d->messages_.dequeue();
} }
bool Queue::hasMessage() const bool Queue::hasMessage() const
{ {
Q_D(const QAMQP::Queue);
if (d->messages_.isEmpty())
return false;
if(pd_func()->messages_.isEmpty()) const MessagePtr &q = d->messages_.head();
{ return q->leftSize == 0;
return false;
}
const MessagePtr &q = pd_func()->messages_.head();
return q->leftSize == 0;
} }
void Queue::consume(ConsumeOptions options) void Queue::consume(ConsumeOptions options)
{ {
pd_func()->consume(options); Q_D(QAMQP::Queue);
d->consume(options);
} }
void Queue::setConsumerTag( const QString &consumerTag ) void Queue::setConsumerTag(const QString &consumerTag)
{ {
pd_func()->consumerTag = consumerTag; Q_D(QAMQP::Queue);
d->consumerTag = consumerTag;
} }
QString Queue::consumerTag() const QString Queue::consumerTag() const
{ {
return pd_func()->consumerTag; Q_D(const QAMQP::Queue);
return d->consumerTag;
} }
void Queue::get() void Queue::get()
{ {
pd_func()->get(); Q_D(QAMQP::Queue);
d->get();
} }
void Queue::ack(const MessagePtr &message)
void Queue::ack( const MessagePtr & message )
{ {
pd_func()->ack(message); Q_D(QAMQP::Queue);
d->ack(message);
} }
////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////
QueuePrivate::QueuePrivate(Queue * q) QueuePrivate::QueuePrivate(Queue * q)
:ChannelPrivate(q) : ChannelPrivate(q),
, delayedDeclare(false) delayedDeclare(false),
, declared(false) declared(false),
, noAck(true) noAck(true),
, recievingMessage(false) recievingMessage(false)
{ {
} }
QueuePrivate::~QueuePrivate() QueuePrivate::~QueuePrivate()
{ {
} }
bool QueuePrivate::_q_method(const QAMQP::Frame::Method &frame)
bool QueuePrivate::_q_method( const QAMQP::Frame::Method & frame )
{ {
if(ChannelPrivate::_q_method(frame)) Q_Q(QAMQP::Queue);
return true; if (ChannelPrivate::_q_method(frame))
return true;
if(frame.methodClass() == QAMQP::Frame::fcQueue) if (frame.methodClass() == QAMQP::Frame::fcQueue) {
{ switch (frame.id()) {
switch(frame.id()) case miDeclareOk:
{ declareOk(frame);
case miDeclareOk: break;
declareOk(frame); case miDelete:
break; deleteOk(frame);
case miDelete: break;
deleteOk(frame); case miBindOk:
break; bindOk(frame);
case miBindOk: break;
bindOk(frame); case miUnbindOk:
break; unbindOk(frame);
case miUnbindOk: break;
unbindOk(frame); case miPurgeOk:
break; deleteOk(frame);
case miPurgeOk: break;
deleteOk(frame); default:
break; break;
default: }
break;
}
return true;
}
if(frame.methodClass() == QAMQP::Frame::fcBasic) return true;
{ }
switch(frame.id())
{
case bmConsumeOk:
consumeOk(frame);
break;
case bmDeliver:
deliver(frame);
break;
case bmGetOk:
getOk(frame);
break;
case bmGetEmpty:
QMetaObject::invokeMethod(pq_func(), "empty");
break;
default:
break;
}
return true;
}
if (frame.methodClass() == QAMQP::Frame::fcBasic) {
switch(frame.id()) {
case bmConsumeOk:
consumeOk(frame);
break;
case bmDeliver:
deliver(frame);
break;
case bmGetOk:
getOk(frame);
break;
case bmGetEmpty:
QMetaObject::invokeMethod(q, "empty");
break;
default:
break;
}
return true;
}
return false;
return false;
} }
void QueuePrivate::declareOk( const QAMQP::Frame::Method & frame ) void QueuePrivate::declareOk(const QAMQP::Frame::Method &frame)
{ {
Q_Q(QAMQP::Queue);
qDebug() << "Declared queue: " << name;
declared = true;
qDebug() << "Declared queue: " << name; QByteArray data = frame.arguments();
declared = true; QDataStream stream(&data, QIODevice::ReadOnly);
QByteArray data = frame.arguments(); name = readField('s', stream).toString();
QDataStream stream(&data, QIODevice::ReadOnly); qint32 messageCount = 0, consumerCount = 0;
stream >> messageCount >> consumerCount;
qDebug("Message count %d\nConsumer count: %d", messageCount, consumerCount);
name = readField('s', stream).toString(); QMetaObject::invokeMethod(q, "declared");
qint32 messageCount = 0, consumerCount = 0;
stream >> messageCount >> consumerCount;
qDebug("Message count %d\nConsumer count: %d", messageCount, consumerCount);
QMetaObject::invokeMethod(pq_func(), "declared");
} }
void QueuePrivate::deleteOk( const QAMQP::Frame::Method & frame ) void QueuePrivate::deleteOk(const QAMQP::Frame::Method &frame)
{ {
qDebug() << "Deleted or purged queue: " << name; Q_Q(QAMQP::Queue);
declared = false; qDebug() << "Deleted or purged queue: " << name;
declared = false;
QByteArray data = frame.arguments(); QByteArray data = frame.arguments();
QDataStream stream(&data, QIODevice::ReadOnly); QDataStream stream(&data, QIODevice::ReadOnly);
qint32 messageCount = 0; qint32 messageCount = 0;
stream >> messageCount; stream >> messageCount;
qDebug("Message count %d", messageCount); qDebug("Message count %d", messageCount);
QMetaObject::invokeMethod(pq_func(), "removed"); QMetaObject::invokeMethod(q, "removed");
} }
void QueuePrivate::bindOk(const QAMQP::Frame::Method &frame)
void QueuePrivate::bindOk( const QAMQP::Frame::Method & )
{ {
qDebug() << "Binded to queue: " << name; Q_UNUSED(frame)
QMetaObject::invokeMethod(pq_func(), "binded", Q_ARG(bool, true)); Q_Q(QAMQP::Queue);
qDebug() << "Binded to queue: " << name;
QMetaObject::invokeMethod(q, "binded", Q_ARG(bool, true));
} }
void QueuePrivate::unbindOk( const QAMQP::Frame::Method & ) void QueuePrivate::unbindOk(const QAMQP::Frame::Method &frame)
{ {
qDebug() << "Unbinded queue: " << name; Q_UNUSED(frame)
QMetaObject::invokeMethod(pq_func(), "binded", Q_ARG(bool, false)); Q_Q(QAMQP::Queue);
qDebug() << "Unbinded queue: " << name;
QMetaObject::invokeMethod(q, "binded", Q_ARG(bool, false));
} }
void QueuePrivate::declare() void QueuePrivate::declare()
{ {
if(!opened) if (!opened) {
{ delayedDeclare = true;
delayedDeclare = true; return;
return; }
}
QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miDeclare);
frame.setChannel(number);
QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1
writeField('s', out, name);
out << qint8(options);
writeField('F', out, TableField());
frame.setArguments(arguments_);
sendFrame(frame);
delayedDeclare = false;
QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miDeclare);
frame.setChannel(number);
QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1
writeField('s', out, name);
out << qint8(options);
writeField('F', out, TableField());
frame.setArguments(arguments_);
sendFrame(frame);
delayedDeclare = false;
} }
void QueuePrivate::remove( bool ifUnused /*= true*/, bool ifEmpty /*= true*/, bool noWait /*= true*/ ) void QueuePrivate::remove(bool ifUnused, bool ifEmpty, bool noWait)
{ {
if(!declared) if (!declared)
return; return;
QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miDelete); QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miDelete);
frame.setChannel(number); frame.setChannel(number);
QByteArray arguments_; QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly); QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1 out << qint16(0); //reserver 1
writeField('s', out, name); writeField('s', out, name);
qint8 flag = 0; qint8 flag = 0;
flag |= (ifUnused ? 0x1 : 0); flag |= (ifUnused ? 0x1 : 0);
flag |= (ifEmpty ? 0x2 : 0); flag |= (ifEmpty ? 0x2 : 0);
flag |= (noWait ? 0x4 : 0); flag |= (noWait ? 0x4 : 0);
out << flag; out << flag;
frame.setArguments(arguments_);
sendFrame(frame);
frame.setArguments(arguments_);
sendFrame(frame);
} }
void QueuePrivate::purge() void QueuePrivate::purge()
{ {
if(!opened) if (!opened)
{ return;
return;
}
QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miPurge);
frame.setChannel(number);
QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1
writeField('s', out, name);
out << qint8(0); // no-wait
frame.setArguments(arguments_);
sendFrame(frame);
QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miPurge);
frame.setChannel(number);
QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1
writeField('s', out, name);
out << qint8(0); // no-wait
frame.setArguments(arguments_);
sendFrame(frame);
} }
void QueuePrivate::bind( const QString & exchangeName, const QString & key ) void QueuePrivate::bind(const QString & exchangeName, const QString &key)
{ {
if(!opened) if (!opened) {
{ delayedBindings.append(QPair<QString,QString>(exchangeName, key));
delayedBindings.append(QPair<QString,QString>(exchangeName, key)); return;
return; }
}
QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miBind); QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miBind);
frame.setChannel(number); frame.setChannel(number);
QByteArray arguments_; QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly); QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1 out << qint16(0); //reserver 1
writeField('s', out, name); writeField('s', out, name);
writeField('s', out, exchangeName); writeField('s', out, exchangeName);
writeField('s', out, key); writeField('s', out, key);
out << qint8(0); // no-wait out << qint8(0); // no-wait
writeField('F', out, TableField()); writeField('F', out, TableField());
frame.setArguments(arguments_);
sendFrame(frame);
frame.setArguments(arguments_);
sendFrame(frame);
} }
void QueuePrivate::unbind( const QString & exchangeName, const QString & key ) void QueuePrivate::unbind(const QString &exchangeName, const QString &key)
{ {
if(!opened) if (!opened)
{ return;
return;
}
QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miUnbind); QAMQP::Frame::Method frame(QAMQP::Frame::fcQueue, miUnbind);
frame.setChannel(number); frame.setChannel(number);
QByteArray arguments_; QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly); QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1 out << qint16(0); //reserver 1
writeField('s', out, name); writeField('s', out, name);
writeField('s', out, exchangeName); writeField('s', out, exchangeName);
writeField('s', out, key); writeField('s', out, key);
writeField('F', out, TableField()); writeField('F', out, TableField());
frame.setArguments(arguments_);
sendFrame(frame);
frame.setArguments(arguments_);
sendFrame(frame);
} }
void QueuePrivate::get() void QueuePrivate::get()
{ {
if(!opened) if (!opened)
{ return;
return;
}
QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, bmGet); QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, bmGet);
frame.setChannel(number); frame.setChannel(number);
QByteArray arguments_; QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly); QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1 out << qint16(0); //reserver 1
writeField('s', out, name); writeField('s', out, name);
out << qint8(noAck ? 1 : 0); // noAck out << qint8(noAck ? 1 : 0); // noAck
frame.setArguments(arguments_); frame.setArguments(arguments_);
sendFrame(frame); sendFrame(frame);
} }
void QueuePrivate::getOk(const QAMQP::Frame::Method &frame)
void QueuePrivate::getOk( const QAMQP::Frame::Method & frame )
{ {
QByteArray data = frame.arguments(); QByteArray data = frame.arguments();
QDataStream in(&data, QIODevice::ReadOnly); QDataStream in(&data, QIODevice::ReadOnly);
qlonglong deliveryTag = readField('L',in).toLongLong(); qlonglong deliveryTag = readField('L',in).toLongLong();
bool redelivered = readField('t',in).toBool(); bool redelivered = readField('t',in).toBool();
QString exchangeName = readField('s',in).toString(); QString exchangeName = readField('s',in).toString();
QString routingKey = readField('s',in).toString(); QString routingKey = readField('s',in).toString();
Q_UNUSED(redelivered) Q_UNUSED(redelivered)
MessagePtr newMessage = MessagePtr(new Message); MessagePtr newMessage = MessagePtr(new Message);
newMessage->routeKey = routingKey; newMessage->routeKey = routingKey;
newMessage->exchangeName = exchangeName; newMessage->exchangeName = exchangeName;
newMessage->deliveryTag = deliveryTag; newMessage->deliveryTag = deliveryTag;
messages_.enqueue(newMessage); messages_.enqueue(newMessage);
} }
void QueuePrivate::ack(const MessagePtr &Message)
void QueuePrivate::ack( const MessagePtr & Message )
{ {
if(!opened) if (!opened)
{ return;
return;
}
QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, bmAck); QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, bmAck);
frame.setChannel(number); frame.setChannel(number);
QByteArray arguments_; QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly); QDataStream out(&arguments_, QIODevice::WriteOnly);
out << Message->deliveryTag; //reserver 1 out << Message->deliveryTag; //reserver 1
out << qint8(0); // noAck out << qint8(0); // noAck
frame.setArguments(arguments_); frame.setArguments(arguments_);
sendFrame(frame); sendFrame(frame);
} }
void QueuePrivate::consume( Queue::ConsumeOptions options ) void QueuePrivate::consume(Queue::ConsumeOptions options)
{ {
if(!opened) if (!opened)
{ return;
return;
}
QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, bmConsume); QAMQP::Frame::Method frame(QAMQP::Frame::fcBasic, bmConsume);
frame.setChannel(number); frame.setChannel(number);
QByteArray arguments_; QByteArray arguments_;
QDataStream out(&arguments_, QIODevice::WriteOnly); QDataStream out(&arguments_, QIODevice::WriteOnly);
out << qint16(0); //reserver 1 out << qint16(0); //reserver 1
writeField('s', out, name); writeField('s', out, name);
writeField('s', out, consumerTag); writeField('s', out, consumerTag);
out << qint8(options); // no-wait out << qint8(options); // no-wait
writeField('F', out, TableField()); writeField('F', out, TableField());
frame.setArguments(arguments_);
sendFrame(frame);
frame.setArguments(arguments_);
sendFrame(frame);
} }
void QueuePrivate::consumeOk(const QAMQP::Frame::Method &frame)
void QueuePrivate::consumeOk( const QAMQP::Frame::Method & frame )
{ {
qDebug() << "Consume ok: " << name;
declared = false;
qDebug() << "Consume ok: " << name; QByteArray data = frame.arguments();
declared = false; QDataStream stream(&data, QIODevice::ReadOnly);
consumerTag = readField('s',stream).toString();
QByteArray data = frame.arguments(); qDebug("Consumer tag = %s", qPrintable(consumerTag));
QDataStream stream(&data, QIODevice::ReadOnly);
consumerTag = readField('s',stream).toString();
qDebug("Consumer tag = %s", qPrintable(consumerTag));
} }
void QueuePrivate::deliver(const QAMQP::Frame::Method &frame)
void QueuePrivate::deliver( const QAMQP::Frame::Method & frame )
{ {
QByteArray data = frame.arguments();
QDataStream in(&data, QIODevice::ReadOnly);
QString consumer_ = readField('s',in).toString();
if (consumer_ != consumerTag)
return;
QByteArray data = frame.arguments(); qlonglong deliveryTag = readField('L',in).toLongLong();
QDataStream in(&data, QIODevice::ReadOnly); bool redelivered = readField('t',in).toBool();
QString consumer_ = readField('s',in).toString(); QString exchangeName = readField('s',in).toString();
if(consumer_ != consumerTag) QString routingKey = readField('s',in).toString();
{
return;
}
qlonglong deliveryTag = readField('L',in).toLongLong(); Q_UNUSED(redelivered)
bool redelivered = readField('t',in).toBool();
QString exchangeName = readField('s',in).toString();
QString routingKey = readField('s',in).toString();
Q_UNUSED(redelivered)
MessagePtr newMessage = MessagePtr(new Message);
newMessage->routeKey = routingKey;
newMessage->exchangeName = exchangeName;
newMessage->deliveryTag = deliveryTag;
messages_.enqueue(newMessage);
MessagePtr newMessage = MessagePtr(new Message);
newMessage->routeKey = routingKey;
newMessage->exchangeName = exchangeName;
newMessage->deliveryTag = deliveryTag;
messages_.enqueue(newMessage);
} }
void QueuePrivate::_q_content( const QAMQP::Frame::Content & frame ) void QueuePrivate::_q_content(const QAMQP::Frame::Content &frame)
{ {
Q_ASSERT(frame.channel() == number); Q_ASSERT(frame.channel() == number);
if(frame.channel() != number) if (frame.channel() != number)
return; return;
if(messages_.isEmpty())
{ if (messages_.isEmpty()) {
qErrnoWarning("Received content-header without method frame before"); qErrnoWarning("Received content-header without method frame before");
return; return;
} }
MessagePtr &message = messages_.last();
message->leftSize = frame.bodySize(); MessagePtr &message = messages_.last();
QHash<int, QVariant>::ConstIterator i; message->leftSize = frame.bodySize();
for (i = frame.properties_.begin(); i != frame.properties_.end(); ++i) QHash<int, QVariant>::ConstIterator i;
{ for (i = frame.properties_.begin(); i != frame.properties_.end(); ++i)
message->property[Message::MessageProperty(i.key())]= i.value(); message->property[Message::MessageProperty(i.key())]= i.value();
}
} }
void QueuePrivate::_q_body(const QAMQP::Frame::ContentBody & frame) void QueuePrivate::_q_body(const QAMQP::Frame::ContentBody &frame)
{ {
Q_ASSERT(frame.channel() == number); Q_Q(QAMQP::Queue);
if(frame.channel() != number) Q_ASSERT(frame.channel() == number);
return; if (frame.channel() != number)
return;
if(messages_.isEmpty()) if (messages_.isEmpty()) {
{ qErrnoWarning("Received content-body without method frame before");
qErrnoWarning("Received content-body without method frame before"); return;
return; }
}
MessagePtr &message = messages_.last();
message->payload.append(frame.body());
message->leftSize -= frame.body().size();
if(message->leftSize == 0 && messages_.size() == 1) MessagePtr &message = messages_.last();
{ message->payload.append(frame.body());
emit pq_func()->messageReceived(pq_func()); message->leftSize -= frame.body().size();
}
if (message->leftSize == 0 && messages_.size() == 1)
Q_EMIT q->messageReceived(q);
} }

130
src/qamqp/amqp_queue.h
View File

@ -6,84 +6,88 @@
namespace QAMQP namespace QAMQP
{ {
class Client;
class ClientPrivate;
class Exchange;
class QueuePrivate;
class Queue : public Channel, public Frame::ContentHandler, public Frame::ContentBodyHandler
{
Q_OBJECT
Queue(int channelNumber = -1, Client * parent = 0);
Q_PROPERTY(QueueOptions option READ option ); class Client;
Q_PROPERTY(QString consumerTag READ consumerTag WRITE setConsumerTag) class ClientPrivate;
Q_PROPERTY(bool noAck READ noAck WRITE setNoAck) class Exchange;
class QueuePrivate;
class Queue : public Channel, public Frame::ContentHandler, public Frame::ContentBodyHandler
{
Q_OBJECT
Q_ENUMS(QueueOptions)
Q_PROPERTY(QueueOptions option READ option)
Q_PROPERTY(QString consumerTag READ consumerTag WRITE setConsumerTag)
Q_PROPERTY(bool noAck READ noAck WRITE setNoAck)
P_DECLARE_PRIVATE(QAMQP::Queue) public:
Q_DISABLE_COPY(Queue); enum QueueOption {
friend class ClientPrivate; NoOptions = 0x0,
Passive = 0x01,
Durable = 0x02,
Exclusive = 0x4,
AutoDelete = 0x8,
NoWait = 0x10
};
Q_DECLARE_FLAGS(QueueOptions, QueueOption)
protected: enum ConsumeOption {
void onOpen(); coNoLocal = 0x1,
void onClose(); coNoAck = 0x02,
coExclusive = 0x04,
coNoWait = 0x8
};
Q_DECLARE_FLAGS(ConsumeOptions, ConsumeOption)
public: ~Queue();
enum QueueOption { QueueOptions option() const;
NoOptions = 0x0,
Passive = 0x01,
Durable = 0x02,
Exclusive = 0x4,
AutoDelete = 0x8,
NoWait = 0x10
};
Q_DECLARE_FLAGS(QueueOptions, QueueOption)
enum ConsumeOption { void declare();
coNoLocal = 0x1, void declare(const QString &name, QueueOptions options);
coNoAck = 0x02, void remove(bool ifUnused = true, bool ifEmpty = true, bool noWait = true);
coExclusive = 0x04,
coNoWait = 0x8
};
Q_DECLARE_FLAGS(ConsumeOptions, ConsumeOption)
~Queue(); void purge();
QueueOptions option() const; void bind(const QString & exchangeName, const QString & key);
void bind(Exchange * exchange, const QString & key);
void declare(); void unbind(const QString & exchangeName, const QString & key);
void declare(const QString &name, QueueOptions options); void unbind(Exchange * exchange, const QString & key);
void remove(bool ifUnused = true, bool ifEmpty = true, bool noWait = true);
void purge(); MessagePtr getMessage();
void get();
void ack(const MessagePtr & message);
bool hasMessage() const;
void consume(ConsumeOptions options = ConsumeOptions(NoOptions));
void setConsumerTag(const QString &consumerTag);
QString consumerTag() const;
void bind(const QString & exchangeName, const QString & key); void setNoAck(bool noAck);
void bind(Exchange * exchange, const QString & key); bool noAck() const;
void unbind(const QString & exchangeName, const QString & key); Q_SIGNALS:
void unbind(Exchange * exchange, const QString & key); void declared();
void binded(bool);
void removed();
void messageReceived(QAMQP::Queue *pQueue);
void empty();
MessagePtr getMessage(); protected:
void get(); void onOpen();
void ack(const MessagePtr & message); void onClose();
bool hasMessage() const;
void consume(ConsumeOptions options = ConsumeOptions(NoOptions));
void setConsumerTag(const QString &consumerTag);
QString consumerTag() const;
void setNoAck(bool noAck); private:
bool noAck() const; Queue(int channelNumber = -1, Client * parent = 0);
Q_SIGNALS: void _q_content(const QAMQP::Frame::Content & frame);
void declared(); void _q_body(const QAMQP::Frame::ContentBody & frame);
void binded(bool);
void removed(); Q_DISABLE_COPY(Queue)
void messageReceived(QAMQP::Queue* pQueue); Q_DECLARE_PRIVATE(QAMQP::Queue)
void empty();
friend class ClientPrivate;
};
private:
void _q_content(const QAMQP::Frame::Content & frame);
void _q_body(const QAMQP::Frame::ContentBody & frame);
};
} }
#ifdef QAMQP_P_INCLUDE #ifdef QAMQP_P_INCLUDE
# include "amqp_queue_p.h" # include "amqp_queue_p.h"

95
src/qamqp/amqp_queue_p.h
View File

@ -7,64 +7,67 @@
namespace QAMQP namespace QAMQP
{ {
using namespace QAMQP::Frame;
class QueuePrivate: public ChannelPrivate
{
P_DECLARE_PUBLIC(QAMQP::Queue)
public:
enum MethodId using namespace QAMQP::Frame;
{
METHOD_ID_ENUM(miDeclare, 10),
METHOD_ID_ENUM(miBind, 20),
METHOD_ID_ENUM(miUnbind, 50),
METHOD_ID_ENUM(miPurge, 30),
METHOD_ID_ENUM(miDelete, 40)
};
QueuePrivate(Queue * q); class QueuePrivate: public ChannelPrivate
~QueuePrivate(); {
public:
enum MethodId {
METHOD_ID_ENUM(miDeclare, 10),
METHOD_ID_ENUM(miBind, 20),
METHOD_ID_ENUM(miUnbind, 50),
METHOD_ID_ENUM(miPurge, 30),
METHOD_ID_ENUM(miDelete, 40)
};
void declare(); QueuePrivate(Queue *q);
void remove(bool ifUnused = true, bool ifEmpty = true, bool noWait = true); ~QueuePrivate();
void purge();
void bind(const QString & exchangeName, const QString & key);
void unbind(const QString & exchangeName, const QString & key);
void declareOk(const QAMQP::Frame::Method & frame); void declare();
void deleteOk(const QAMQP::Frame::Method & frame); void remove(bool ifUnused = true, bool ifEmpty = true, bool noWait = true);
void bindOk(const QAMQP::Frame::Method & frame); void purge();
void unbindOk(const QAMQP::Frame::Method & frame); void bind(const QString &exchangeName, const QString &key);
void unbind(const QString &exchangeName, const QString &key);
/************************************************************************/ void declareOk(const QAMQP::Frame::Method &frame);
/* CLASS BASIC METHODS */ void deleteOk(const QAMQP::Frame::Method &frame);
/************************************************************************/ void bindOk(const QAMQP::Frame::Method &frame);
void unbindOk(const QAMQP::Frame::Method &frame);
void consume(Queue::ConsumeOptions options); /************************************************************************/
void consumeOk(const QAMQP::Frame::Method & frame); /* CLASS BASIC METHODS */
void deliver(const QAMQP::Frame::Method & frame); /************************************************************************/
void get(); void consume(Queue::ConsumeOptions options);
void getOk(const QAMQP::Frame::Method & frame); void consumeOk(const QAMQP::Frame::Method &frame);
void ack(const MessagePtr & Message); void deliver(const QAMQP::Frame::Method &frame);
QString type; void get();
Queue::QueueOptions options; void getOk(const QAMQP::Frame::Method &frame);
void ack(const MessagePtr &Message);
bool _q_method(const QAMQP::Frame::Method & frame); QString type;
Queue::QueueOptions options;
bool delayedDeclare; bool _q_method(const QAMQP::Frame::Method &frame);
bool declared;
bool noAck;
QString consumerTag;
QQueue<QPair<QString, QString> > delayedBindings; bool delayedDeclare;
QQueue<QAMQP::MessagePtr> messages_; bool declared;
bool noAck;
QString consumerTag;
bool recievingMessage; QQueue<QPair<QString, QString> > delayedBindings;
QQueue<QAMQP::MessagePtr> messages_;
bool recievingMessage;
void _q_content(const QAMQP::Frame::Content &frame);
void _q_body(const QAMQP::Frame::ContentBody &frame);
Q_DECLARE_PUBLIC(QAMQP::Queue)
};
void _q_content(const QAMQP::Frame::Content & frame);
void _q_body(const QAMQP::Frame::ContentBody & frame);
};
} }
#endif // amqp_queue_p_h__ #endif // amqp_queue_p_h__

1
src/qamqp/qamqp.pri
View File

@ -24,5 +24,6 @@ SOURCES += $$PWD/amqp.cpp \
$$PWD/amqp_connection.cpp \ $$PWD/amqp_connection.cpp \
$$PWD/amqp_exchange.cpp \ $$PWD/amqp_exchange.cpp \
$$PWD/amqp_frame.cpp \ $$PWD/amqp_frame.cpp \
$$PWD/amqp_message.cpp \
$$PWD/amqp_network.cpp \ $$PWD/amqp_network.cpp \
$$PWD/amqp_queue.cpp \ $$PWD/amqp_queue.cpp \