diff options
Diffstat (limited to 'cpp/examples/old_api/xml-exchange')
| -rw-r--r-- | cpp/examples/old_api/xml-exchange/CMakeLists.txt | 24 | ||||
| -rw-r--r-- | cpp/examples/old_api/xml-exchange/Makefile.am | 49 | ||||
| -rw-r--r-- | cpp/examples/old_api/xml-exchange/README.txt | 53 | ||||
| -rw-r--r-- | cpp/examples/old_api/xml-exchange/declare_queues.cpp | 98 | ||||
| -rw-r--r-- | cpp/examples/old_api/xml-exchange/listener.cpp | 107 | ||||
| -rw-r--r-- | cpp/examples/old_api/xml-exchange/xml_producer.cpp | 113 |
6 files changed, 444 insertions, 0 deletions
diff --git a/cpp/examples/old_api/xml-exchange/CMakeLists.txt b/cpp/examples/old_api/xml-exchange/CMakeLists.txt new file mode 100644 index 0000000000..3fea47a208 --- /dev/null +++ b/cpp/examples/old_api/xml-exchange/CMakeLists.txt @@ -0,0 +1,24 @@ +# +# 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. +# + +if (BUILD_XML) +add_example(xml-exchange declare_queues) +add_example(xml-exchange xml_producer) +add_example(xml-exchange listener) +endif (BUILD_XML) diff --git a/cpp/examples/old_api/xml-exchange/Makefile.am b/cpp/examples/old_api/xml-exchange/Makefile.am new file mode 100644 index 0000000000..3e1082cdb2 --- /dev/null +++ b/cpp/examples/old_api/xml-exchange/Makefile.am @@ -0,0 +1,49 @@ +# +# 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. +# +examplesdir=$(pkgdatadir)/examples/old_api/xml-exchange + +MAKELDFLAGS=$(CLIENTFLAGS) +include $(top_srcdir)/examples/makedist.mk + +noinst_PROGRAMS=declare_queues xml_producer listener + +declare_queues_SOURCES=declare_queues.cpp +declare_queues_LDADD=$(CLIENT_LIB) + +xml_producer_SOURCES=xml_producer.cpp +xml_producer_LDADD=$(CLIENT_LIB) + +listener_SOURCES=listener.cpp +listener_LDADD=$(CLIENT_LIB) + +EXTRA_DIST= \ + README.txt \ + CMakeLists.txt + +examples_DATA= \ + $(EXTRA_DIST) \ + declare_queues.cpp \ + listener.cpp \ + xml_producer.cpp \ + $(MAKEDIST) + + + + + diff --git a/cpp/examples/old_api/xml-exchange/README.txt b/cpp/examples/old_api/xml-exchange/README.txt new file mode 100644 index 0000000000..85caebe352 --- /dev/null +++ b/cpp/examples/old_api/xml-exchange/README.txt @@ -0,0 +1,53 @@ +This example shows how to program a simple application +using the XML Exchange. + +[Note: The XML Exchange is not a standard AMQP exchange type. To run +this example you need to have a broker that has support for the xml +exchange. If you are compiling the broker from source please refer to +the INSTALL notes from qpid.] + +To run the example, execute the programs in the +following order: + +1 ./declare_queues +2 ./listener +3 ./message_producer (in a separate window) + +The XML Exchange must be explicitly declared. Bindings +are established using queries in XQuery. These queries +can reference message content, message application +properties (which are declared as external variables +in the XQuery), or both. + +Once this is done, message producers publish to the +exchange using the exchange name and a routing key, +just as for other exchange types. Message consumers +read from the queues to which messages are routed. +If a message does not have XML content, or is +missing message application properties needed by +the query, the query is not routed. + +Queries can use message application headers to +provide functionality similar to JMS selectors. +If a query does not use the content of a message, +the message content is not parsed, and need not +be XML. + +The XQuery processor, XQilla, does path-based +document projection, so once the portion of +a document needed to evaluate a query has +been read, it stops parsing the document. +Suppose a long document has a header section. +You can indicate in the query that only +one header section needs to be queried, +and there is no need to parse the entire +document to see if there are further header +sections, using a path like this: + +./message/header[1]/date + +If you used a path like this, all children +of the message element would be read to +see if there are further headers: + +./message/header/date diff --git a/cpp/examples/old_api/xml-exchange/declare_queues.cpp b/cpp/examples/old_api/xml-exchange/declare_queues.cpp new file mode 100644 index 0000000000..ad08642019 --- /dev/null +++ b/cpp/examples/old_api/xml-exchange/declare_queues.cpp @@ -0,0 +1,98 @@ +/* + * + * 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. + * + */ + + +/** + * + * declare_queues.cpp + * + * This is one of three programs used to implement XML-based content + * routing in C++. + * + * declare_queues.cpp (this program) + * + * Creates a queue named "message_qaueue" on the broker, + * declares an XML Exchange, subscribes the queue to the XML + * Exchange using an XQuery in the binding, then exits. + * + * xml_producer.cpp + * + * Publishes messages to the XML Exchange. + * + * listener.cpp + * + * Reads messages from the "message_queue" queue. + */ + +#include <qpid/client/Connection.h> +#include <qpid/client/Session.h> + +#include <cstdlib> +#include <iostream> + +using namespace qpid::client; +using namespace qpid::framing; + +using std::string; + + +int main(int argc, char** argv) { + const char* host = argc>1 ? argv[1] : "127.0.0.1"; + int port = argc>2 ? atoi(argv[2]) : 5672; + Connection connection; + + try { + connection.open(host, port); + Session session = connection.newSession(); + + + //--------- Main body of program -------------------------------------------- + + // Set up queues, bind them with queries. Note that the XML exchange + // is not in the AMQP specification, so it is called "xml", not "amq.xml". + // Note that the XML exchange is not predeclared in Qpid, it must + // be declared by the application. + + session.queueDeclare(arg::queue="message_queue"); + session.exchangeDeclare(arg::exchange="xml", arg::type="xml"); + + // Application message properties are mapped to external variables + // in the XQuery. An XML Exchange can query message properties much + // like JMS, query the XML content of the message, or both. + + FieldTable binding; + binding.setString("xquery", "declare variable $control external;" + "./message/id mod 2 = 1 or $control = 'end'"); + session.exchangeBind(arg::exchange="xml", arg::queue="message_queue", arg::bindingKey="content_feed", arg::arguments=binding); + + //----------------------------------------------------------------------------- + + connection.close(); + return 0; + } catch(const std::exception& error) { + std::cout << error.what() << std::endl; + } + return 1; + +} + + + diff --git a/cpp/examples/old_api/xml-exchange/listener.cpp b/cpp/examples/old_api/xml-exchange/listener.cpp new file mode 100644 index 0000000000..11bcb9f669 --- /dev/null +++ b/cpp/examples/old_api/xml-exchange/listener.cpp @@ -0,0 +1,107 @@ +/* + * + * 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. + * + */ + + +/** + * + * listener.cpp + * + * This is one of three programs used to implement XML-based content + * routing in C++. + * + * declare_queues.cpp + * + * Creates a queue named "message_qaueue" on the broker, + * declares an XML Exchange, subscribes the queue to the XML + * Exchange using an XQuery in the binding, then exits. + * + * xml_producer.cpp + * + * Publishes messages to the XML Exchange. + * + * listener.cpp (this program) + * + * Reads messages from the "message_queue" queue. + */ + + +#include <qpid/client/Connection.h> +#include <qpid/client/Session.h> +#include <qpid/client/Message.h> +#include <qpid/client/SubscriptionManager.h> + +#include <cstdlib> +#include <iostream> + +using namespace qpid::client; +using namespace qpid::framing; + + +class Listener : public MessageListener{ + private: + SubscriptionManager& subscriptions; + public: + Listener(SubscriptionManager& subscriptions); + virtual void received(Message& message); +}; + +Listener::Listener(SubscriptionManager& subs) : subscriptions(subs) +{} + +void Listener::received(Message& message) { + std::cout << "Message: " << message.getData() << std::endl; + if (message.getHeaders().getAsString("control") == "end") { + std::cout << "Shutting down listener for " << message.getDestination() + << std::endl; + subscriptions.cancel(message.getDestination()); + } +} + +int main(int argc, char** argv) { + const char* host = argc>1 ? argv[1] : "127.0.0.1"; + int port = argc>2 ? atoi(argv[2]) : 5672; + + Connection connection; + try { + connection.open(host, port); + Session session = connection.newSession(); + + //--------- Main body of program -------------------------------------------- + + SubscriptionManager subscriptions(session); + // Create a listener and subscribe it to the queue named "message_queue" + Listener listener(subscriptions); + subscriptions.subscribe(listener, "message_queue"); + // Receive messages until the subscription is cancelled + // by Listener::received() + subscriptions.run(); + + //--------------------------------------------------------------------------- + + connection.close(); + return 0; + } catch(const std::exception& error) { + std::cout << error.what() << std::endl; + } + return 1; +} + + diff --git a/cpp/examples/old_api/xml-exchange/xml_producer.cpp b/cpp/examples/old_api/xml-exchange/xml_producer.cpp new file mode 100644 index 0000000000..af1a7e60c7 --- /dev/null +++ b/cpp/examples/old_api/xml-exchange/xml_producer.cpp @@ -0,0 +1,113 @@ +/* + * + * 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. + * + */ + + +/** + * + * xml_producer.cpp + * + * This is one of three programs used to implement XML-based content + * routing in C++. + * + * declare_queues.cpp + * + * Creates a queue named "message_qaueue" on the broker, + * declares an XML Exchange, subscribes the queue to the XML + * Exchange using an XQuery in the binding, then exits. + * + * xml_producer.cpp (this program) + * + * Publishes messages to the XML Exchange. + * + * listener.cpp + * + * Reads messages from the "message_queue" queue. + */ + + +#include <qpid/client/Connection.h> +#include <qpid/client/Session.h> +#include <qpid/client/AsyncSession.h> +#include <qpid/client/Message.h> + + +#include <cstdlib> +#include <iostream> + +#include <sstream> + +using namespace qpid::client; +using namespace qpid::framing; + +using std::stringstream; +using std::string; + +int main(int argc, char** argv) { + const char* host = argc>1 ? argv[1] : "127.0.0.1"; + int port = argc>2 ? atoi(argv[2]) : 5672; + Connection connection; + Message message; + try { + connection.open(host, port); + Session session = connection.newSession(); + + //--------- Main body of program -------------------------------------------- + + // Publish some XML messages. Use the control property to + // indicate when we are finished. + // + // In the XML exchange, the routing key and the name of + // the query match. + + message.getDeliveryProperties().setRoutingKey("content_feed"); + message.getHeaders().setString("control","continue"); + + // Now send some messages ... + + for (int i=0; i<10; i++) { + stringstream message_data; + message_data << "<message><id>" << i << "</id></message>"; + + std::cout << "Message data: " << message_data.str() << std::endl; + + message.setData(message_data.str()); + // Asynchronous transfer sends messages as quickly as + // possible without waiting for confirmation. + async(session).messageTransfer(arg::content=message, arg::destination="xml"); + } + + // And send a final message to indicate termination. + + message.getHeaders().setString("control","end"); + message.setData("<end>That's all, folks!</end>"); + session.messageTransfer(arg::content=message, arg::destination="xml"); + + //----------------------------------------------------------------------------- + + connection.close(); + return 0; + } catch(const std::exception& error) { + std::cout << error.what() << std::endl; + } + return 1; +} + + |