|
- #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)
- {
- 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<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();
- }
|