From b457754d24725da1cb8ed8c1fcf82114496161c6 Mon Sep 17 00:00:00 2001 From: David Wragg Date: Tue, 25 May 2010 10:33:16 +0100 Subject: Move the common_consume code into consume.c 95% of the options provided by the common_consume code don't actually make sense for amqp-get, and now I find myself documenting them, it seems more sensible to restrict them to amqp-consume. --- tools/Makefile.am | 6 +- tools/common.c | 2 +- tools/common_consume.c | 160 ------------------------------------------------- tools/common_consume.h | 54 ----------------- tools/consume.c | 119 +++++++++++++++++++++++++++++++++--- tools/get.c | 16 +++-- tools/publish.c | 2 +- 7 files changed, 127 insertions(+), 232 deletions(-) delete mode 100644 tools/common_consume.c delete mode 100644 tools/common_consume.h (limited to 'tools') diff --git a/tools/Makefile.am b/tools/Makefile.am index cf14ba1..dc9603b 100644 --- a/tools/Makefile.am +++ b/tools/Makefile.am @@ -5,8 +5,8 @@ AM_LDFLAGS = $(top_builddir)/librabbitmq/librabbitmq.la LDADD=$(LIBPOPT) -noinst_HEADERS = common.h common_consume.h +noinst_HEADERS = common.h amqp_publish_SOURCES = publish.c common.c -amqp_get_SOURCES = get.c common.c common_consume.c -amqp_consume_SOURCES = consume.c common.c common_consume.c +amqp_get_SOURCES = get.c common.c +amqp_consume_SOURCES = consume.c common.c diff --git a/tools/common.c b/tools/common.c index 0772738..37fd301 100644 --- a/tools/common.c +++ b/tools/common.c @@ -377,7 +377,7 @@ poptContext process_options(int argc, const char **argv, poptSetOtherOptionHelp(opts, help); while ((c = poptGetNextOpt(opts)) >= 0) { - // no options require explicit handling + /* no options require explicit handling */ } if (c < -1) { diff --git a/tools/common_consume.c b/tools/common_consume.c deleted file mode 100644 index 318fa2c..0000000 --- a/tools/common_consume.c +++ /dev/null @@ -1,160 +0,0 @@ -/* - * ***** BEGIN LICENSE BLOCK ***** - * Version: MPL 1.1/GPL 2.0 - * - * The contents of this file are subject to the Mozilla Public License - * Version 1.1 (the "License"); you may not use this file except in - * compliance with the License. You may obtain a copy of the License at - * http://www.mozilla.org/MPL/ - * - * Software distributed under the License is distributed on an "AS IS" - * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See - * the License for the specific language governing rights and - * limitations under the License. - * - * The Original Code is librabbitmq. - * - * The Initial Developers of the Original Code are LShift Ltd, Cohesive - * Financial Technologies LLC, and Rabbit Technologies Ltd. Portions - * created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, Cohesive - * Financial Technologies LLC, or Rabbit Technologies Ltd are Copyright - * (C) 2007-2008 LShift Ltd, Cohesive Financial Technologies LLC, and - * Rabbit Technologies Ltd. - * - * Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift - * Ltd. Portions created by Cohesive Financial Technologies LLC are - * Copyright (C) 2007-2009 Cohesive Financial Technologies - * LLC. Portions created by Rabbit Technologies Ltd are Copyright (C) - * 2007-2009 Rabbit Technologies Ltd. - * - * Portions created by Tony Garnock-Jones are Copyright (C) 2009-2010 - * LShift Ltd and Tony Garnock-Jones. - * - * All Rights Reserved. - * - * Contributor(s): ______________________________________. - * - * Alternatively, the contents of this file may be used under the terms - * of the GNU General Public License Version 2 or later (the "GPL"), in - * which case the provisions of the GPL are applicable instead of those - * above. If you wish to allow use of your version of this file only - * under the terms of the GPL, and not to allow others to use your - * version of this file under the terms of the MPL, indicate your - * decision by deleting the provisions above and replace them with the - * notice and other provisions required by the GPL. If you do not - * delete the provisions above, a recipient may use your version of - * this file under the terms of any one of the MPL or the GPL. - * - * ***** END LICENSE BLOCK ***** - */ - -#include "config.h" - -#include -#include - -#include - -#include "common.h" - -static char *queue; -static char *exchange; -static char *exchange_type; -static char *routing_key; - -const char *consume_queue_options_title = "Source queue options"; -struct poptOption consume_queue_options[] = { - {"queue", 'q', POPT_ARG_STRING, &queue, 0, - "the queue to consume from", "queue"}, - {"exchange", 'e', POPT_ARG_STRING, &exchange, 0, - "bind the queue to this exchange", "exchange"}, - {"exchange-type", 't', POPT_ARG_STRING, &exchange_type, 0, - "create auto-delete exchange of this type for binding", "type"}, - {"routing-key", 'r', POPT_ARG_STRING, &routing_key, 0, - "the routing key to bind with", "routing key"}, - { NULL, 0, 0, NULL, 0 } -}; - -/* Convert a amqp_bytes_t to an escaped string form for printing. We - use the same escaping conventions as rabbitmqctl. */ -static char *stringify_bytes(amqp_bytes_t bytes) -{ - /* W will need up to 4 chars per byte, plus the terminating 0 */ - char *res = malloc(bytes.len * 4 + 1); - uint8_t *data = bytes.bytes; - char *p = res; - size_t i; - - for (i = 0; i < bytes.len; i++) { - if (data[i] >= 32 && data[i] != 127) { - *p++ = data[i]; - } - else { - *p++ = '\\'; - *p++ = '0' + (data[i] >> 6); - *p++ = '0' + (data[i] >> 3 & 0x7); - *p++ = '0' + (data[i] & 0x7); - } - } - - *p = 0; - return res; -} - -amqp_bytes_t setup_queue(amqp_connection_state_t conn) -{ - /* if an exchange name wasn't provided, check that we don't - have options that require it. */ - if (!exchange) { - char *opt = NULL; - if (routing_key) - opt = "--routing-key"; - else if (exchange_type) - opt = "--exchange-type"; - - if (opt) { - fprintf(stderr, - "%s option requires an exchange name to be " - "provided with --exchange\n", opt); - exit(1); - } - } - - /* Declare the queue. If the queue already exists, this won't have - any effect. */ - amqp_bytes_t queue_bytes = cstring_bytes(queue); - amqp_queue_declare_ok_t *res - = amqp_queue_declare(conn, 1, queue_bytes, 0, 0, 0, 1, - AMQP_EMPTY_TABLE); - if (!res) - die_rpc(amqp_get_rpc_reply(conn), "queue.declare"); - - if (!queue) { - // the server should have provided a queue name - char *sq; - queue_bytes = amqp_bytes_malloc_dup(res->queue); - sq = stringify_bytes(queue_bytes); - fprintf(stderr, "Server provided queue name: %s\n", sq); - free(sq); - } - - /* Bind to an exchange if requested */ - if (exchange) { - amqp_bytes_t eb = amqp_cstring_bytes(exchange); - - if (exchange_type) { - // we should create the exchange - if (!amqp_exchange_declare(conn, 1, eb, - amqp_cstring_bytes(exchange_type), - 0, 0, 1, AMQP_EMPTY_TABLE)) - die_rpc(amqp_get_rpc_reply(conn), "exchange.declare"); - } - - if (!amqp_queue_bind(conn, 1, queue_bytes, eb, - cstring_bytes(routing_key), - AMQP_EMPTY_TABLE)) - die_rpc(amqp_get_rpc_reply(conn), "queue.bind"); - } - - return queue_bytes; -} diff --git a/tools/common_consume.h b/tools/common_consume.h deleted file mode 100644 index 703f4bd..0000000 --- a/tools/common_consume.h +++ /dev/null @@ -1,54 +0,0 @@ -/* - * ***** BEGIN LICENSE BLOCK ***** - * Version: MPL 1.1/GPL 2.0 - * - * The contents of this file are subject to the Mozilla Public License - * Version 1.1 (the "License"); you may not use this file except in - * compliance with the License. You may obtain a copy of the License at - * http://www.mozilla.org/MPL/ - * - * Software distributed under the License is distributed on an "AS IS" - * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See - * the License for the specific language governing rights and - * limitations under the License. - * - * The Original Code is librabbitmq. - * - * The Initial Developers of the Original Code are LShift Ltd, Cohesive - * Financial Technologies LLC, and Rabbit Technologies Ltd. Portions - * created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, Cohesive - * Financial Technologies LLC, or Rabbit Technologies Ltd are Copyright - * (C) 2007-2008 LShift Ltd, Cohesive Financial Technologies LLC, and - * Rabbit Technologies Ltd. - * - * Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift - * Ltd. Portions created by Cohesive Financial Technologies LLC are - * Copyright (C) 2007-2009 Cohesive Financial Technologies - * LLC. Portions created by Rabbit Technologies Ltd are Copyright (C) - * 2007-2009 Rabbit Technologies Ltd. - * - * Portions created by Tony Garnock-Jones are Copyright (C) 2009-2010 - * LShift Ltd and Tony Garnock-Jones. - * - * All Rights Reserved. - * - * Contributor(s): ______________________________________. - * - * Alternatively, the contents of this file may be used under the terms - * of the GNU General Public License Version 2 or later (the "GPL"), in - * which case the provisions of the GPL are applicable instead of those - * above. If you wish to allow use of your version of this file only - * under the terms of the GPL, and not to allow others to use your - * version of this file under the terms of the MPL, indicate your - * decision by deleting the provisions above and replace them with the - * notice and other provisions required by the GPL. If you do not - * delete the provisions above, a recipient may use your version of - * this file under the terms of any one of the MPL or the GPL. - * - * ***** END LICENSE BLOCK ***** - */ - -extern const char *consume_queue_options_title; -extern struct poptOption consume_queue_options[]; -extern amqp_bytes_t setup_queue(amqp_connection_state_t conn); - diff --git a/tools/consume.c b/tools/consume.c index 7ede926..40b61d1 100644 --- a/tools/consume.c +++ b/tools/consume.c @@ -51,17 +51,105 @@ #include "config.h" #include +#include #include #include "common.h" -#include "common_consume.h" -static void do_consume(amqp_connection_state_t conn, int no_ack, - const char * const *argv) +/* Convert a amqp_bytes_t to an escaped string form for printing. We + use the same escaping conventions as rabbitmqctl. */ +static char *stringify_bytes(amqp_bytes_t bytes) { - if (!amqp_basic_consume(conn, 1, setup_queue(conn), - AMQP_EMPTY_BYTES, 0, no_ack, 0)) + /* We will need up to 4 chars per byte, plus the terminating 0 */ + char *res = malloc(bytes.len * 4 + 1); + uint8_t *data = bytes.bytes; + char *p = res; + size_t i; + + for (i = 0; i < bytes.len; i++) { + if (data[i] >= 32 && data[i] != 127) { + *p++ = data[i]; + } + else { + *p++ = '\\'; + *p++ = '0' + (data[i] >> 6); + *p++ = '0' + (data[i] >> 3 & 0x7); + *p++ = '0' + (data[i] & 0x7); + } + } + + *p = 0; + return res; +} + +static amqp_bytes_t setup_queue(amqp_connection_state_t conn, + char *queue, char *exchange, + char *exchange_type, char *routing_key) +{ + amqp_bytes_t queue_bytes; + amqp_queue_declare_ok_t *res; + + /* if an exchange name wasn't provided, check that we don't + have options that require it. */ + if (!exchange) { + char *opt = NULL; + if (routing_key) + opt = "--routing-key"; + else if (exchange_type) + opt = "--exchange-type"; + + if (opt) { + fprintf(stderr, + "%s option requires an exchange name to be " + "provided with --exchange\n", opt); + exit(1); + } + } + + /* Declare the queue as auto-delete. If the queue already + exists, this won't have any effect. */ + queue_bytes = cstring_bytes(queue); + res = amqp_queue_declare(conn, 1, queue_bytes, 0, 0, 0, 1, + AMQP_EMPTY_TABLE); + if (!res) + die_rpc(amqp_get_rpc_reply(conn), "queue.declare"); + + if (!queue) { + /* the server should have provided a queue name */ + char *sq; + queue_bytes = amqp_bytes_malloc_dup(res->queue); + sq = stringify_bytes(queue_bytes); + fprintf(stderr, "Server provided queue name: %s\n", sq); + free(sq); + } + + /* Bind to an exchange if requested */ + if (exchange) { + amqp_bytes_t eb = amqp_cstring_bytes(exchange); + + if (exchange_type) { + /* we should create the exchange */ + if (!amqp_exchange_declare(conn, 1, eb, + amqp_cstring_bytes(exchange_type), + 0, 0, 1, AMQP_EMPTY_TABLE)) + die_rpc(amqp_get_rpc_reply(conn), "exchange.declare"); + } + + if (!amqp_queue_bind(conn, 1, queue_bytes, eb, + cstring_bytes(routing_key), + AMQP_EMPTY_TABLE)) + die_rpc(amqp_get_rpc_reply(conn), "queue.bind"); + } + + return queue_bytes; +} + +static void do_consume(amqp_connection_state_t conn, amqp_bytes_t queue, + int no_ack, const char * const *argv) +{ + if (!amqp_basic_consume(conn, 1, queue, AMQP_EMPTY_BYTES, 0, no_ack, + 0)) die_rpc(amqp_get_rpc_reply(conn), "basic.consume"); for (;;) { @@ -97,10 +185,23 @@ int main(int argc, const char **argv) int no_ack; amqp_connection_state_t conn; const char * const *cmd_argv; - + char *queue = NULL; + char *exchange = NULL; + char *exchange_type = NULL; + char *routing_key = NULL; + amqp_bytes_t queue_bytes; + struct poptOption options[] = { INCLUDE_OPTIONS(connect_options), - INCLUDE_OPTIONS(consume_queue_options), + {"queue", 'q', POPT_ARG_STRING, &queue, 0, + "the queue to consume from", "queue"}, + {"exchange", 'e', POPT_ARG_STRING, &exchange, 0, + "bind the queue to this exchange", "exchange"}, + {"exchange-type", 't', POPT_ARG_STRING, &exchange_type, 0, + "create auto-delete exchange of this type for binding", + "type"}, + {"routing-key", 'r', POPT_ARG_STRING, &routing_key, 0, + "the routing key to bind with", "routing key"}, {"no-ack", 'A', POPT_ARG_NONE, &no_ack, 0, "consume in no-ack mode", NULL}, POPT_AUTOHELP @@ -118,7 +219,9 @@ int main(int argc, const char **argv) } conn = make_connection(); - do_consume(conn, no_ack, cmd_argv); + queue_bytes = setup_queue(conn, queue, exchange, exchange_type, + routing_key); + do_consume(conn, queue_bytes, no_ack, cmd_argv); close_connection(conn); return 0; diff --git a/tools/get.c b/tools/get.c index b433e2b..f746fd1 100644 --- a/tools/get.c +++ b/tools/get.c @@ -55,12 +55,11 @@ #include #include "common.h" -#include "common_consume.h" -static int do_get(amqp_connection_state_t conn) +static int do_get(amqp_connection_state_t conn, char *queue) { amqp_rpc_reply_t r - = amqp_basic_get(conn, 1, setup_queue(conn), 1); + = amqp_basic_get(conn, 1, cstring_bytes(queue), 1); die_rpc(r, "basic.get"); if (r.reply.id == AMQP_BASIC_GET_EMPTY_METHOD) @@ -73,19 +72,26 @@ static int do_get(amqp_connection_state_t conn) int main(int argc, const char **argv) { amqp_connection_state_t conn; + char *queue = NULL; int got_something; struct poptOption options[] = { INCLUDE_OPTIONS(connect_options), - INCLUDE_OPTIONS(consume_queue_options), + {"queue", 'q', POPT_ARG_STRING, &queue, 0, + "the queue to consume from", "queue"}, POPT_AUTOHELP { NULL, 0, 0, NULL, 0 } }; process_all_options(argc, argv, options); + + if (!queue) { + fprintf(stderr, "queue not specified\n"); + return 1; + } conn = make_connection(); - got_something = do_get(conn); + got_something = do_get(conn, queue); close_connection(conn); return got_something ? 0 : 2; } diff --git a/tools/publish.c b/tools/publish.c index a5b86b2..21314b2 100644 --- a/tools/publish.c +++ b/tools/publish.c @@ -111,7 +111,7 @@ int main(int argc, const char **argv) memset(&props, 0, sizeof props); props._flags = AMQP_BASIC_DELIVERY_MODE_FLAG; - props.delivery_mode = 2; // persistent delivery mode + props.delivery_mode = 2; /* persistent delivery mode */ if (content_type) { props._flags |= AMQP_BASIC_CONTENT_TYPE_FLAG; -- cgit v1.2.1