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.

7 files changed, 127 insertions, 232 deletions

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 <stdio.h>
-#include <stdlib.h>
-
-#include <popt.h>
-
-#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 <stdio.h>
+#include <stdlib.h>
 
 #include <popt.h>
 
 #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 <popt.h>
 
 #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;