cppamqp/src/cppamqp/channel.cpp

#include <cppamqp/types.inl>
#include <cppamqp/helper.inl>
#include <cppamqp/channel.inl>
#include <cppamqp/exception.inl>
#include <cppamqp/connection.inl>

using namespace ::cppamqp;

channel::internal::internal(const cppamqp::connection& p_connection, channel_number p_handle)
    : connection(p_connection)
    , handle    (p_handle)
{
    if (handle <= 0)
        throw utl::argument_exception("handle", "handle must be greater than 0");
    amqp_channel_open(connection.handle(), handle);
    __impl::check_and_raise(
        amqp_get_rpc_reply(connection.handle()),
        std::string("unable to open channel ") + std::to_string(handle),
        false);
}

channel::internal::~internal()
{
    if (handle <= 0 || !static_cast<bool>(connection))
        return;
    __impl::check_and_raise(
        amqp_channel_close(connection.handle(), handle, AMQP_REPLY_SUCCESS),
        "unable to close amqp channel",
        true);
}

queue_declaration channel::declare_queue(const std::string& name, const queue_flags& flags, const table& args)
{
    size_t i = 0;
    amqp_table_t arg_table;
    amqp_table_entry_t entries[args.size()];
    arg_table.num_entries = static_cast<int>(args.size());
    arg_table.entries = &entries[0];
    for (auto& kvp : args)
    {
        arg_table.entries[i].key   = __impl::make_bytes(kvp.first);
        arg_table.entries[i].value = kvp.second.get();
        ++i;
    }

    auto ret = amqp_queue_declare(
                    connection().handle(),
                    handle(),
                    __impl::make_bytes(name),
                    flags.is_set(queue_flag::passive) ? 1 : 0,
                    flags.is_set(queue_flag::durable) ? 1 : 0,
                    flags.is_set(queue_flag::exclusive) ? 1 : 0,
                    flags.is_set(queue_flag::auto_delete) ? 1 : 0,
                    arg_table);
    __impl::check_and_raise(
        amqp_get_rpc_reply(connection().handle()),
        std::string("unable to declare queue '") + name + "'",
        false);
    if (!ret)
        throw exception(std::string("unable to declare queue '") + name + "'");
    return queue_declaration
    {
        std::string(static_cast<const char*>(ret->queue.bytes), ret->queue.len),
        ret->message_count,
        ret->consumer_count
    };
}

void channel::bind_queue(const std::string& queue, const std::string& exchange, const std::string& routing_key)
{
    amqp_queue_bind(
        connection().handle(),
        handle(),
        __impl::make_bytes(queue),
        __impl::make_bytes(exchange),
        __impl::make_bytes(routing_key),
        amqp_empty_table);
    __impl::check_and_raise(
        amqp_get_rpc_reply(connection().handle()),
        static_cast<std::ostringstream&>(std::ostringstream() <<
            "unable to bind queue '"    << queue <<
            "' to '"                    << exchange <<
            "' using '"                 << routing_key <<
            "'").str(),
        false);
}

void channel::publish(
    const std::string& exchange,
    const std::string& routing_key,
    const publish_flags& flags,
    const std::string& message,
    const publish_options* options)
{
    __impl::check_and_raise(
        amqp_basic_publish(
            connection().handle(),
            handle(),
            __impl::make_bytes(exchange),
            __impl::make_bytes(routing_key),
            flags.is_set(publish_flag::mandatory) ? 1 : 0,
            flags.is_set(publish_flag::immediate) ? 1 : 0,
            options ? options->properties() : nullptr,
            __impl::make_bytes(message)),
        "error while publishing message",
        false);
}

std::string channel::consume(const std::string& queue, const std::string& consumer_tag, const consume_flags& flags)
{
    auto ret = amqp_basic_consume(
        connection().handle(),
        handle(),
        __impl::make_bytes(queue),
        __impl::make_bytes(consumer_tag),
        flags.is_set(consume_flag::no_local)   ? 1 : 0,
        flags.is_set(consume_flag::no_ack)     ? 1 : 0,
        flags.is_set(consume_flag::exclusive) ? 1 : 0,
        amqp_empty_table);
    __impl::check_and_raise(
        amqp_get_rpc_reply(connection().handle()),
        static_cast<std::ostringstream&>(std::ostringstream() << "unable to consume from queue '" << queue << "'").str(),
        false);
    if (!ret)
        throw exception(std::string("unable to consume from queue '") + queue + "'");
    return __impl::from_bytes(ret->consumer_tag);
}

void channel::qos(uint32_t prefetch_size, uint16_t prefetch_count, bool global)
{
    auto ret = amqp_basic_qos(
        connection().handle(),
        handle(),
        prefetch_size,
        prefetch_count,
        global);
    __impl::check_and_raise(
        amqp_get_rpc_reply(connection().handle()),
        static_cast<std::ostringstream&>(std::ostringstream() << "unable to set QOS values").str(),
        false);
    if (!ret)
        throw exception("unable to set QOS values");
}

void channel::close(int status)
{
    if (!_internal)
        return;
    __impl::check_and_raise(
        amqp_channel_close(connection().handle(), handle(), status),
        "unable to close channel",
        false);
    _internal->handle = 0;
    _internal.reset();
}