#include #include #include #include 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) 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) { 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, amqp_empty_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(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() << "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() << "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::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(); }