jessequ
/
AMQP-CPP
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Work in progress to convert channel handler to callback system

This commit is contained in:
Martijn Otto 2014-04-08 14:42:07 +02:00
parent 400c619b01
commit 2939272bc8
22 changed files with 1137 additions and 628 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
amqpcpp.h
View File

@ -49,6 +49,8 @@
// mid level includes // mid level includes
#include <amqpcpp/exchangetype.h> #include <amqpcpp/exchangetype.h>
#include <amqpcpp/flags.h> #include <amqpcpp/flags.h>
#include <amqpcpp/deferred.h>
#include <amqpcpp/callbacks.h>
#include <amqpcpp/channelhandler.h> #include <amqpcpp/channelhandler.h>
#include <amqpcpp/channelimpl.h> #include <amqpcpp/channelimpl.h>
#include <amqpcpp/channel.h> #include <amqpcpp/channel.h>

157
include/callbacks.h Normal file
View File

@ -0,0 +1,157 @@
/**
* Callbacks.h
*
* Class storing deferred callbacks of different type.
*
* @copyright 2014 Copernica BV
*/
/**
* Set up namespace
*/
namespace AMQP {
/**
* Class for managing deferred callbacks
*/
class Callbacks
{
private:
/**
* Different callback types supported
*/
std::tuple<
std::deque<Deferred<>>,
std::deque<Deferred<const std::string&, uint32_t, uint32_t>>,
std::deque<Deferred<uint32_t>>
> _callbacks;
/**
* If all else fails, we have gotten the wrong
* type, which is not present in the arguments.
*
* This should result in a compile error.
*/
template <class T, std::size_t N, class... Arguments>
struct getIndex
{
// if this structure is used, we went past the last argument
// and this static_assert should trigger a compile failure.
static_assert(N < sizeof...(Arguments), "Type T not found in Arguments");
// we still have to provide this member though
static constexpr std::size_t value = N;
};
/**
* This structure has one static member that represents
* the index of T in Arguments. This variant is used where U
* does equal T, so a match is found, meaning the current
* index given is the right one.
*/
template <class T, std::size_t N, class... Arguments>
struct getIndex<T, N, T, Arguments...>
{
// element is same type as we are looking for
static constexpr std::size_t value = N;
};
/**
* This structure has one static member that represents
* the index of T in Arguments. This variant is used where U
* does not equal T, so we need to look at the next member.
*/
template <class T, std::size_t N, class U, class... Arguments>
struct getIndex<T, N, U, Arguments...>
{
// current N is not correct, unroll to next element
static constexpr std::size_t value = getIndex<T, N + 1, Arguments...>::value;
};
/**
* Retrieve the list of callbacks matching the type
*
* @param tuple tuple with callbacks
*/
template <class T, class... Arguments>
T& get(std::tuple<Arguments...>& tuple)
{
// retrieve the index at which the requested callbacks can be found
constexpr std::size_t index = getIndex<T, 0, Arguments...>::value;
// retrieve the callbacks
return std::get<index>(tuple);
}
public:
/**
* Add a deferred to the available callbacks
*
* @param deferred the deferred to add
* @return reference to the inserted deferred
*/
template <typename... Arguments>
Deferred<Arguments...>& push_back(const Deferred<Arguments...>& item)
{
// retrieve the container
auto &container = get<std::deque<Deferred<Arguments...>>>(_callbacks);
// add the element
container.push_back(item);
// return reference to the new item
return container.back();
}
/**
* Report success to the relevant callback
*
* @param mixed... additional parameters
*/
template <typename... Arguments>
void reportSuccess(Arguments ...parameters)
{
// retrieve the container and element
auto &container = get<std::deque<Deferred<Arguments...>>>(_callbacks);
auto &callback = container.front();
// execute the callback
callback.success(parameters...);
// remove the executed callback
container.pop_front();
}
/**
* Report a failure
*
* @param error a description of the error
*/
template <std::size_t N = 0>
typename std::enable_if<N == std::tuple_size<decltype(_callbacks)>::value>::type
reportFailure(const std::string& message)
{}
/**
* Report a failure
*
* @param error a description of the error
*/
template <std::size_t N = 0>
typename std::enable_if<N < std::tuple_size<decltype(_callbacks)>::value>::type
reportFailure(const std::string& message)
{
// retrieve the callbacks at current index
auto &callbacks = std::get<N>(_callbacks);
// report errors to all callbacks of the current type
for (auto &callback : callbacks) callback.error(message);
// execute the next type
reportFailure<N + 1>(message);
}
};
/**
* End namespace
*/
}

198
include/channel.h
View File

@ -35,18 +35,43 @@ public:
*/ */
virtual ~Channel() {} virtual ~Channel() {}
/**
* Callback that is called when the channel was succesfully created.
*
* Only one callback can be registered. Calling this function multiple
* times will remove the old callback.
*
* @param callback the callback to execute
*/
void onReady(const std::function<void(Channel *channel)>& callback)
{
// store callback in implementation
_implementation._readyCallback = callback;
}
/**
* Callback that is called when an error occurs.
*
* Only one error callback can be registered. Calling this function
* multiple times will remove the old callback.
*
* @param callback the callback to execute
*/
void onError(const std::function<void(Channel *channel, const std::string& message)>& callback)
{
// store callback in implementation
_implementation._errorCallback = callback;
}
/** /**
* Pause deliveries on a channel * Pause deliveries on a channel
* *
* This will stop all incoming messages * This will stop all incoming messages
* *
* This method returns true if the request to pause has been sent to the * This function returns a deferred handler. Callbacks can be installed
* broker. This does not necessarily mean that the channel is already * using onSuccess(), onError() and onFinalize() methods.
* paused.
*
* @return bool
*/ */
bool pause() Deferred<>& pause()
{ {
return _implementation.pause(); return _implementation.pause();
} }
@ -54,9 +79,12 @@ public:
/** /**
* Resume a paused channel * Resume a paused channel
* *
* @return bool * This will resume incoming messages
*
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool resume() Deferred<>& resume()
{ {
return _implementation.resume(); return _implementation.resume();
} }
@ -72,27 +100,33 @@ public:
/** /**
* Start a transaction * Start a transaction
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool startTransaction() Deferred<>& startTransaction()
{ {
return _implementation.startTransaction(); return _implementation.startTransaction();
} }
/** /**
* Commit the current transaction * Commit the current transaction
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool commitTransaction() Deferred<>& commitTransaction()
{ {
return _implementation.commitTransaction(); return _implementation.commitTransaction();
} }
/** /**
* Rollback the current transaction * Rollback the current transaction
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool rollbackTransaction() Deferred<>& rollbackTransaction()
{ {
return _implementation.rollbackTransaction(); return _implementation.rollbackTransaction();
} }
@ -112,14 +146,16 @@ public:
* @param type exchange type * @param type exchange type
* @param flags exchange flags * @param flags exchange flags
* @param arguments additional arguments * @param arguments additional arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool declareExchange(const std::string &name, ExchangeType type, int flags, const Table &arguments) { return _implementation.declareExchange(name, type, flags, arguments); } Deferred<>& declareExchange(const std::string &name, ExchangeType type, int flags, const Table &arguments) { return _implementation.declareExchange(name, type, flags, arguments); }
bool declareExchange(const std::string &name, ExchangeType type, const Table &arguments) { return _implementation.declareExchange(name, type, 0, arguments); } Deferred<>& declareExchange(const std::string &name, ExchangeType type, const Table &arguments) { return _implementation.declareExchange(name, type, 0, arguments); }
bool declareExchange(const std::string &name, ExchangeType type = fanout, int flags = 0) { return _implementation.declareExchange(name, type, flags, Table()); } Deferred<>& declareExchange(const std::string &name, ExchangeType type = fanout, int flags = 0) { return _implementation.declareExchange(name, type, flags, Table()); }
bool declareExchange(ExchangeType type, int flags, const Table &arguments) { return _implementation.declareExchange(std::string(), type, flags, arguments); } Deferred<>& declareExchange(ExchangeType type, int flags, const Table &arguments) { return _implementation.declareExchange(std::string(), type, flags, arguments); }
bool declareExchange(ExchangeType type, const Table &arguments) { return _implementation.declareExchange(std::string(), type, 0, arguments); } Deferred<>& declareExchange(ExchangeType type, const Table &arguments) { return _implementation.declareExchange(std::string(), type, 0, arguments); }
bool declareExchange(ExchangeType type = fanout, int flags = 0) { return _implementation.declareExchange(std::string(), type, flags, Table()); } Deferred<>& declareExchange(ExchangeType type = fanout, int flags = 0) { return _implementation.declareExchange(std::string(), type, flags, Table()); }
/** /**
* Remove an exchange * Remove an exchange
@ -130,9 +166,11 @@ public:
* @param name name of the exchange to remove * @param name name of the exchange to remove
* @param flags optional flags * @param flags optional flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool removeExchange(const std::string &name, int flags = 0) { return _implementation.removeExchange(name, flags); } Deferred<>& removeExchange(const std::string &name, int flags = 0) { return _implementation.removeExchange(name, flags); }
/** /**
* Bind two exchanges to each other * Bind two exchanges to each other
@ -146,11 +184,13 @@ public:
* @param routingkey the routing key * @param routingkey the routing key
* @param flags optional flags * @param flags optional flags
* @param arguments additional bind arguments * @param arguments additional bind arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments) { return _implementation.bindExchange(source, target, routingkey, flags, arguments); } Deferred<>& bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments) { return _implementation.bindExchange(source, target, routingkey, flags, arguments); }
bool bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, const Table &arguments) { return _implementation.bindExchange(source, target, routingkey, 0, arguments); } Deferred<>& bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, const Table &arguments) { return _implementation.bindExchange(source, target, routingkey, 0, arguments); }
bool bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags = 0) { return _implementation.bindExchange(source, target, routingkey, flags, Table()); } Deferred<>& bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags = 0) { return _implementation.bindExchange(source, target, routingkey, flags, Table()); }
/** /**
* Unbind two exchanges from one another * Unbind two exchanges from one another
@ -164,11 +204,13 @@ public:
* @param routingkey the routing key * @param routingkey the routing key
* @param flags optional flags * @param flags optional flags
* @param arguments additional unbind arguments * @param arguments additional unbind arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool unbindExchange(const std::string &target, const std::string &source, const std::string &routingkey, int flags, const Table &arguments) { return _implementation.unbindExchange(target, source, routingkey, flags, arguments); } Deferred<>& unbindExchange(const std::string &target, const std::string &source, const std::string &routingkey, int flags, const Table &arguments) { return _implementation.unbindExchange(target, source, routingkey, flags, arguments); }
bool unbindExchange(const std::string &target, const std::string &source, const std::string &routingkey, const Table &arguments) { return _implementation.unbindExchange(target, source, routingkey, 0, arguments); } Deferred<>& unbindExchange(const std::string &target, const std::string &source, const std::string &routingkey, const Table &arguments) { return _implementation.unbindExchange(target, source, routingkey, 0, arguments); }
bool unbindExchange(const std::string &target, const std::string &source, const std::string &routingkey, int flags = 0) { return _implementation.unbindExchange(target, source, routingkey, flags, Table()); } Deferred<>& unbindExchange(const std::string &target, const std::string &source, const std::string &routingkey, int flags = 0) { return _implementation.unbindExchange(target, source, routingkey, flags, Table()); }
/** /**
* Declare a queue * Declare a queue
@ -185,13 +227,26 @@ public:
* @param name name of the queue * @param name name of the queue
* @param flags combination of flags * @param flags combination of flags
* @param arguments optional arguments * @param arguments optional arguments
*
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*
* The onSuccess() callback that you can install should have the following signature:
*
* void myCallback(AMQP::Channel *channel, const std::string &name, uint32_t messageCount, uint32_t consumerCount);
*
* For example: channel.declareQueue("myqueue").onSuccess([](AMQP::Channel *channel, const std::string &name, uint32_t messageCount, uint32_t consumerCount) {
*
* std::cout << "Queue '" << name << "' has been declared with " << messageCount << " messages and " << consumerCount << " consumers" << std::endl;
*
* });
*/ */
bool declareQueue(const std::string &name, int flags, const Table &arguments) { return _implementation.declareQueue(name, flags, arguments); } Deferred<const std::string&, uint32_t, uint32_t>& declareQueue(const std::string &name, int flags, const Table &arguments) { return _implementation.declareQueue(name, flags, arguments); }
bool declareQueue(const std::string &name, const Table &arguments) { return _implementation.declareQueue(name, 0, arguments); } Deferred<const std::string&, uint32_t, uint32_t>& declareQueue(const std::string &name, const Table &arguments) { return _implementation.declareQueue(name, 0, arguments); }
bool declareQueue(const std::string &name, int flags = 0) { return _implementation.declareQueue(name, flags, Table()); } Deferred<const std::string&, uint32_t, uint32_t>& declareQueue(const std::string &name, int flags = 0) { return _implementation.declareQueue(name, flags, Table()); }
bool declareQueue(int flags, const Table &arguments) { return _implementation.declareQueue(std::string(), flags, arguments); } Deferred<const std::string&, uint32_t, uint32_t>& declareQueue(int flags, const Table &arguments) { return _implementation.declareQueue(std::string(), flags, arguments); }
bool declareQueue(const Table &arguments) { return _implementation.declareQueue(std::string(), 0, arguments); } Deferred<const std::string&, uint32_t, uint32_t>& declareQueue(const Table &arguments) { return _implementation.declareQueue(std::string(), 0, arguments); }
bool declareQueue(int flags = 0) { return _implementation.declareQueue(std::string(), flags, Table()); } Deferred<const std::string&, uint32_t, uint32_t>& declareQueue(int flags = 0) { return _implementation.declareQueue(std::string(), flags, Table()); }
/** /**
* Bind a queue to an exchange * Bind a queue to an exchange
@ -205,11 +260,13 @@ public:
* @param routingkey the routing key * @param routingkey the routing key
* @param flags additional flags * @param flags additional flags
* @param arguments additional bind arguments * @param arguments additional bind arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool bindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, int flags, const Table &arguments) { return _implementation.bindQueue(exchange, queue, routingkey, flags, arguments); } Deferred<>& bindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, int flags, const Table &arguments) { return _implementation.bindQueue(exchange, queue, routingkey, flags, arguments); }
bool bindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, const Table &arguments) { return _implementation.bindQueue(exchange, queue, routingkey, 0, arguments); } Deferred<>& bindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, const Table &arguments) { return _implementation.bindQueue(exchange, queue, routingkey, 0, arguments); }
bool bindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, int flags = 0) { return _implementation.bindQueue(exchange, queue, routingkey, flags, Table()); } Deferred<>& bindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, int flags = 0) { return _implementation.bindQueue(exchange, queue, routingkey, flags, Table()); }
/** /**
* Unbind a queue from an exchange * Unbind a queue from an exchange
@ -217,10 +274,12 @@ public:
* @param queue the target queue * @param queue the target queue
* @param routingkey the routing key * @param routingkey the routing key
* @param arguments additional bind arguments * @param arguments additional bind arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool unbindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, const Table &arguments) { return _implementation.unbindQueue(exchange, queue, routingkey, arguments); } Deferred<>& unbindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, const Table &arguments) { return _implementation.unbindQueue(exchange, queue, routingkey, arguments); }
bool unbindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey) { return _implementation.unbindQueue(exchange, queue, routingkey, Table()); } Deferred<>& unbindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey) { return _implementation.unbindQueue(exchange, queue, routingkey, Table()); }
/** /**
* Purge a queue * Purge a queue
@ -231,9 +290,21 @@ public:
* *
* @param name name of the queue * @param name name of the queue
* @param flags additional flags * @param flags additional flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*
* The onSuccess() callback that you can install should have the following signature:
*
* void myCallback(AMQP::Channel *channel, uint32_t messageCount);
*
* For example: channel.declareQueue("myqueue").onSuccess([](AMQP::Channel *channel, uint32_t messageCount) {
*
* std::cout << "Queue purged, all " << messageCount << " messages removed" << std::endl;
*
* });
*/ */
bool purgeQueue(const std::string &name, int flags = 0){ return _implementation.purgeQueue(name, flags); } Deferred<uint32_t>& purgeQueue(const std::string &name, int flags = 0){ return _implementation.purgeQueue(name, flags); }
/** /**
* Remove a queue * Remove a queue
@ -245,9 +316,21 @@ public:
* *
* @param name name of the queue to remove * @param name name of the queue to remove
* @param flags optional flags * @param flags optional flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*
* The onSuccess() callback that you can install should have the following signature:
*
* void myCallback(AMQP::Channel *channel, uint32_t messageCount);
*
* For example: channel.declareQueue("myqueue").onSuccess([](AMQP::Channel *channel, uint32_t messageCount) {
*
* std::cout << "Queue deleted, along with " << messageCount << " messages" << std::endl;
*
* });
*/ */
bool removeQueue(const std::string &name, int flags = 0) { return _implementation.removeQueue(name, flags); } Deferred<uint32_t>& removeQueue(const std::string &name, int flags = 0) { return _implementation.removeQueue(name, flags); }
/** /**
* Publish a message to an exchange * Publish a message to an exchange
@ -279,7 +362,7 @@ public:
/** /**
* Set the Quality of Service (QOS) for this channel * Set the Quality of Service (QOS) for this channel
* *
* When you consume messages, every single messages needs to be ack'ed to inform * When you consume messages, every single message needs to be ack'ed to inform
* the RabbitMQ server that is has been received. The Qos setting specifies the * the RabbitMQ server that is has been received. The Qos setting specifies the
* number of unacked messages that may exist in the client application. The server * number of unacked messages that may exist in the client application. The server
* stops delivering more messages if the number of unack'ed messages has reached * stops delivering more messages if the number of unack'ed messages has reached
@ -288,7 +371,7 @@ public:
* @param prefetchCount maximum number of messages to prefetch * @param prefetchCount maximum number of messages to prefetch
* @return bool whether the Qos frame is sent. * @return bool whether the Qos frame is sent.
*/ */
bool setQos(uint16_t prefetchCount) Deferred<>& setQos(uint16_t prefetchCount)
{ {
return _implementation.setQos(prefetchCount); return _implementation.setQos(prefetchCount);
} }
@ -392,18 +475,19 @@ public:
* - requeue if set, the server will requeue the messages, so the could also end up with at different consumer * - requeue if set, the server will requeue the messages, so the could also end up with at different consumer
* *
* @param flags * @param flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool recover(int flags = 0) { return _implementation.recover(flags); } Deferred<>& recover(int flags = 0) { return _implementation.recover(flags); }
/** /**
* Close the current channel * Close the current channel
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool close() Deferred<>& close() { return _implementation.close(); }
{
return _implementation.close();
}
/** /**
* Get the channel we're working on * Get the channel we're working on

122
include/channelhandler.h
View File

@ -22,12 +22,6 @@ namespace AMQP {
class ChannelHandler class ChannelHandler
{ {
public: public:
/**
* Method that is called when the channel was succesfully created.
* @param channel the channel that is ready
*/
virtual void onReady(Channel *channel) {}
/** /**
* An error has occured on the channel * An error has occured on the channel
* The channel is no longer usable after an error has occured on it. * The channel is no longer usable after an error has occured on it.
@ -36,108 +30,6 @@ public:
*/ */
virtual void onError(Channel *channel, const std::string &message) {} virtual void onError(Channel *channel, const std::string &message) {}
/**
* Method that is called when the channel was paused
* This is the result of a call to Channel::pause()
* @param channel the channel that is now paused
*/
virtual void onPaused(Channel *channel) {}
/**
* Method that is called when the channel was resumed
* This is the result of a call to Channel::resume()
* @param channel the channel that is no longer paused
*/
virtual void onResumed(Channel *channel) {}
/**
* Method that is called when a channel is closed
* This is the result of a call to Channel::close()
* @param channel the channel that is closed
*/
virtual void onClosed(Channel *channel) {}
/**
* Method that is called when a transaction was started
* This is the result of a call to Channel::startTransaction()
* @param channel the channel on which the transaction was started
*/
virtual void onTransactionStarted(Channel *channel) {}
/**
* Method that is called when a transaction was committed
* This is the result of a call to Channel::commitTransaction()
* @param channel the channel on which the transaction was committed
*/
virtual void onTransactionCommitted(Channel *channel) {}
/**
* Method that is called when a transaction was rolled back
* This is the result of a call to Channel::rollbackTransaction()
* @param channel the channel on which the transaction was rolled back
*/
virtual void onTransactionRolledBack(Channel *channel) {}
/**
* Method that is called when an exchange is bound
* This is the result of a call to Channel::bindExchange()
* @param channel the channel on which the exchange was bound
*/
virtual void onExchangeBound(Channel *channel) {}
/**
* Method that is called when an exchange is unbound
* This is the result of a call to Channel::unbindExchange()
* @param channel the channel on which the exchange was unbound
*/
virtual void onExchangeUnbound(Channel *channel) {}
/**
* Method that is called when an exchange is deleted
* This is the result of a call to Channel::deleteExchange()
* @param channel the channel on which the exchange was deleted
*/
virtual void onExchangeDeleted(Channel *channel) {}
/**
* Mehod that is called when an exchange is declared
* This is the result of a call to Channel::declareExchange()
* @param channel the channel on which the exchange was declared
*/
virtual void onExchangeDeclared(Channel *channel) {}
/**
* Method that is called when a queue is declared
* This is the result of a call to Channel::declareQueue()
* @param channel the channel on which the queue was declared
* @param name name of the queue
* @param messageCount number of messages in queue
* @param consumerCount number of active consumers
*/
virtual void onQueueDeclared(Channel *channel, const std::string &name, uint32_t messageCount, uint32_t consumerCount) {}
/**
* Method that is called when a queue is bound
* This is the result of a call to Channel::bindQueue()
* @param channel the channel on which the queue was bound
*/
virtual void onQueueBound(Channel *channel) {}
/**
* Method that is called when a queue is deleted
* This is the result of a call to Channel::deleteQueue()
* @param channel the channel on which the queue was deleted
* @param messageCount number of messages deleted along with the queue
*/
virtual void onQueueDeleted(Channel *channel, uint32_t messageCount) {}
/**
* Method that is called when a queue is unbound
* This is the result of a call to Channel::unbindQueue()
* @param channel the channel on which the queue was unbound
*/
virtual void onQueueUnbound(Channel *channel) {}
/** /**
* Method that is called when a queue is purged * Method that is called when a queue is purged
* This is the result of a call to Channel::purgeQueue() * This is the result of a call to Channel::purgeQueue()
@ -146,13 +38,6 @@ public:
*/ */
virtual void onQueuePurged(Channel *channel, uint32_t messageCount) {} virtual void onQueuePurged(Channel *channel, uint32_t messageCount) {}
/**
* Method that is called when the quality-of-service was changed
* This is the result of a call to Channel::setQos()
* @param channel the channel on which the qos was set
*/
virtual void onQosSet(Channel *channel) {}
/** /**
* Method that is called when a consumer was started * Method that is called when a consumer was started
* This is the result of a call to Channel::consume() * This is the result of a call to Channel::consume()
@ -194,13 +79,6 @@ public:
*/ */
virtual void onReturned(Channel *channel, const Message &message, int16_t code, const std::string &text) {} virtual void onReturned(Channel *channel, const Message &message, int16_t code, const std::string &text) {}
/**
* Method that is called when the server starts recovering messages
* This is the result of a call to Channel::recover()
* @param channel the channel on which the recover method was called
*/
virtual void onRecovering(Channel *channel) {}
}; };
/** /**

276
include/channelimpl.h
View File

@ -38,6 +38,27 @@ private:
*/ */
ChannelHandler *_handler; ChannelHandler *_handler;
/**
* Callback when the channel is ready
*/
std::function<void(Channel *channel)> _readyCallback;
/**
* Callback when the channel errors out
*/
std::function<void(Channel *channel, const std::string& message)> _errorCallback;
/**
* The callbacks waiting to be called
*/
std::deque<Deferred<>> _callbacks;
/**
* Callbacks with additional parameters
*/
std::deque<Deferred<const std::string&, uint32_t, uint32_t>> _queueDeclaredCallbacks;
std::deque<Deferred<uint32_t>> _queueRemovedCallbacks;
/** /**
* The channel number * The channel number
* @var uint16_t * @var uint16_t
@ -99,20 +120,20 @@ public:
* *
* This will stop all incoming messages * This will stop all incoming messages
* *
* This method returns true if the request to pause has been sent to the * This function returns a deferred handler. Callbacks can be installed
* broker. This does not necessarily mean that the channel is already * using onSuccess(), onError() and onFinalize() methods.
* paused.
*
* @return bool
*/ */
bool pause(); Deferred<>& pause();
/** /**
* Resume a paused channel * Resume a paused channel
* *
* @return bool * This will resume incoming messages
*
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool resume(); Deferred<>& resume();
/** /**
* Is the channel connected? * Is the channel connected?
@ -125,107 +146,154 @@ public:
/** /**
* Start a transaction * Start a transaction
* @return bool
*/ */
bool startTransaction(); Deferred<>& startTransaction();
/** /**
* Commit the current transaction * Commit the current transaction
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool commitTransaction(); Deferred<>& commitTransaction();
/** /**
* Rollback the current transaction * Rollback the current transaction
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool rollbackTransaction(); Deferred<>& rollbackTransaction();
/** /**
* declare an exchange * declare an exchange
*
* @param name name of the exchange to declare * @param name name of the exchange to declare
* @param type type of exchange * @param type type of exchange
* @param flags additional settings for the exchange * @param flags additional settings for the exchange
* @param arguments additional arguments * @param arguments additional arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool declareExchange(const std::string &name, ExchangeType type, int flags, const Table &arguments); Deferred<>& declareExchange(const std::string &name, ExchangeType type, int flags, const Table &arguments);
/** /**
* bind two exchanges * bind two exchanges
* @param source exchange which binds to target * @param source exchange which binds to target
* @param target exchange to bind to * @param target exchange to bind to
* @param routingKey routing key * @param routingKey routing key
* @param glags additional flags * @param glags additional flags
* @param arguments additional arguments for binding * @param arguments additional arguments for binding
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments); Deferred<>& bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments);
/** /**
* unbind two exchanges * unbind two exchanges
* @param source the source exchange * @param source the source exchange
* @param target the target exchange * @param target the target exchange
* @param routingkey the routing key * @param routingkey the routing key
* @param flags optional flags * @param flags optional flags
* @param arguments additional unbind arguments * @param arguments additional unbind arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool unbindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments); Deferred<>& unbindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments);
/** /**
* remove an exchange * remove an exchange
*
* @param name name of the exchange to remove * @param name name of the exchange to remove
* @param flags additional settings for deleting the exchange * @param flags additional settings for deleting the exchange
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool removeExchange(const std::string &name, int flags); Deferred<>& removeExchange(const std::string &name, int flags);
/** /**
* declare a queue * declare a queue
* @param name queue name * @param name queue name
* @param flags additional settings for the queue * @param flags additional settings for the queue
* @param arguments additional arguments * @param arguments additional arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool declareQueue(const std::string &name, int flags, const Table &arguments); Deferred<const std::string&, uint32_t, uint32_t>& declareQueue(const std::string &name, int flags, const Table &arguments);
/** /**
* Bind a queue to an exchange * Bind a queue to an exchange
*
* @param exchangeName name of the exchange to bind to * @param exchangeName name of the exchange to bind to
* @param queueName name of the queue * @param queueName name of the queue
* @param routingkey routingkey * @param routingkey routingkey
* @param flags additional flags * @param flags additional flags
* @param arguments additional arguments * @param arguments additional arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool bindQueue(const std::string &exchangeName, const std::string &queueName, const std::string &routingkey, int flags, const Table &arguments); Deferred<>& bindQueue(const std::string &exchangeName, const std::string &queueName, const std::string &routingkey, int flags, const Table &arguments);
/** /**
* Unbind a queue from an exchange * Unbind a queue from an exchange
*
* @param exchange the source exchange * @param exchange the source exchange
* @param queue the target queue * @param queue the target queue
* @param routingkey the routing key * @param routingkey the routing key
* @param arguments additional bind arguments * @param arguments additional bind arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool unbindQueue(const std::string &exchangeName, const std::string &queueName, const std::string &routingkey, const Table &arguments); Deferred<>& unbindQueue(const std::string &exchangeName, const std::string &queueName, const std::string &routingkey, const Table &arguments);
/** /**
* Purge a queue * Purge a queue
* @param queue queue to purge * @param queue queue to purge
* @param flags additional flags * @param flags additional flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*
* The onSuccess() callback that you can install should have the following signature:
*
* void myCallback(AMQP::Channel *channel, uint32_t messageCount);
*
* For example: channel.declareQueue("myqueue").onSuccess([](AMQP::Channel *channel, uint32_t messageCount) {
*
* std::cout << "Queue purged, all " << messageCount << " messages removed" << std::endl;
*
* });
*/ */
bool purgeQueue(const std::string &name, int flags); Deferred<uint32_t>& purgeQueue(const std::string &name, int flags);
/** /**
* Remove a queue * Remove a queue
* @param queue queue to remove * @param queue queue to remove
* @param flags additional flags * @param flags additional flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*
* The onSuccess() callback that you can install should have the following signature:
*
* void myCallback(AMQP::Channel *channel, uint32_t messageCount);
*
* For example: channel.declareQueue("myqueue").onSuccess([](AMQP::Channel *channel, uint32_t messageCount) {
*
* std::cout << "Queue deleted, along with " << messageCount << " messages" << std::endl;
*
* });
*/ */
bool removeQueue(const std::string &name, int flags); Deferred<uint32_t>& removeQueue(const std::string &name, int flags);
/** /**
* Publish a message to an exchange * Publish a message to an exchange
@ -251,9 +319,11 @@ public:
/** /**
* Set the Quality of Service (QOS) of the entire connection * Set the Quality of Service (QOS) of the entire connection
* @param prefetchCount maximum number of messages to prefetch * @param prefetchCount maximum number of messages to prefetch
* @return bool whether the Qos frame is sent. *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool setQos(uint16_t prefetchCount); Deferred<>& setQos(uint16_t prefetchCount);
/** /**
* Tell the RabbitMQ server that we're ready to consume messages * Tell the RabbitMQ server that we're ready to consume messages
@ -292,15 +362,19 @@ public:
/** /**
* Recover messages that were not yet ack'ed * Recover messages that were not yet ack'ed
* @param flags optional flags * @param flags optional flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool recover(int flags); Deferred<>& recover(int flags);
/** /**
* Close the current channel * Close the current channel
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool close(); Deferred<>& close();
/** /**
* Get the channel we're working on * Get the channel we're working on
@ -318,6 +392,26 @@ public:
*/ */
bool send(const Frame &frame); bool send(const Frame &frame);
/**
* Send a frame over the channel and
* get a deferred handler for it.
*
* @param frame frame to send
* @param message the message to trigger if the frame cannot be send at all
*/
Deferred<>& send(const Frame &frame, const char *message);
/**
* Send a frame over the channel and
* get a deferred handler for it.
*
* @param frame frame to send
* @param message the message to trigger if the frame cannot be send at all
* @param queue the queue to store the callbacks in
*/
template <typename... Arguments>
Deferred<Arguments...>& send(const Frame &frame, const char *message, std::deque<Deferred<Arguments...>>& queue);
/** /**
* Report to the handler that the channel is closed * Report to the handler that the channel is closed
*/ */
@ -327,25 +421,20 @@ public:
_state = state_closed; _state = state_closed;
// inform handler // inform handler
if (_handler) _handler->onClosed(_parent); reportSuccess();
} }
/** /**
* Report to the handler that the channel is paused * Report success
*
* This function is called to report success for all
* cases where the callback does not receive any parameters
*/ */
void reportPaused() void reportSuccess()
{ {
// inform handler // report success for the oldest request
if (_handler) _handler->onPaused(_parent); _callbacks.front().success();
} _callbacks.pop_front();
/**
* Report to the handler that the channel is resumed
*/
void reportResumed()
{
// inform handler
if (_handler) _handler->onResumed(_parent);
} }
/** /**
@ -354,7 +443,7 @@ public:
void reportReady() void reportReady()
{ {
// inform handler // inform handler
if (_handler) _handler->onReady(_parent); if (_readyCallback) _readyCallback(_parent);
} }
/** /**
@ -367,39 +456,12 @@ public:
_state = state_closed; _state = state_closed;
// inform handler // inform handler
if (_handler) _handler->onError(_parent, message); if (_errorCallback) _errorCallback(_parent, message);
}
/** // and all waiting deferred callbacks
* Report that the exchange is succesfully declared for (auto &deferred : _callbacks) deferred.error(message);
*/ for (auto &deferred : _queueDeclaredCallbacks) deferred.error(message);
void reportExchangeDeclared() for (auto &deferred : _queueRemovedCallbacks) deferred.error(message);
{
if (_handler) _handler->onExchangeDeclared(_parent);
}
/**
* Report that the exchange is succesfully deleted
*/
void reportExchangeDeleted()
{
if (_handler) _handler->onExchangeDeleted(_parent);
}
/**
* Report that the exchange is bound
*/
void reportExchangeBound()
{
if (_handler) _handler->onExchangeBound(_parent);
}
/**
* Report that the exchange is unbound
*/
void reportExchangeUnbound()
{
if (_handler) _handler->onExchangeUnbound(_parent);
} }
/** /**
@ -410,23 +472,9 @@ public:
*/ */
void reportQueueDeclared(const std::string &queueName, uint32_t messageCount, uint32_t consumerCount) void reportQueueDeclared(const std::string &queueName, uint32_t messageCount, uint32_t consumerCount)
{ {
if (_handler) _handler->onQueueDeclared(_parent, queueName, messageCount, consumerCount); // report success for the oldest queue declare callbacks
} _queueDeclaredCallbacks.front().success(queueName, messageCount, consumerCount);
_queueDeclaredCallbacks.pop_front();
/**
* Report that a queue was succesfully bound
*/
void reportQueueBound()
{
if (_handler) _handler->onQueueBound(_parent);
}
/**
* Report that a queue was succesfully unbound
*/
void reportQueueUnbound()
{
if (_handler) _handler->onQueueUnbound(_parent);
} }
/** /**
@ -435,7 +483,9 @@ public:
*/ */
void reportQueueDeleted(uint32_t messageCount) void reportQueueDeleted(uint32_t messageCount)
{ {
if (_handler) _handler->onQueueDeleted(_parent, messageCount); // report success for the oldest queue remove callbacks
_queueRemovedCallbacks.front().success(messageCount);
_queueRemovedCallbacks.pop_front();
} }
/** /**
@ -444,15 +494,9 @@ public:
*/ */
void reportQueuePurged(uint32_t messageCount) void reportQueuePurged(uint32_t messageCount)
{ {
if (_handler) _handler->onQueuePurged(_parent, messageCount); // report success for the oldest queue remove callbacks
} _queueRemovedCallbacks.front().success(messageCount);
_queueRemovedCallbacks.pop_front();
/**
* Report that the qos has been set
*/
void reportQosSet()
{
if (_handler) _handler->onQosSet(_parent);
} }
/** /**
@ -478,14 +522,6 @@ public:
*/ */
void reportMessage(); void reportMessage();
/**
* Report that the recover operation has started
*/
void reportRecovering()
{
if (_handler) _handler->onRecovering(_parent);
}
/** /**
* Create an incoming message * Create an incoming message
* @param frame * @param frame

18
include/connection.h
View File

@ -22,6 +22,24 @@ private:
*/ */
ConnectionImpl _implementation; ConnectionImpl _implementation;
/**
* Function to execute code after a certain timeout.
*
* If the timeout is 0, the code is supposed to be run
* in the next iteration of the event loop.
*
* This is a simple placeholder function that will just
* execute the code immediately, it should be overridden
* by the timeout function the used event loop has.
*
* @param timeout the amount of time to wait
* @param callback the callback to execute after the timeout
*/
std::function<void(double timeout, const std::function<void()>)> _timeoutHandler = [](double timeout, const std::function<void()>& callback) {
// execute callback immediately
callback();
};
public: public:
/** /**
* Construct an AMQP object based on full login data * Construct an AMQP object based on full login data

1
include/connectionimpl.h
View File

@ -100,7 +100,6 @@ protected:
*/ */
bool sendClose(); bool sendClose();
private: private:
/** /**
* Construct an AMQP object based on full login data * Construct an AMQP object based on full login data

179
include/deferred.h Normal file
View File

@ -0,0 +1,179 @@
/**
* Deferred.h
*
* Class describing a set of actions that could
* possibly happen in the future that can be
* caught.
*
* @copyright 2014 Copernica BV
*/
/**
* Set up namespace
*/
namespace AMQP {
// forward declaration
class ChannelImpl;
/**
* Class definition
*/
template <typename... Arguments>
class Deferred
{
private:
/**
* The channel we operate under
*/
Channel *_channel;
/**
* Do we already know we failed?
*/
bool _failed;
/**
* Callback to execute on success
*/
std::function<void(Channel *channel, Arguments ...parameters)> _successCallback;
/**
* Callback to execute on failure
*/
std::function<void(Channel *channel, const std::string& error)> _errorCallback;
/**
* Callback to execute either way
*/
std::function<void(Channel *channel, const std::string& error)> _finalizeCallback;
/**
* The channel implementation may call our
* private members and construct us
*/
friend class ChannelImpl;
/**
* Indicate success
*
* @param parameters... the extra parameters relevant for this deferred handler
*/
void success(Arguments ...parameters)
{
// execute callbacks if registered
if (_successCallback) _successCallback(_channel, parameters...);
if (_finalizeCallback) _finalizeCallback(_channel, "");
}
/**
* Indicate failure
*
* @param error description of the error that occured
*/
void error(const std::string& error)
{
// we are now in a failed state
_failed = true;
// execute callbacks if registered
if (_errorCallback) _errorCallback(_channel, error);
if (_finalizeCallback) _finalizeCallback(_channel, error);
}
/**
* Private constructor that can only be called
* from within the channel implementation
*
* @param channel the channel we operate under
* @param boolea are we already failed?
*/
Deferred(Channel *channel, bool failed = false) :
_channel(channel),
_failed(failed)
{}
public:
/**
* Deleted copy constructor
*/
Deferred(const Deferred& that) = delete;
/**
* Move constructor
*/
Deferred(Deferred&& that) :
_successCallback(std::move(that._successCallback)),
_errorCallback(std::move(that._errorCallback)),
_finalizeCallback(std::move(that._finalizeCallback))
{}
/**
* Cast to a boolean
*/
operator bool ()
{
return !_failed;
}
/**
* Register a function to be called
* if and when the operation succesfully
* completes.
*
* Only one callback can be registered at a time.
* Successive calls to this function will clear
* callbacks registered before.
*
* @param callback the callback to execute
*/
Deferred& onSuccess(const std::function<void(Channel *channel, Arguments ...parameters)>& callback)
{
// store callback
_successCallback = callback;
return *this;
}
/**
* Register a function to be called
* if and when the operation fails.
*
* Only one callback can be registered at a time.
* Successive calls to this function will clear
* callbacks registered before.
*
* @param callback the callback to execute
*/
Deferred& onError(const std::function<void(Channel *channel, const std::string& error)>& callback)
{
// store callback
_errorCallback = callback;
return *this;
}
/**
* Register a function to be called
* if and when the operation completes
* or fails. This function will be called
* either way.
*
* In the case of success, the provided
* error parameter will be an empty string.
*
* Only one callback can be registered at at time.
* Successive calls to this function will clear
* callbacks registered before.
*
* @param callback the callback to execute
*/
Deferred& onFinalize(const std::function<void(Channel *channel, const std::string& error)>& callback)
{
// store callback
_finalizeCallback = callback;
return *this;
}
};
/**
* End namespace
*/
}

2
src/basicqosokframe.h
View File

@ -67,7 +67,7 @@ public:
if (!channel) return false; if (!channel) return false;
// report // report
channel->reportQosSet(); channel->reportSuccess();
// done // done
return true; return true;

2
src/basicrecoverokframe.h
View File

@ -72,7 +72,7 @@ public:
if (!channel) return false; if (!channel) return false;
// report // report
channel->reportRecovering(); channel->reportSuccess();
// done // done
return true; return true;

5
src/channelflowokframe.h
View File

@ -92,9 +92,8 @@ public:
// channel does not exist // channel does not exist
if (!channel) return false; if (!channel) return false;
// is the flow active? // report success for the call
if (active()) channel->reportResumed(); channel->reportSuccess();
else channel->reportPaused();
// done // done
return true; return true;

234
src/channelimpl.cpp
View File

@ -96,90 +96,98 @@ ChannelImpl::~ChannelImpl()
* *
* This will stop all incoming messages * This will stop all incoming messages
* *
* This method returns true if the request to pause has been sent to the * This function returns a deferred handler. Callbacks can be installed
* broker. This does not necessarily mean that the channel is already * using onSuccess(), onError() and onFinalize() methods.
* paused.
*
* @return bool
*/ */
bool ChannelImpl::pause() Deferred<>& ChannelImpl::pause()
{ {
// send a flow frame // send a channel flow frame
return send(ChannelFlowFrame(_id, false)); return send(ChannelFlowFrame(_id, false), "Cannot send channel flow frame");
} }
/** /**
* Resume a paused channel * Resume a paused channel
* *
* @return bool * This will resume incoming messages
*
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::resume() Deferred<>& ChannelImpl::resume()
{ {
// send a flow frame // send a channel flow frame
return send(ChannelFlowFrame(_id, true)); return send(ChannelFlowFrame(_id, true), "Cannot send channel flow frame");
} }
/** /**
* Start a transaction * Start a transaction
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::startTransaction() Deferred<>& ChannelImpl::startTransaction()
{ {
// send a flow frame // send a transaction frame
return send(TransactionSelectFrame(_id)); return send(TransactionSelectFrame(_id), "Cannot send transaction start frame");
} }
/** /**
* Commit the current transaction * Commit the current transaction
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::commitTransaction() Deferred<>& ChannelImpl::commitTransaction()
{ {
// send a flow frame // send a transaction frame
return send(TransactionCommitFrame(_id)); return send(TransactionCommitFrame(_id), "Cannot send transaction commit frame");
} }
/** /**
* Rollback the current transaction * Rollback the current transaction
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::rollbackTransaction() Deferred<>& ChannelImpl::rollbackTransaction()
{ {
// send a flow frame // send a transaction frame
return send(TransactionRollbackFrame(_id)); return send(TransactionRollbackFrame(_id), "Cannot send transaction commit frame");
} }
/** /**
* Close the current channel * Close the current channel
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::close() Deferred<>& ChannelImpl::close()
{ {
// channel could be dead after send operation, we need to monitor that // channel could be dead after send operation, we need to monitor that
Monitor monitor(this); Monitor monitor(this);
// send a flow frame // send a channel close frame
if (!send(ChannelCloseFrame(_id))) return false; auto &handler = send(ChannelCloseFrame(_id), "Cannot send channel close frame");
// leap out if channel was destructed // was the frame sent and are we still alive?
if (!monitor.valid()) return true; if (handler && monitor.valid()) _state = state_closing;
// now it is closing
_state = state_closing;
// done // done
return true; return handler;
} }
/** /**
* declare an exchange * declare an exchange
* @param name name of the exchange to declare * @param name name of the exchange to declare
* @param type type of exchange * @param type type of exchange
* @param flags additional settings for the exchange * @param flags additional settings for the exchange
* @param arguments additional arguments * @param arguments additional arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::declareExchange(const std::string &name, ExchangeType type, int flags, const Table &arguments) Deferred<>& ChannelImpl::declareExchange(const std::string &name, ExchangeType type, int flags, const Table &arguments)
{ {
// convert exchange type // convert exchange type
std::string exchangeType; std::string exchangeType;
@ -189,49 +197,58 @@ bool ChannelImpl::declareExchange(const std::string &name, ExchangeType type, in
if (type == ExchangeType::headers)exchangeType = "headers"; if (type == ExchangeType::headers)exchangeType = "headers";
// send declare exchange frame // send declare exchange frame
return send(ExchangeDeclareFrame(_id, name, exchangeType, flags & passive, flags & durable, flags & nowait, arguments)); return send(ExchangeDeclareFrame(_id, name, exchangeType, flags & passive, flags & durable, flags & nowait, arguments), "Cannot send exchange declare frame");
} }
/** /**
* bind an exchange * bind an exchange
*
* @param source exchange which binds to target * @param source exchange which binds to target
* @param target exchange to bind to * @param target exchange to bind to
* @param routingKey routing key * @param routingKey routing key
* @param flags additional flags * @param flags additional flags
* @param arguments additional arguments for binding * @param arguments additional arguments for binding
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments) Deferred<>& ChannelImpl::bindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments)
{ {
// send exchange bind frame // send exchange bind frame
return send(ExchangeBindFrame(_id, target, source, routingkey, flags & nowait, arguments)); return send(ExchangeBindFrame(_id, target, source, routingkey, flags & nowait, arguments), "Cannot send exchange bind frame");
} }
/** /**
* unbind two exchanges * unbind two exchanges
*
* @param source the source exchange * @param source the source exchange
* @param target the target exchange * @param target the target exchange
* @param routingkey the routing key * @param routingkey the routing key
* @param flags optional flags * @param flags optional flags
* @param arguments additional unbind arguments * @param arguments additional unbind arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::unbindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments) Deferred<>& ChannelImpl::unbindExchange(const std::string &source, const std::string &target, const std::string &routingkey, int flags, const Table &arguments)
{ {
// send exchange unbind frame // send exchange unbind frame
return send(ExchangeUnbindFrame(_id, target, source, routingkey, flags & nowait, arguments)); return send(ExchangeUnbindFrame(_id, target, source, routingkey, flags & nowait, arguments), "Cannot send exchange unbind frame");
} }
/** /**
* remove an exchange * remove an exchange
*
* @param name name of the exchange to remove * @param name name of the exchange to remove
* @param flags additional settings for deleting the exchange * @param flags additional settings for deleting the exchange
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::removeExchange(const std::string &name, int flags) Deferred<>& ChannelImpl::removeExchange(const std::string &name, int flags)
{ {
// send delete exchange frame // send delete exchange frame
return send(ExchangeDeleteFrame(_id, name, flags & ifunused, flags & nowait)); return send(ExchangeDeleteFrame(_id, name, flags & ifunused, flags & nowait), "Cannot send exchange delete frame");
} }
/** /**
@ -239,65 +256,97 @@ bool ChannelImpl::removeExchange(const std::string &name, int flags)
* @param name queue name * @param name queue name
* @param flags additional settings for the queue * @param flags additional settings for the queue
* @param arguments additional arguments * @param arguments additional arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::declareQueue(const std::string &name, int flags, const Table &arguments) Deferred<const std::string&, uint32_t, uint32_t>& ChannelImpl::declareQueue(const std::string &name, int flags, const Table &arguments)
{ {
// send the queuedeclareframe // send the queuedeclareframe
return send(QueueDeclareFrame(_id, name, flags & passive, flags & durable, flags & exclusive, flags & autodelete, flags & nowait, arguments)); return send(QueueDeclareFrame(_id, name, flags & passive, flags & durable, flags & exclusive, flags & autodelete, flags & nowait, arguments), "Cannot send queue declare frame", _queueDeclaredCallbacks);
} }
/** /**
* Bind a queue to an exchange * Bind a queue to an exchange
*
* @param exchangeName name of the exchange to bind to * @param exchangeName name of the exchange to bind to
* @param queueName name of the queue * @param queueName name of the queue
* @param routingkey routingkey * @param routingkey routingkey
* @param flags additional flags * @param flags additional flags
* @param arguments additional arguments * @param arguments additional arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::bindQueue(const std::string &exchangeName, const std::string &queueName, const std::string &routingkey, int flags, const Table &arguments) Deferred<>& ChannelImpl::bindQueue(const std::string &exchangeName, const std::string &queueName, const std::string &routingkey, int flags, const Table &arguments)
{ {
// send the bind queue frame // send the bind queue frame
return send(QueueBindFrame(_id, queueName, exchangeName, routingkey, flags & nowait, arguments)); return send(QueueBindFrame(_id, queueName, exchangeName, routingkey, flags & nowait, arguments), "Cannot send queue bind frame");
} }
/** /**
* Unbind a queue from an exchange * Unbind a queue from an exchange
*
* @param exchange the source exchange * @param exchange the source exchange
* @param queue the target queue * @param queue the target queue
* @param routingkey the routing key * @param routingkey the routing key
* @param arguments additional bind arguments * @param arguments additional bind arguments
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::unbindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, const Table &arguments) Deferred<>& ChannelImpl::unbindQueue(const std::string &exchange, const std::string &queue, const std::string &routingkey, const Table &arguments)
{ {
// send the unbind queue frame // send the unbind queue frame
return send(QueueUnbindFrame(_id, queue, exchange, routingkey, arguments)); return send(QueueUnbindFrame(_id, queue, exchange, routingkey, arguments), "Cannot send queue unbind frame");
} }
/** /**
* Purge a queue * Purge a queue
* @param queue queue to purge * @param queue queue to purge
* @param flags additional flags * @param flags additional flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*
* The onSuccess() callback that you can install should have the following signature:
*
* void myCallback(AMQP::Channel *channel, uint32_t messageCount);
*
* For example: channel.declareQueue("myqueue").onSuccess([](AMQP::Channel *channel, uint32_t messageCount) {
*
* std::cout << "Queue purged, all " << messageCount << " messages removed" << std::endl;
*
* });
*/ */
bool ChannelImpl::purgeQueue(const std::string &name, int flags) Deferred<uint32_t>& ChannelImpl::purgeQueue(const std::string &name, int flags)
{ {
// send the queue purge frame // send the queue purge frame
return send(QueuePurgeFrame(_id, name, flags & nowait)); return send(QueuePurgeFrame(_id, name, flags & nowait), "Cannot send queue purge frame", _queueRemovedCallbacks);
} }
/** /**
* Remove a queue * Remove a queue
* @param queue queue to remove * @param queue queue to remove
* @param flags additional flags * @param flags additional flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*
* The onSuccess() callback that you can install should have the following signature:
*
* void myCallback(AMQP::Channel *channel, uint32_t messageCount);
*
* For example: channel.declareQueue("myqueue").onSuccess([](AMQP::Channel *channel, uint32_t messageCount) {
*
* std::cout << "Queue deleted, along with " << messageCount << " messages" << std::endl;
*
* });
*/ */
bool ChannelImpl::removeQueue(const std::string &name, int flags) Deferred<uint32_t>& ChannelImpl::removeQueue(const std::string &name, int flags)
{ {
// send the remove queue frame // send the remove queue frame
return send(QueueDeleteFrame(_id, name, flags & ifunused, flags & ifempty, flags & nowait)); return send(QueueDeleteFrame(_id, name, flags & ifunused, flags & ifempty, flags & nowait), "Cannot send remove queue frame", _queueRemovedCallbacks);
} }
/** /**
@ -367,12 +416,14 @@ bool ChannelImpl::publish(const std::string &exchange, const std::string &routin
/** /**
* Set the Quality of Service (QOS) for this channel * Set the Quality of Service (QOS) for this channel
* @param prefetchCount maximum number of messages to prefetch * @param prefetchCount maximum number of messages to prefetch
* @return bool whether the Qos frame is sent. *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::setQos(uint16_t prefetchCount) Deferred<>& ChannelImpl::setQos(uint16_t prefetchCount)
{ {
// send a qos frame // send a qos frame
return send(BasicQosFrame(_id, prefetchCount, false)); return send(BasicQosFrame(_id, prefetchCount, false), "Cannot send basic QOS frame");
} }
/** /**
@ -427,12 +478,14 @@ bool ChannelImpl::reject(uint64_t deliveryTag, int flags)
/** /**
* Recover un-acked messages * Recover un-acked messages
* @param flags optional flags * @param flags optional flags
* @return bool *
* This function returns a deferred handler. Callbacks can be installed
* using onSuccess(), onError() and onFinalize() methods.
*/ */
bool ChannelImpl::recover(int flags) Deferred<>& ChannelImpl::recover(int flags)
{ {
// send a nack frame // send a nack frame
return send(BasicRecoverFrame(_id, flags & requeue)); return send(BasicRecoverFrame(_id, flags & requeue), "Cannot send basic recover frame");
} }
/** /**
@ -449,6 +502,53 @@ bool ChannelImpl::send(const Frame &frame)
return _connection->send(frame); return _connection->send(frame);
} }
/**
* Send a frame over the channel and
* get a deferred handler for it.
*
* @param frame frame to send
* @param message the message to trigger if the frame cannot be send at all
*/
Deferred<>& ChannelImpl::send(const Frame &frame, const char *message)
{
// use the generic implementation
return send<>(frame, message, _callbacks);
}
/**
* Send a frame over the channel and
* get a deferred handler for it.
*
* @param frame frame to send
* @param message the message to trigger if the frame cannot be send at all
* @param queue the queue to store the callbacks in
*/
template <typename... Arguments>
Deferred<Arguments...>& ChannelImpl::send(const Frame &frame, const char *message, std::deque<Deferred<Arguments...>>& queue)
{
// create a new deferred handler and get a pointer to it
queue.push_back(Deferred<Arguments...>(_parent));
auto *handler = &queue.back();
// send the frame over the channel
if (!send(frame))
{
// we can immediately put the handler in failed state
handler->_failed = true;
// the frame could not be send
// we should register an error
// on the handler, but only after
// a timeout, so a handler can
// be attached first
// TODO
}
// return the new handler
return *handler;
}
/** /**
* Report the received message * Report the received message
*/ */

2
src/exchangebindokframe.h
View File

@ -67,7 +67,7 @@ public:
if(!channel) return false; if(!channel) return false;
// report to handler // report to handler
channel->reportExchangeBound(); channel->reportSuccess();
// done // done
return true; return true;

2
src/exchangedeclareokframe.h
View File

@ -70,7 +70,7 @@ public:
if(!channel) return false; if(!channel) return false;
// report exchange declare ok // report exchange declare ok
channel->reportExchangeDeclared(); channel->reportSuccess();
// done // done
return true; return true;

2
src/exchangedeleteokframe.h
View File

@ -71,7 +71,7 @@ public:
if(!channel) return false; if(!channel) return false;
// report to handler // report to handler
channel->reportExchangeDeleted(); channel->reportSuccess();
// done // done
return true; return true;

2
src/exchangeunbindokframe.h
View File

@ -68,7 +68,7 @@ public:
if(!channel) return false; if(!channel) return false;
// report to handler // report to handler
channel->reportExchangeUnbound(); channel->reportSuccess();
// done // done
return true; return true;

5
src/methodframe.h
View File

@ -75,13 +75,10 @@ public:
* @param connection The connection over which it was received * @param connection The connection over which it was received
* @return bool Was it succesfully processed? * @return bool Was it succesfully processed?
*/ */
virtual bool process(ConnectionImpl *connection) override virtual bool process[[ noreturn ]](ConnectionImpl *connection) override
{ {
// this is an exception // this is an exception
throw ProtocolException("unimplemented frame type " + std::to_string(type()) + " class " + std::to_string(classID()) + " method " + std::to_string(methodID())); throw ProtocolException("unimplemented frame type " + std::to_string(type()) + " class " + std::to_string(classID()) + " method " + std::to_string(methodID()));
// unreachable
return false;
} }
}; };

2
src/queuebindokframe.h
View File

@ -69,7 +69,7 @@ public:
if(!channel) return false; if(!channel) return false;
// report to handler // report to handler
channel->reportQueueBound(); channel->reportSuccess();
// done // done
return true; return true;

2
src/queueunbindokframe.h
View File

@ -73,7 +73,7 @@ public:
if(!channel) return false; if(!channel) return false;
// report queue unbind success // report queue unbind success
channel->reportQueueUnbound(); channel->reportSuccess();
// done // done
return true; return true;

20
src/transactioncommitokframe.h
View File

@ -59,6 +59,26 @@ public:
{ {
return 21; return 21;
} }
/**
* Process the frame
* @param connection The connection over which it was received
* @return bool Was it succesfully processed?
*/
virtual bool process(ConnectionImpl *connection) override
{
// we need the appropriate channel
ChannelImpl *channel = connection->channel(this->channel());
// channel does not exist
if(!channel) return false;
// report that the channel is open
channel->reportSuccess();
// done
return true;
}
}; };
/** /**

20
src/transactionrollbackokframe.h
View File

@ -59,6 +59,26 @@ public:
{ {
return 31; return 31;
} }
/**
* Process the frame
* @param connection The connection over which it was received
* @return bool Was it succesfully processed?
*/
virtual bool process(ConnectionImpl *connection) override
{
// we need the appropriate channel
ChannelImpl *channel = connection->channel(this->channel());
// channel does not exist
if(!channel) return false;
// report that the channel is open
channel->reportSuccess();
// done
return true;
}
}; };
/** /**

20
src/transactionselectokframe.h
View File

@ -59,6 +59,26 @@ public:
{ {
return 11; return 11;
} }
/**
* Process the frame
* @param connection The connection over which it was received
* @return bool Was it succesfully processed?
*/
virtual bool process(ConnectionImpl *connection) override
{
// we need the appropriate channel
ChannelImpl *channel = connection->channel(this->channel());
// channel does not exist
if(!channel) return false;
// report that the channel is open
channel->reportSuccess();
// done
return true;
}
}; };
/** /**