/* * * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. * */ #include #include #include #include #include #include #include #include #include using namespace qpid::messaging; using namespace qpid::types; struct Options : public qpid::Options { bool help; std::string url; std::string address; std::string connectionOptions; int64_t timeout; bool forever; qpid::log::Options log; Options(const std::string& argv0=std::string()) : qpid::Options("Options"), help(false), url("amqp:tcp:127.0.0.1"), timeout(0), forever(false), log(argv0) { addOptions() ("broker,b", qpid::optValue(url, "URL"), "url of broker to connect to") ("address,a", qpid::optValue(address, "ADDRESS"), "address to drain from") ("timeout,t", qpid::optValue(timeout, "TIMEOUT"), "timeout in seconds to wait before exiting") ("forever,f", qpid::optValue(forever), "ignore timeout and wait forever") ("connection-options", qpid::optValue(connectionOptions,"OPTIONS"), "connection options string in the form {name1=value1, name2=value2}") ("help", qpid::optValue(help), "print this usage statement"); add(log); } Duration getTimeout() { if (forever) return Duration::FOREVER; else return timeout*Duration::SECOND; } bool parse(int argc, char** argv) { try { qpid::Options::parse(argc, argv); if (address.empty()) throw qpid::Exception("Address must be specified!"); qpid::log::Logger::instance().configure(log); if (help) { std::ostringstream msg; std::cout << msg << *this << std::endl << std::endl << "Drains messages from the specified address" << std::endl; return false; } else { return true; } } catch (const std::exception& e) { std::cerr << *this << std::endl << std::endl << e.what() << std::endl; return false; } } }; int main(int argc, char** argv) { Options options(argv[0]); if (options.parse(argc, argv)) { Connection connection(options.connectionOptions); try { connection.open(options.url); Session session = connection.createSession(); Receiver receiver = session.createReceiver(options.address); Duration timeout = options.getTimeout(); Message message; while (receiver.fetch(message, timeout)) { std::cout << "Message(properties=" << message.getProperties() << ", content='" ; if (message.getContentType() == "amqp/map") { Variant::Map map; decode(message, map); std::cout << map; } else { std::cout << message.getContent(); } std::cout << "')" << std::endl; session.acknowledge(); } receiver.close(); session.close(); connection.close(); return 0; } catch(const std::exception& error) { std::cout << error.what() << std::endl; connection.close(); } } return 1; }