diff options
author | Tony Garnock-Jones <tonyg@kcbbs.gen.nz> | 2009-08-27 12:06:27 +0100 |
---|---|---|
committer | Tony Garnock-Jones <tonyg@kcbbs.gen.nz> | 2009-08-27 12:06:27 +0100 |
commit | 62fcf89ed4e4356033b65a9a3a99f47ff2e9581a (patch) | |
tree | 73d56ad6484e5c9e362969e881df921a59853198 /examples | |
parent | 1a2c6dd4f82bf5610b5aca25d60ef9c228a810f1 (diff) | |
download | rabbitmq-c-62fcf89ed4e4356033b65a9a3a99f47ff2e9581a.tar.gz |
Add amqp_listenq, like amqp_listen but for queues rather than exchanges
Diffstat (limited to 'examples')
-rw-r--r-- | examples/Makefile.am | 3 | ||||
-rw-r--r-- | examples/amqp_listenq.c | 128 |
2 files changed, 130 insertions, 1 deletions
diff --git a/examples/Makefile.am b/examples/Makefile.am index 6ac915a..dbbb9f7 100644 --- a/examples/Makefile.am +++ b/examples/Makefile.am @@ -1,5 +1,5 @@ bin_PROGRAMS = amqp_sendstring amqp_exchange_declare amqp_listen amqp_producer amqp_consumer \ - amqp_unbind amqp_bind + amqp_unbind amqp_bind amqp_listenq AM_CFLAGS = -I$(top_srcdir)/librabbitmq AM_LDFLAGS = $(top_builddir)/librabbitmq/librabbitmq.la @@ -13,3 +13,4 @@ amqp_producer_SOURCES = amqp_producer.c example_utils.c amqp_consumer_SOURCES = amqp_consumer.c example_utils.c amqp_unbind_SOURCES = amqp_unbind.c example_utils.c amqp_bind_SOURCES = amqp_bind.c example_utils.c +amqp_listenq_SOURCES = amqp_listenq.c example_utils.c diff --git a/examples/amqp_listenq.c b/examples/amqp_listenq.c new file mode 100644 index 0000000..b6d37fc --- /dev/null +++ b/examples/amqp_listenq.c @@ -0,0 +1,128 @@ +#include <stdlib.h> +#include <stdio.h> +#include <string.h> +#include <errno.h> + +#include <stdint.h> +#include <amqp.h> +#include <amqp_framing.h> + +#include <unistd.h> +#include <assert.h> + +#include "example_utils.h" + +/* Private: compiled out in NDEBUG mode */ +extern void amqp_dump(void const *buffer, size_t len); + +int main(int argc, char const * const *argv) { + char const *hostname; + int port; + char const *queuename; + + int sockfd; + amqp_connection_state_t conn; + + if (argc < 4) { + fprintf(stderr, "Usage: amqp_listenq host port queuename\n"); + return 1; + } + + hostname = argv[1]; + port = atoi(argv[2]); + queuename = argv[3]; + + conn = amqp_new_connection(); + + die_on_error(sockfd = amqp_open_socket(hostname, port), "Opening socket"); + amqp_set_sockfd(conn, sockfd); + die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), + "Logging in"); + amqp_channel_open(conn, 1); + die_on_amqp_error(amqp_rpc_reply, "Opening channel"); + + amqp_basic_consume(conn, 1, amqp_cstring_bytes(queuename), AMQP_EMPTY_BYTES, 0, 0, 0); + die_on_amqp_error(amqp_rpc_reply, "Consuming"); + + { + amqp_frame_t frame; + int result; + + amqp_basic_deliver_t *d; + amqp_basic_properties_t *p; + size_t body_target; + size_t body_received; + + while (1) { + amqp_maybe_release_buffers(conn); + result = amqp_simple_wait_frame(conn, &frame); + printf("Result %d\n", result); + if (result <= 0) + break; + + printf("Frame type %d, channel %d\n", frame.frame_type, frame.channel); + if (frame.frame_type != AMQP_FRAME_METHOD) + continue; + + printf("Method %s\n", amqp_method_name(frame.payload.method.id)); + if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD) + continue; + + d = (amqp_basic_deliver_t *) frame.payload.method.decoded; + printf("Delivery %u, exchange %.*s routingkey %.*s\n", + (unsigned) d->delivery_tag, + (int) d->exchange.len, (char *) d->exchange.bytes, + (int) d->routing_key.len, (char *) d->routing_key.bytes); + + result = amqp_simple_wait_frame(conn, &frame); + if (result <= 0) + break; + + if (frame.frame_type != AMQP_FRAME_HEADER) { + fprintf(stderr, "Expected header!"); + abort(); + } + p = (amqp_basic_properties_t *) frame.payload.properties.decoded; + if (p->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { + printf("Content-type: %.*s\n", + (int) p->content_type.len, (char *) p->content_type.bytes); + } + printf("----\n"); + + body_target = frame.payload.properties.body_size; + body_received = 0; + + while (body_received < body_target) { + result = amqp_simple_wait_frame(conn, &frame); + if (result <= 0) + break; + + if (frame.frame_type != AMQP_FRAME_BODY) { + fprintf(stderr, "Expected body!"); + abort(); + } + + body_received += frame.payload.body_fragment.len; + assert(body_received <= body_target); + + amqp_dump(frame.payload.body_fragment.bytes, + frame.payload.body_fragment.len); + } + + if (body_received != body_target) { + /* Can only happen when amqp_simple_wait_frame returns <= 0 */ + /* We break here to close the connection */ + break; + } + + amqp_basic_ack(conn, 1, d->delivery_tag, 0); + } + } + + die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel"); + die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection"); + amqp_destroy_connection(conn); + die_on_error(close(sockfd), "Closing socket"); + + return 0; +} |