You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by jo...@apache.org on 2010/04/09 22:59:39 UTC
svn commit: r932610 - /qpid/trunk/qpid/doc/book/src/High-Level-API.xml
Author: jonathan
Date: Fri Apr 9 20:59:38 2010
New Revision: 932610
URL: http://svn.apache.org/viewvc?rev=932610&view=rev
Log:
Now relatively complete for C++ and Python, but properties were changing today as I was writing this, and this will need to be revised Monday.
The next big step: adding information for Java clients.
Modified:
qpid/trunk/qpid/doc/book/src/High-Level-API.xml
Modified: qpid/trunk/qpid/doc/book/src/High-Level-API.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/doc/book/src/High-Level-API.xml?rev=932610&r1=932609&r2=932610&view=diff
==============================================================================
--- qpid/trunk/qpid/doc/book/src/High-Level-API.xml (original)
+++ qpid/trunk/qpid/doc/book/src/High-Level-API.xml Fri Apr 9 20:59:38 2010
@@ -1,4 +1,30 @@
-<chapter>
+<?xml version='1.0' encoding='utf-8' ?>
+
+<!--
+
+ 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.
+
+-->
+
+<!DOCTYPE bookinfo PUBLIC "-//OASIS//DTD DocBook XML V4.5//EN" "http://www.oasis-open.org/docbook/xml/4.5/docbookx.dtd" [
+]>
+
+<chapter id="high-level-client-api">
<title>Qpid High Level Client API</title>
<para>The Apache Qpid High Level Client API is a reliable,
@@ -210,7 +236,7 @@ options ::= { <key> : <value>, ... }
</listitem>
<listitem><para>When a Sender sends a message to a queue, the
message is sent to the AQMP 0-10 default queue, using the name
- of teh queue as the routing key.</para></listitem>
+ of the queue as the routing key.</para></listitem>
<listitem><para>When a Receiver receives messages from a queue,
it is treated as a normal AMQP 0-10 queue subscription. The
<varname>accept-mode</varname> property can be set using
@@ -250,13 +276,13 @@ options ::= { <key> : <value>, ... }
<para>Create a queue with qpid-config, send a message using
spout, and read it using drain:</para>
- <programlisting>
+ <screen>
$ qpid-config add queue hello-world
$ ./spout -a hello-world
$ ./drain -a hello-world
Message(properties={spout-id:c877e622-d57b-4df2-bf3e-6014c68da0ea:0}, content='')
- </programlisting>
+ </screen>
<para>Exchanges and queues are addressed in exactly the same way
in the Qpid High Level Client API. If we delete the queue named
@@ -264,13 +290,13 @@ Message(properties={spout-id:c877e622-d5
same name, we can write to and read from the exchange in the
same way as for the queue:</para>
- <programlisting>
+ <screen>
$ qpid-config del queue hello-world
$ qpid-config add exchange topic hello-world
$ ./spout -a hello-world
$ ./drain -a hello-world
$
- </programlisting>
+ </screen>
<para>However, in AMQP 0-10, exchanges discard messages if no
queue is bound to the exchange, unlike queues, which store
@@ -283,25 +309,28 @@ $
run <command>spout</command> in another window to send a message
for <command>drain</command> to receive.</para>
- <para>First window:</para>
+
+
+ <para><emphasis>First Window:</emphasis></para>
- <programlisting>
+ <screen>
$ ./drain -a hello-word -t 30
- </programlisting>
+ </screen>
+
- <para>Second window:</para>
+ <para><emphasis>Second Window:</emphasis></para>
- <programlisting>
+ <screen>
$ ./spout -a hello-word
- </programlisting>
+ </screen>
<para>Once <command>spout</command> has sent a message, return
to the first window to see the output from
<command>drain</command>:</para>
- <programlisting>
+ <screen>
Message(properties={spout-id:7da2d27d-93e6-4803-8a61-536d87b8d93f:0}, content='')
- </programlisting>
+ </screen>
<para>You can run <command>drain</command> in several separate
windows; each will create a subscription for the exchange, and
@@ -350,29 +379,29 @@ Message(properties={spout-id:7da2d27d-93
<para>Let's create a direct exchange and listen for messages
whose subject is <literal>sports</literal>:</para>
- <para>First window:</para>
- <programlisting>
+ <para><emphasis>First Window:</emphasis></para>
+ <screen>
$ qpid-config add exchange direct direct-exchange
$ ./drain -a direct-exchange/sports -t 30
- </programlisting>
+ </screen>
<para>In a second window, let's send messages to the
exchange we created:</para>
- <para>Second window:</para>
- <programlisting>
+ <para><emphasis>Second Window:</emphasis></para>
+ <screen>
$ ./spout -a direct-exchange/sports
$ ./spout -a direct-exchange/news
- </programlisting>
+ </screen>
<para>Now look at the first window, and you will see the
message with the subject <literal>sports</literal> has been
received, but not the message with the subject
<literal>news</literal>:</para>
- <programlisting>
+ <screen>
Message(properties={qpid.subject:sports, spout-id:9441674e-a157-4780-a78e-f7ccea998291:0}, content='')
- </programlisting>
+ </screen>
<para>If you run <command>drain</command> in multiple
windows using the same subject, all instances of
@@ -403,21 +432,21 @@ Message(properties={qpid.subject:sports,
<para>Let's create a topic exchange and listen for messages
whose subject is <literal>news</literal>:</para>
- <para>First window:</para>
+ <para><emphasis>First Window:</emphasis></para>
- <programlisting>
+ <screen>
$ qpid-config add exchange topic topic-exchange
$ ./drain -a topic-exchange/news -t 30
- </programlisting>
+ </screen>
<para>In a second window, let's send messages to the
exchange we created:</para>
- <para>Second window:</para>
- <programlisting>
+ <para><emphasis>Second Window:</emphasis></para>
+ <screen>
$ ./spout -a topic-exchange/news
$ ./spout -a topic-exchange/sports
- </programlisting>
+ </screen>
<para>Now look at the first window, and you will see the
@@ -425,9 +454,9 @@ $ ./spout -a topic-exchange/sports
received, but not the message with the subject
<literal>sports</literal>:</para>
- <programlisting>
+ <screen>
Message(properties={qpid.subject:news, spout-id:bafefb74-c5be-4a8b-9e4b-45f7a855e250:0}, content='')
- </programlisting>
+ </screen>
<para>Now let's use the topic exchange with wildcards in the
@@ -438,33 +467,33 @@ Message(properties={qpid.subject:news, s
<literal>news</literal>:</para>
- <para>First window:</para>
+ <para><emphasis>First Window:</emphasis></para>
- <programlisting>
+ <screen>
$ ./drain -a topic-exchange/*.news -t 30
- </programlisting>
+ </screen>
<para>Now let's send messages using several different
two-word keys:</para>
- <para>Second window:</para>
+ <para><emphasis>Second Window:</emphasis></para>
- <programlisting>
+ <screen>
$ ./spout -a topic-exchange/usa.news
$ ./spout -a topic-exchange/usa.sports
$ ./spout -a topic-exchange/europe.sports
$ ./spout -a topic-exchange/europe.news
$
- </programlisting>
+ </screen>
<para>Now look at the first window, and you will see the
messages with <literal>news</literal> in the second word of
the key have been received:</para>
- <programlisting>
+ <screen>
Message(properties={qpid.subject:usa.news, spout-id:73fc8058-5af6-407c-9166-b49a9076097a:0}, content='')
Message(properties={qpid.subject:europe.news, spout-id:f72815aa-7be4-4944-99fd-c64c9747a876:0}, content='')
- </programlisting>
+ </screen>
<para>Finally, let's use the <literal>#</literal> wildcard
in the Receiver to match any number of words in the key. The
@@ -473,35 +502,35 @@ Message(properties={qpid.subject:europe.
<literal>news</literal>, no matter how many words are in the
key:</para>
- <para>First window:</para>
+ <para><emphasis>First Window:</emphasis></para>
- <programlisting>
+ <screen>
$ ./drain -a topic-exchange/#.news -t 30
- </programlisting>
+ </screen>
<para>Now let's send messages using a variety of different
multi-word keys:</para>
- <para>Second window:</para>
+ <para><emphasis>Second Window:</emphasis></para>
- <programlisting>
+ <screen>
$ ./spout -a topic-exchange/news
$ ./spout -a topic-exchange/sports
$ ./spout -a topic-exchange/usa.news
$ ./spout -a topic-exchange/usa.sports
$ ./spout -a topic-exchange/usa.faux.news
$ ./spout -a topic-exchange/usa.faux.sports
- </programlisting>
+ </screen>
<para>Now look at the first window, and you will see the
messages with <literal>news</literal> in the last word of
the key have been received:</para>
- <programlisting>
+ <screen>
Message(properties={qpid.subject:news, spout-id:cbd42b0f-c87b-4088-8206-26d7627c9640:0}, content='')
Message(properties={qpid.subject:usa.news, spout-id:234a78d7-daeb-4826-90e1-1c6540781eac:0}, content='')
Message(properties={qpid.subject:usa.faux.news, spout-id:6029430a-cfcb-4700-8e9b-cbe4a81fca5f:0}, content='')
- </programlisting>
+ </screen>
</section>
@@ -517,29 +546,29 @@ Message(properties={qpid.subject:usa.fau
actually used to filter messages:</para>
- <para>First window:</para>
+ <para><emphasis>First Window:</emphasis></para>
- <programlisting>
+ <screen>
$ qpid-config add exchange fanout fanout-exchange
$ ./drain -a fanout-exchange/news -t 30
- </programlisting>
+ </screen>
<para>Now let's send a message using a different
subject:</para>
- <para>Second window:</para>
+ <para><emphasis>Second Window:</emphasis></para>
- <programlisting>
+ <screen>
$ ./spout -a fanout-exchange/sports
- </programlisting>
+ </screen>
<para>Returning to the first window, we see that the message
was received even though the Receiver's subject was
different from the Sender's subject:</para>
- <programlisting>
+ <screen>
Message(properties={qpid.subject:sports, spout-id:931399a1-27fc-471c-8dbe-3048260f9441:0}, content='')
- </programlisting>
+ </screen>
<para>This happens because of the routing semantics of the AMQP 0-10 fanout exchange.</para>
@@ -556,38 +585,39 @@ Message(properties={qpid.subject:sports,
<para>Let's create a queue and listen for messages on it.</para>
- <para>First window:</para>
+ <para><emphasis>First Window:</emphasis></para>
- <programlisting>
+ <screen>
$ ./qpid-config add queue amqp010-queue
$ ./drain -a amqp010-queue -t 30
- </programlisting>
+ </screen>
<para>Now let's send some messages. The subject is not used for routing purposes.</para>
- <programlisting>
+ <screen>
$ ./spout -a amqp010-queue/news
$ ./spout -a amqp010-queue
- </programlisting>
+ </screen>
<para>Now look at the first window, and you will see that
both messages have been received:</para>
- <programlisting>
+ <screen>
Message(properties={qpid.subject:news, spout-id:6c769437-60be-4bc0-9bf6-5a77cb6ba65f:0}, content='')
Message(properties={spout-id:c8ab5013-a19e-4f54-967c-797c8ad6568b:0}, content='')
- </programlisting>
+ </screen>
</section>
- <!-- ### header exchange? -->
+ <!-- ### match exchange? -->
+
<section>
<title>Custom Exchanges</title>
- <para>AMQP 0-10 also supports custom exchanges. The
- Qpid messaging broker includes the XML Exchange, which uses an
- XQuery to filter messages based on message properties and XML
- message content.</para>
+ <para>AMQP 0-10 also supports custom exchanges. The
+ Qpid messaging broker includes the XML Exchange, which uses an
+ XQuery to filter messages based on message properties and XML
+ message content.</para>
<!-- ### Do drain / spout support the XML Exchange? -->
@@ -597,7 +627,63 @@ Message(properties={spout-id:c8ab5013-a1
<section>
- <title>Options</title>
+ <title>Extended Address Options</title>
+<!-- #### Not a good name. The options don't configure the address.
+
+Not sure how best to fix this. One possibility is to change the names of the non-terminals.
+
+Instead of:
+
+address ::= <name> [ / <subject> ] [ ; <options> ]
+options ::= { <key> : <value>, ... }
+
+We could use something along these lines:
+
+configuration ::= <address> [ / <subject> ] [ ; <options> ]
+options ::= { <key> : <value>, ... }
+
+We could then have tables showing the available options. Some of these would be client configuration options, others would specify broker host/port/connection, etc.
+
+-->
+
+
+ <para>Extended Address Options are parameters that affect the behavior of Senders and Receivers.</para>
+
+ <para>Some of these options specify aspects of the resolution process; for instance, they may make assertions that must be satisfied in order for resolution to succeed, or they may state that the node should be created if it does not already exist.</para>
+
+ <para>Let's use <command>drain</command> and <command>spout</command> to show how this works. First, let's use the <literal>assert</literal> option to insist that an address must resolve to a queue. In the High Level Client API, a node is either a queue or a topic. A queue is used the same way as in AMQP 0-10, but for consistency with Java JMS, a topic node is the same thing AMQP 0-10 calls an exchange.(In this section, we will use the term <quote>topic node</quote> for a High Level Client API topic node, and <quote>AMQP 0-10 topic exchange</quote> for the exchange type that has the same name.) </para>
+
+ <para>AMQP 0-10 has several built-in exchanges that are predeclared: <literal>amq.topic</literal>, <literal>amq.direct</literal>, <literal>amq.match</literal>, and <literal>amq.fanout</literal>. To the High Level Exchange, any of these exchanges is considered a topic node. Let's use <command>drain</command>, and assert that <literal>amq.fanout</literal> is a topic node:</para>
+
+ <screen>$ ./drain -a "amq.fanout; { assert: always, node: { type: topic }}"</screen>
+
+ <para>The address resolves succesfully. No exception is raised, because a topic node named <literal>amq.fanout</literal> exists. Now let's assert that <literal>amq.fanout</literal> is a queue node:</para>
+
+ <screen>$ ./drain -a "amq.fanout; { assert: always, node: { type: queue }}"
+2010-04-09 14:01:35 warning Exception received from broker: not-found: not-found: Queue not found: amq.fanout (qpid/broker/SessionAdapter.cpp:753) [caused by 0 \x08:\x01]
+Queue amq.fanout does not exist</screen>
+
+ <para>An exception was raised because there is no queue named <literal>amq.fanout</literal>, so address resolution failed.</para>
+
+
+ <para>Now let's use the <literal>create</literal> option with <command>drain</command>, telling it to create the queue <literal>xoxox</literal> if it does not already exist:</para>
+
+ <para><emphasis>First Window:</emphasis></para>
+ <screen>$ ./drain -a "xoxox ; {create: always}" -t 30</screen>
+
+ <para>In previous examples, we created the queue before listening for messages on it. Using <literal>create: always</literal>, the queue is automatically created if it does not exist. Now we can send messages to this queue:</para>
+
+ <para><emphasis>Second Window:</emphasis></para>
+ <screen>$ ./drain -a "xoxox ; {create: always}" -t 30</screen>
+
+ <para>Returning to the first window, we see that <command>drain</command> has received this message:</para>
+
+ <screen>Message(properties={spout-id:1a1a3842-1a8b-4f88-8940-b4096e615a7d:0}, content='')</screen>
+
+<!--
+TODO: Add some x-declare, x-subscribe, link, x-bindings examples
+-->
+
<!--
In all the above cases, the address is resolved to an existing node.
@@ -610,18 +696,18 @@ Message(properties={spout-id:c8ab5013-a1
You can customize the properties of the queue::
- my-queue; {create: always, node-properties: {durable: True}}
+ my-queue; {create: always, node: {durable: True}}
You can create a topic instead if you want::
- my-queue; {create: always, node-properties: {type: topic}}
+ my-queue; {create: always, node: {type: topic}}
You can assert that the address resolves to a node with particular
properties::
my-transient-topic; {
assert: always,
- node-properties: {
+ node: {
type: topic,
durable: False
}
@@ -629,80 +715,248 @@ Message(properties={spout-id:c8ab5013-a1
; {create: always}'
-spout 'hello-whirled; {assert: always, node-properties: {type: topic}}'
-SH: spout 'small-world; {create: always, node-properties: {type: topic}}'
-SH: spout 'cruel-world; {create: always, node-properties: {type: topic, x-properties: {type: direct}}}'
-SH1: drain -f 'world-news; {create: always, node-properties: {x-properties: {bindings: ["small-world/news.#", "hello-world/news.#", "cruel-world/news"]}}}'
+spout 'hello-whirled; {assert: always, node: {type: topic}}'
+SH: spout 'small-world; {create: always, node: {type: topic}}'
+SH: spout 'cruel-world; {create: always, node: {type: topic, x-declare: {type: direct}}}'
+SH1: drain -f 'world-news; {create: always, node: {x-declare: {bindings: ["small-world/news.#", "hello-world/news.#", "cruel-world/news"]}}}'
SH: spout small-world/news.local
-->
-<!--
-#### What is the syntax for names?
--->
-<!--
- - we'll cover the precise details of the mapping later
- - an extended address can include a subject and options (we'll get to the options later)
- + subject is a standard message property
- - subjects classify messages
- + when you specify a subject in a target address, that value
- becomes the default subject property on outgoing Messages (this
- can be overriden by setting the subject explicitly on each
- messages)
-
-
- - extended address options adjust the semantics of senders/receivers
- + some options control the resolution process:
- - assert/create/delete, node-properties
- - more on this later
-
- + some options control message transfer semantics:
- - reliability: unreliable/at-least-once/at-most-once/exactly-once
- - mode: browse/consume (only for receivers)
--->
+ <para>Other options specify message transfer semantics; for instance, they may state whether messages should be consumed or read in browsing mode, or specify reliability characteristics. For instance, we can use browse mode to receive messages without removing them from the queue, thus allowing other Readers to receive them:</para>
+
+ <screen>
+$ ./drain 'hello-queue; {mode: browse}'
+ </screen>
<!--
-<rhs> POLICIES = Values("always", "sender", "receiver", "never")
-<rhs> RELIABILITY = Values("unreliable", "at-most-once", "at-least-once",
-<rhs> "exactly-once")
-<rhs> DECLARE = Map({}, restricted=False)
-<rhs> BINDINGS = List(Map({
-<rhs> "exchange": Types(basestring),
-<rhs> "queue": Types(basestring),
-<rhs> "key": Types(basestring),
-
-####### Is this ignored for sending in the AMQP 0-10 binding?
-
-<rhs> "arguments": Map({}, restricted=False)
-<rhs> }))
-<rhs> COMMON_OPTS = {
-<rhs> "create": POLICIES,
-<rhs> "delete": POLICIES,
-<rhs> "assert": POLICIES,
-<rhs> "node": Map({
-<rhs> "type": Values("queue", "topic"),
-<rhs> "durable": Types(bool),
-<rhs> "x-declare": DECLARE,
-<rhs> "x-bindings": BINDINGS
-<rhs> }),
-<rhs> "link": Map({
-<rhs> "name": Types(basestring),
-<rhs> "durable": Types(bool),
-<rhs> "reliability": RELIABILITY,
-<rhs> "x-declare": DECLARE,
-<rhs> "x-bindings": BINDINGS,
-<rhs> "x-subscribe": Map({}, restricted=False)
-<rhs> })
-<rhs> }
-<rhs> RECEIVE_MODES = Values("browse", "consume")
-<rhs> SOURCE_OPTS = COMMON_OPTS.copy()
-<rhs> SOURCE_OPTS.update({
-<rhs> "mode": RECEIVE_MODES
-<rhs> })
-<rhs> TARGET_OPTS = COMMON_OPTS.copy()
+TODO: Add some reliability option examples
-->
-<para></para>
+
+
+ <table>
+ <title>Extended Address Options</title>
+ <tgroup cols="3">
+ <thead>
+ <row>
+ <entry>option</entry>
+ <entry>parameters</entry>
+ <entry>semantics</entry>
+ </row>
+ </thead>
+ <tbody>
+ <row>
+ <entry>
+ assert
+ </entry>
+ <entry>
+ node
+ </entry>
+ <entry>
+ Asserts that the node properties are satisfied for a
+ node. If they are not, node resolution fails and an
+ exception is raised. <!-- ### Which exception -->
+ </entry>
+ </row>
+
+ <row>
+ <entry>
+ create
+ </entry>
+ <entry>
+ node (optional)
+ </entry>
+ <entry>
+ Creates the node if it does not exist. No error is raised if the node does exist.
+ </entry>
+ </row>
+ <row>
+ <entry>
+ delete
+ </entry>
+ <entry>
+ N/A
+ </entry>
+ <entry>
+ Delete the node when the Sender or Receiver is closed.
+ </entry>
+ </row>
+ <row>
+ <entry>
+ reliability
+ </entry>
+ <entry>
+ unreliable, at-least-once, at-most-once, exactly-once
+ </entry>
+ <entry>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ mode
+ </entry>
+ <entry>
+ browse, consume
+ </entry>
+ <entry>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ reconnect
+ </entry>
+ <entry>
+ True, False
+ </entry>
+ <entry>
+ Transparently reconnect if the connection is lost.
+ </entry>
+ </row>
+ <row>
+ <entry>
+ reconnect_timeout
+ </entry>
+ <entry>
+ N
+ </entry>
+ <entry>
+ Total number of seconds to continue reconnection attempts before giving up and raising an exception.
+ </entry>
+ </row>
+ <row>
+ <entry>
+ reconnect_limit
+ </entry>
+ <entry>
+ N
+ </entry>
+ <entry>
+ Maximum number of reconnection attempts before giving up and raising an exception.
+ </entry>
+ </row>
+ <row>
+ <entry>
+ reconnect_interval_min
+ </entry>
+ <entry>
+ N
+ </entry>
+ <entry>
+ Minimum number of seconds between reconnection attempts. The first reconnection attempt is made immediately; if that fails, the first reconnection delay is set to the value of <literal>reconnect_interval_min</literal>; if that attempt fails, the reconnection interval increases exponentially until a reconnection attempt succeeds or <literal>reconnect_interval_max</literal> is reached.
+ </entry>
+ </row>
+ <row>
+ <entry>
+ reconnect_interval_max
+ </entry>
+ <entry>
+ N
+ </entry>
+ <entry>
+ Maximum reconnection interval.
+ </entry>
+ </row>
+ <row>
+ <entry>
+ reconnection_interval
+ </entry>
+ <entry>
+ N
+ </entry>
+ <entry>
+ Sets both <literal>reconnection_interval_min</literal> and <literal>reconnection_interval_max</literal> to the same value.
+ </entry>
+ </row>
+ </tbody>
+ </tgroup>
+ </table>
+
+
+ <table>
+ <title>Node Properties</title>
+ <tgroup cols="3">
+ <thead>
+ <row>
+ <entry>property</entry>
+ <entry>parameters</entry>
+ <entry>semantics</entry>
+ </row>
+ </thead>
+ <tbody>
+ <row>
+ <entry>
+ type
+ </entry>
+ <entry>
+ topic, queue
+ </entry>
+ <entry>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ durable
+ </entry>
+ <entry>
+ True, False
+ </entry>
+ <entry>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ x-declare
+ </entry>
+ <entry>
+ unrestricted map
+ </entry>
+ <entry>
+ If the property is defined in the underlying protocol (AMQP 0-10), the values and semantics are defined by the protocol. Otherwise, values are added to the arguments map used to declare topics<footnote><para>The High Level Client API, like Java JMS, uses the term topic to refer to what AMQP 0-10 calls an exchange. One kind of AMQP 0-10 exchange is called a topic exchange, that is not what is meant here.</para></footnote> or queues.
+ </entry>
+ </row>
+ </tbody>
+ </tgroup>
+ </table>
+
+
+
+ <table>
+ <title>x-declare properties for AMQP 0-10</title>
+ <tgroup cols="3">
+ <thead>
+ <row>
+ <entry>property</entry>
+ <entry>parameters</entry>
+ <entry>semantics</entry>
+ </row>
+ </thead>
+ <tbody>
+ <row>
+ <entry>
+ type
+ </entry>
+ <entry>
+ direct, topic, fanout, header, xml
+ </entry>
+ <entry>
+ The AMQP 0-10 exchange type.
+ </entry>
+ </row>
+ <row>
+ <entry>
+ bindings
+ </entry>
+ <entry>
+ ["exchange/binding-key", ... ]
+ </entry>
+ <entry>
+ This property is used to create bindings for queues. If the Address does not resolve to a queue, an error is raised.
+ </entry>
+ </row>
+ </tbody>
+ </tgroup>
+ </table>
</section>
@@ -715,6 +969,24 @@ SH: spout small-world/news.local
properties are mapped to AMQP message properties and delivery
properties.</para>
+ <para>Request-response applications frequently use a reply-to property to tell a server where to send a response. The following code shows how a server extracts the reply-to property and uses it to set the address to respond to a client.</para>
+
+ <programlisting>
+ Message request = receiver.fetch();
+ const Address& address = request.getReplyTo(); <lineannotation>Get "reply-to" field from request ...</lineannotation>
+ if (address) {
+ Sender sender = session.createSender(address); <lineannotation>... and use it as the address to send response</lineannotation>
+ std::string s = request.getContent();
+ std::transform(s.begin(), s.end(), s.begin(), toupper);
+ Message response(s);
+ sender.send(response);
+ std::cout << "Processed request: "
+ << request.getContent()
+ << " -> "
+ << response.getContent() << std::endl;
+ session.acknowledge();
+ </programlisting>
+
<para>In the following table, <varname>msg</varname> refers to the
Message class defined in the High Level Client API,
<varname>mp</varname> refers to an AMQP 0-10
@@ -780,37 +1052,6 @@ Examples - do client / server, pub-sub h
</chapter>
<!--
- - drain and spout are basically command line versions of the API
- + drain is a receiver that operates on an any message source
- - drains all available messages from a source and then exits
- - behavior can be further controlled with options
-SH: drain -h
- + spout is a sender that sends to any message target
- - generates a message and sends it to a target
- - message content can be controlled with options
-SH: spout -h
-
- - source and target are specified as addresses
- + addresses are symetric (sort of), any target is a source, but not
- all sources are targets, e.g. sources might add filtering
-
- - an address is just a name
- + the name is resolved to a broker entity prior to sending/receiving
- - spout/drain to a nonexistent address is an error
-SH: spout hello-world
- + the same error will occur when using the API to create
- senders/receivers
- + normally broker entities are created outside the API
-SH: qpid-config add queue hello-world
- - once created, an address simply refers to them by name
-SH: spout -c 10 hello-world
-SH: drain hello-world
-SH: qpid-config
- + this works the same for exchanges
-SH: qpid-config del queue hello-world
-SH: qpid-config add exchange topic hello-world
-SH: spout -c 10 hello-world
-SH: drain hello-world
- client code remains exactly the same, but routing behavior
changes
- exchanges drop messages if nobody is listening, so we need to
@@ -821,60 +1062,7 @@ SH: drain hello-world
reports it, no fudge factor timeout is required [this assumes
nobody is concurrently publishing of course])
- drain -f invokes blocking fetch (you could use a timeout here also)
-SH1: drain -f hello-world
-SH: spout -c 10 hello-world
-SH2: drain -f hello-world
-SH: spout -c 10 hello-world
- - multiple drains will get all messages because this is an
- exchange
- - for a queue messages will be load balanced between drains
-SH: qpid-config add queue hello-queue
-SH1: drain -f hello-world
-SH2: drain -f hello-world
-SH: spout -c 10 hello-world
- + an address is resolved to a node
- - the API internals will adjust how they send/receive based on
- the type of node
- - in AMQP 0-10 exchanges and queues are the two standard
- categories of nodes
- - in JMS these are called topics and queues
- - we use the topic terminology to be consistent with JMS (note
- that when used in this sense topic refers to any exchange, not
- just a topic exchange)
- - we'll cover the precise details of the mapping later
-
- - an extended address can include a subject and options (we'll get to the options later)
- + subject is a standard message property
- - subjects classify messages
- + when you specify a subject in a target address, that value
- becomes the default subject property on outgoing Messages (this
- can be overriden by setting the subject explicitly on each
- messages)
-SH1: drain -f hello-world
-SH: spout hello-world/how.are.you
- + when you specify a subject in a source address, the value is used
- to filter messages from the source based on matching the value
- against available message subjects
-SH2: drain -f hello-world/how.are.you
-SH: spout hello-world
-SH: spout hello-world/how.are.you
-SH2: drain -f hello-world/how.#
-SH: spout hello-world
-SH: spout hello-world/how.are.you
-SH: spout hello-world/how.cruel.you.are
- - the pattern matching algorithm depends on the type of broker
- entity the address resolves to (e.g. fanout vs topic vs direct)
- - note that this doesn't yet work with queues because we can only
- filter going into the queue at present (i.e. no server side
- selectors)
-
- - extended address options adjust the semantics of senders/receivers
- + some options control the resolution process:
- - assert/create/delete, node-properties
- - more on this later
- + some options control message transfer semantics:
- - reliability: unreliable/at-least-once/at-most-once/exactly-once
- - mode: browse/consume (only for receivers)
+
SH: spout -c 10 hello-queue
SH: drain 'hello-queue; {mode: browse}'
SH: drain 'hello-queue; {mode: browse}'
@@ -884,131 +1072,12 @@ SH: drain 'hello-queue; {mode: consume}'
SH: # durable demo???
- no-local: True/False (only for receivers from topics/exchanges)
SH: #- filter: XXX
-
- - API mapping to AMQP 0-10
- + for 0-10 an address is resolved to a queue and/or exchange by
- querying for them both by name
- - don't ask the obvious question
- + message properties are NOT directly mapped to/from the protocol,
- reference the following table for translation
-->
<!--
- Python API C++ API AMQP 0-10 Protocol Mapping
- |=========================================================================================
- <row>
- <entry>msg.id </entry><entry>msg.{get,set}MessageId()</entry><entry>mp.message_id
- <row>
- <entry>msg.to </entry><entry>- - </entry><entry>mp.application_headers["qpid.to"]
- <row>
- <entry>msg.subject </entry><entry>msg.{get,set}Subject() </entry><entry>mp.application_headers["qpid.subject"]
- <row>
- <entry>msg.user_id </entry><entry>msg.{get,set}UserId() </entry><entry>mp.user_id
- <row>
- <entry>msg.reply_to </entry><entry>msg.{get,set}ReplyTo() </entry><entry>mp.reply_to *
- <row>
- <entry>msg.correlation_id | msg.{get,set}CorrelationId() | mp.correlation_id
- <row>
- <entry>msg.durable </entry><entry>msg.{get,set}Durable() </entry><entry>dp.delivery_mode == delivery_mode.persistent **
- <row>
- <entry>msg.priority </entry><entry>msg.{get,set}Priority() </entry><entry>dp.priority
- <row>
- <entry>msg.ttl </entry><entry>msg.{get,set}Ttl() </entry><entry>dp.ttl
- <row>
- <entry>msg.redelivered | msg.isRedelivered() </entry><entry>dp.redelivered
- <row>
- <entry>msg.properties</entry><entry>msg.{get,set}Headers() </entry><entry>mp.application_headers
- <row>
- <entry>msg.content_type | msg.{get,set}ContentType() | mp.content_type
-
-
- msg: Message class in API
- mp: message-properties protocol struct
- dp: delivery-properties protocol struct
-
- *: the reply_to is converted from the protocol representation into an address
- **: note that msg.durable is a boolean, not an enum
-
- + recall that sending and receiving is adjusted based on the node
- type, there are four basic cases to cover:
- - sending to an exchange
- + the transfer destination is set to the exchange name
- + the routing key is set to the value of the subject (copied from the subject property)
- - receving from an exchange
- + a subscription queue is automatically created and bound to the exchange
- - durability of subscription queue is set based on durable address option
- - auto-delete is set based on reliability address option
- + if no subject is specified, then a default binding key is
- chosen based on the exchange type:
- - topic: wildcard match
- - direct: *error*
- - fanout: N/A
- + if a subject is specified, it is used as the binding key
- - sending to a queue
- + the transfer destination is set to the default/no-name exchange
- + the routing key is set to the queue name
- - receiving from a queue
- + trivially maps to a normal subscribe
- - accept-mode for receiving (from regular or subscription queue)
- is set based on reliability options
-
- - create/delete/assert options
- + these options can all be set to one of: sender, receiver, always, never
- - the value controls when the option is in effect: when used as a
- sender target, when used as a receiver source, always, or never
- + these options specify policies that can be customized via the
- additional node-properties option
- + by default it's an error when resolution fails, options can adjust this
- - assert makes the resolution process more picky by matching the
- resolved node against the specified node-properties
- - if the node doesn't exist, and the create option is in effect
- the node will be created with the specified node-properties (we
- default to a queue if no properties are present)
-SH: spout hello-whirled
-SH: spout 'hello-whirled; {create: always}'
-SH: qpid-config queues
- + if the delete option is in effect, then the node will be deleted
- when the sender and/or receiver is closed
- - note that this option needs broker support to be safely used
- outside of very simple scenarios
- + node-properties:
- - type: topic or queue
- - durable: True/False
- - x-properties: unrestricted map
- + type: (specifies the exchange type when the address type is a topic)
- + bindings: ["exchange/binding-key", ...] (only works for queues)
- + anything else that matches a protocol field gets set in the
- exchange/queue declare
- + anything else that does *not* match a protocol field gets
- added to the arguments map in the exchange/queue declare
-SH: #spout 'hello-whirled; {assert: always, node-properties: {type: topic}}'
-SH: spout 'small-world; {create: always, node-properties: {type: topic}}'
+SH: #spout 'hello-whirled; {assert: always, node: {type: topic}}'
+SH: spout 'small-world; {create: always, node: {type: topic}}'
SH: qpid-config exchanges
-SH: spout 'cruel-world; {create: always, node-properties: {type: topic, x-properties: {type: direct}}}'
+SH: spout 'cruel-world; {create: always, node: {type: topic, x-declare: {type: direct}}}'
SH: qpid-config exchanges
-SH1: drain -f 'world-news; {create: always, node-properties: {x-properties: {bindings: ["small-world/news.#", "hello-world/news.#", "cruel-world/news"]}}}'
-SH: qpid-config queues -b
-SH: spout small-world/news.local
-SH: spout hello-world/news.friendly
-SH: spout cruel-world/news
-
- - connections can be configured to automatically reconnect, this
- behavior is controlled by the following options
- + reconnect: True/False (enables/disables reconnect entirely)
- + reconnect_timeout: number of seconds (give up and report failure after specified time)
- + reconnect_limit: n (give up and report failure after specified number of attempts)
- + reconnect_interval_min: number of seconds (initial delay between failed reconnection attempts)
- + reconnect_interval_max: number of seconds (maximum delay between failed reconnection attempts)
- + reconnect_interval: shorthand for setting the same reconnect_interval_min/max
- + first reconnect attempt is made immediately, if that fails an
- exponentially increasing delay controlled by above parameters is
- introduced
-SH1: drain -rf amq.topic
-
-TODO:
- - don't forget to remove pattern munging from python client!
- - get more advanced filtering in python (XML exchange, etc)
- - define behavior if both exchange and queue exist
- - figure out how to add passthrough options to other things like
- binding and subscription, maybe use this for accessing things like
- the xml exchange
+SH1: drain -f 'world-news; {create: always, node: {x-declare: {bindings: ["small-world/news.#", "hello-world/news.#", "cruel-world/news"]}}}'
-->
\ No newline at end of file
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org