diff options
Diffstat (limited to 'tools/consume.c')
-rw-r--r-- | tools/consume.c | 128 |
1 files changed, 128 insertions, 0 deletions
diff --git a/tools/consume.c b/tools/consume.c new file mode 100644 index 0000000..7ede926 --- /dev/null +++ b/tools/consume.c @@ -0,0 +1,128 @@ +/* + * ***** 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 <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) +{ + if (!amqp_basic_consume(conn, 1, setup_queue(conn), + AMQP_EMPTY_BYTES, 0, no_ack, 0)) + die_rpc(amqp_get_rpc_reply(conn), "basic.consume"); + + for (;;) { + amqp_frame_t frame; + struct pipeline pl; + uint64_t delivery_tag; + int res = amqp_simple_wait_frame(conn, &frame); + if (res < 0) + die_errno(-res, "waiting for header frame"); + + if (frame.frame_type != AMQP_FRAME_METHOD + || frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD) + continue; + + amqp_basic_deliver_t *deliver + = (amqp_basic_deliver_t *)frame.payload.method.decoded; + delivery_tag = deliver->delivery_tag; + + pipeline(argv, &pl); + copy_body(conn, pl.infd); + + if (finish_pipeline(&pl) && !no_ack) + die_errno(-amqp_basic_ack(conn, 1, delivery_tag, 0), + "basic.ack"); + + amqp_maybe_release_buffers(conn); + } +} + +int main(int argc, const char **argv) +{ + poptContext opts; + int no_ack; + amqp_connection_state_t conn; + const char * const *cmd_argv; + + struct poptOption options[] = { + INCLUDE_OPTIONS(connect_options), + INCLUDE_OPTIONS(consume_queue_options), + {"no-ack", 'A', POPT_ARG_NONE, &no_ack, 0, + "consume in no-ack mode", NULL}, + POPT_AUTOHELP + { NULL, 0, 0, NULL, 0 } + }; + + opts = process_options(argc, argv, options, + "[OPTIONS]... <command> <args>"); + + cmd_argv = poptGetArgs(opts); + if (!cmd_argv || !cmd_argv[0]) { + fprintf(stderr, "consuming command not specified\n"); + poptPrintUsage(opts, stderr, 0); + goto error; + } + + conn = make_connection(); + do_consume(conn, no_ack, cmd_argv); + close_connection(conn); + return 0; + +error: + poptFreeContext(opts); + return 1; +} |