Added: qpid/trunk/qpid/doc/book/src/FAQ.xml URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/doc/book/src/FAQ.xml?rev=898329&view=auto ============================================================================== --- qpid/trunk/qpid/doc/book/src/FAQ.xml (added) +++ qpid/trunk/qpid/doc/book/src/FAQ.xml Tue Jan 12 14:05:42 2010 @@ -0,0 +1,504 @@ +{toc} + + +*This page is a collection of FAQ and How to-s for Qpid. If you have a question, post it to the users list and we will place the answer here to build out our FAQ/ How to.* + +h1. FAQ + +h2. About AMQP + +h3. What is AMQP? + +AMQP is a wire-level protocol and model for high performance enterprise messaging. + +[From the AMQP website:|http://www.amqp.org] + + AMQP is an Open Standard for Messaging Middleware. + + By complying to the AMQP standard, middleware products written for different platforms and in different languages can send messages to one another. AMQP addresses the problem of transporting value-bearing messages across and between organizations in a timely manner. + + AMQP enables complete interoperability for messaging middleware; both the networking protocol and the semantics of broker services are defined in AMQP. + +h3. Where did AMQP come from + +AMQP was born out from Frustration by John O'Hara at JPMC. He started a project internally to create commodity messaging that was easy to use. Carl Trieloff from Red Hat had started a project to build messaging for both users and for use in infrastructure, while looking around spoke to John about his work. Out of these discussion was born the AMQP working Group with 6 initial members, under an agreement that it will be eternally be licensed for everyone to use. + +Since then the Working Group has had many join, and has been making solid progress working on revisions of the specification. [For more details see.|http://jira.amqp.org/confluence/display/AMQP/About+AMQP] + +h3. Why use AMQP? + +AMQP is has been designed to be able to handle the hardest workloads, scale to the largest systems, but also deal with reduction of change and maintenance costs by doing a refresh on many aged practices. The specification is also not language specific allowing the freedom from language and platform lock in, without compromise on user experience, security, scalability and consistently excellent performance. + +[Text mostly taken from|http://jira.amqp.org/confluence/display/AMQP/About+AMQP] + +h2. Qpid & AMQP + + +h3. Is Qpid AMQP Compliant? + +Yes, Apache Qpid implements the latest AMQP specifications, providing transaction management, queuing, distribution, security, management, clustering, federation and heterogeneous multi-platform support and a lot more. And Apache Qpid is extremely fast. [Apache Qpid aims to be 100% AMQP Compliant|AMQP compatibility]. + +h3. What Client support does Qpid have? + +Apache Qpid provides AMQP Client APIs for the following languages: +* C+\+ +* C# .NET, using WCF +* Ruby +* Python +* Java JMS, fully conformant with Java CTS1.1 + +If you need another client, join the lists and ask or feel free to contribute one. + +h3. What messaging topologies are supported by AMQP and Qpid? + +AMQP provides the ability to do Point-to-Point, Peer-to-Peer, Pub-Sub, and Eventing. This allows many patterns to be craeted: + ++*Point-to-point*+ + +This is one of the simplest use-cases. AMQP allows for this in a few ways. + a.) A client can create a named queue allowing the producer to publish the message to the direct exchange with the key mapping the queue name. This will route the message to that queue. + b.) The above pattern can be extended by specifying a reply-to address in the published messages allowing for the consumer to reply the producer without knowing who it was send from prior to receiving the message. + ++*One-to-many*+ + +There are a few patterns that can be used. + + a.) AMQP provides a 'fanout' exchange which will send a message to all the queues that have been bound to it. Different domains or topics are created with the 'fanout' exchange by declaring different named fan-out exchanges. + + b.) A 'topic' or 'headers' exchange can also be used. in this case the pattern match is used to send the message to all the bound queues. It can be thought of as a filter allowing you to create just about any One-to-many routing patterns. + ++*Pub-Sub*+ + +Topic can be created with the 'topic' or other 'direct' exchange to allow consumer to bind to into the steams of data they care about. This pattern combined with the use of reply-to and Alternate-routing is the staple of what most people use messaging for today. + ++*FAST Reliable Messaging*+ + +AMQP 0-10 allows for fully reliable transfers between any two peers. This means that you can publish or subscribe to the broker fully reliable without requiring the need for transactions. This can all be done in async mode with the C++ broker allowing for high throughput while running entirely reliable. + ++*Transactional*+ + +AMQP supports two types of transactions in AMQP 0-10, TX and DTX. This allows for local (1PC), and 2PC transaction and the ability to coordinate with a TM (Transaction Manager). The Java broker supports TX, the C++ broker support TX, DTX, XA, JTA for fully ACID transactions. This allows you to commit a single unit of work with may contain enqueues & dequeues either locally on the broker, or in coordination with other transactional resource like RDBMS. + ++*Transient message delivery*+ + +By default messages are transient. Transient message can be sent to queues that are durable. They will not be safe stored or recovered, and will perform as any other transient message - fast! + ++*Durable message delivery*+ + +There is a header on each message where the message properties are specified, one of these is durability. Messages that are marked as durable and published to a durable queue will be safe stored. Durable messages will survive restart of the broker or cluster. + ++*Federation (Hub-spoke, Trees, graphs)*+ + +As AMQP 0-10 is symmetric for peer-to-peer communication all the building block are in place for creating networks of brokers. The C++ broker allows you to link the brokers together using 'qpid-route' and then create routes between the brokers either statically or with dynamic routes. + +This allows for a message to be published to one broker and consumed from another broker in the federated broker network. This feature is great to create data-center, or project isolation, but allow cross communication. It also allows networks to be created to scaled. [For more details see|Using Broker Federation] + ++*And many others, including custom pattern*+ + ++*Message Reply, Rings, Initial Value Caches, Last Value Messaging*+ + +All the above cases can be constructed using the AMQP and features of Qpid. For example reply can be constructed using message browsing and setting TTL on the messages. The C++ broker also support ring queues, last value queues, initial value caches on exchanges. With a bit of throught many additional patterns can be constructed. + +*+Store-and-forward+* + +Store-and-forward can be achieved by publishing to well know durable queues, that are not marked with auto delete. Consumers will be able to 'came back' to consume then at any time, even after restarts. + +h3. What AMQP and other exchanges does Qpid support? + +Both brokers support: + * Direct Exchange + * Topic Exchange + * Fanout Exchange + * Headers Exchange + +In additional the C++ broker support + * XML Exchange - Query routing + * Custom exchange via plug-in. + +Custom exchanges allow you to provide your own custom routing logic and algorithms via a plug-in. If you build an interesting exchange, please feel free to contribute it back to the Qpid project. + +h2. Security + +h3. What encryption does Qpid support? + + * Qpid support SSL/TSL as per the AMQP specification. + * In addition the C++ broker supports Kerberos encryption of messages independent on which transport is used. Support in not yet included in all clients for this but is in process. + +h3. What authentication does Qpid support? + +SASL Authentication is supported. All Clients support PLAIN, and Kerberos support if being added to all the clients. The C++ broker support Kerberbos authentication. + +h3. What authorization does Qpid support? + +Full ACL is supported in the brokers. [For details on configuring ACL see|Qpid ACLs]. + +ACL supports realms and allows for granular permission to be set on all the broker actions including management on an user or group basis. + +h3. How to setup Kerberos with the Java client + +You could force the java client to use kerberos auth by specifying it in the connection URL as follows. +{code} +amqp://guest:gu...@clientid/testpath?brokerlist='tcp://localhost:5672?'&sasl_mechs='GSSAPI' +{code} + +You would then need to pass in the following jvm arguments +{code} +-Djavax.security.auth.useSubjectCrehttp://code.google.com/p/confluence-el/dsOnly=false +# (This will force the SASL GASSPI client to obtain the kerberos credentials explicitly instead of obtaining from the "subject" that owns the currents thread) +-Djava.security.auth.login.config=myjas.conf (this specifies the jass config file) +-Dsun.security.krb5.debug=true (to enable detailed debug info for troubleshooting) +{code} + +Before running the java client you would need to do kinit and grab a kerberos ticket. Alternative you could set useTicketCache=false and when the client loads, it will prompt you for the user/pass and will obtain the ticket +(You would also need to setup your kerberos environment properly -refer to doc links below). + +Sample JASS Config file +{code} +com.sun.security.jgss.initiate { + com.sun.security.auth.module.Krb5LoginModule required useTicketCache=true; +}; +{code} + + +h2. Semantics of Exclusive + +h3. I want to be able to have an exclusive consumer, but when it dies I want another to be able to pick up the queue and then block others, can this be done? + +Yes, Declare you queue exclusive. this will prevent anyone else from connecting to the queue. If the consumer dies the next consumer can attach to the queue by redeclaring it using the exclusive flag. Make sure not to set auto delete. Any consumer trying to declare, while a consumer is attached to the queue will receive an exception. + +h3. When will the queue become free for a re-declare + +Once the session that held the consumer is closed. + + +h2. Performance + +h3. Does Qpid Perform (Latency/Throughput)? + +Yes, The Qpid C++ broker has been achieved great benchmark results in published papers by those that redistribute it. [Red Hat MRG|http://www.redhat.com/mrg] product build on Qpid has shown 760,000msg/sec ingress on an 8 way box or 6,000,000msg/sec OPRA messages. + +Latencies have been recored as low as 180-250us (.18ms-.3ms) for TCP round trip and 60-80us for RDMA round trip using the C++ broker. + +h3. How do I measure throughput? + +There is a great resource supplied in the C++ broker test directory called perftest. If allows you to create load on a broker for all the exchanges, multiple queues, multiple connection, coordinate multiple publishing and consuming processes, beachmark transactions and much much more such as acquire mode, txn size, message size. + +For all the options +{code} +./perftest --help +{code} + +h3. How do I measure latency? + +There is a great resource supplied in the C++ broker test directory called latencytest. It is a loopback test that produces messages by count or at a rate, time stamps them and then consumes them back and record the latency. It supports many of the Qpid options, including the ability to vary things like frame-size. + +Latencies to expect round trip: + * 1G TCP ~ .3ms -.5ms + * 10G TCP - .18ms - .22ms + * RDMA transport - 40us - 80us + +Don't forget to set tune the machine and set --tcp-nodelay on both the C++ broker & client. + +For all the options +{code} +./latencytest --help +{code} + +h3. How do I measure performance with Java clients? + +In Java we provide a utility called QpidBench. It allows you to test the performance of the native AMQP API in Java for 0-10 and the JMS API against both brokers. + +h3. Can I run my Java client with JAVA-RT? + +Yes, recently a thread abstraction layer has been added to the Java client allowing it to be used with both the SUN and IBM RT JVMs. This increases the determinism of latency when using the Java client. + +h3. Does Qpid support flow control? + +yes, AMQP 0-10 allows for flow control on the consumer and producer. + +h3. How do I configure producer side flow control + +from qpidd --help + +set the following in the config file on via cmd line options. +{code} + --max-session-rate MESSAGES/S (0) Sets the maximum message rate per session (0=unlimited) +{code} + +h2. Management + +h3. What Management does Qpid support + +The Java broker supports JMX and provides an Eclipse plug-in and command line tool to manage via JMX. The C++ broker has far more extensive management support via QMF which will be added to the Java broker in a future release. + +The C++ Broker supports a layered management protocol over AMQP called QMF. This allows for the management of resource either in the broker or connected to the broker via the AMQP fabric. This management includes statistics, control, eventing, and reporting/updating properties. + +h3. How do I manage a broker? + +A set of tools are provided to manage the C++ broker, they include + * qpid-tool - telnet type tool to access data, view schema, issue command an and QMF resource + * qpid-config - tool to configure queues, exchanges, etc. all the details on the AMQP model + * qpid-route - tool to configure broker federation + * qpid-events - utility that will print to cmd line or syslog event from a broker like, userconnected, user crested/deleted a queue. + * qpid-stats - utility that will print out queue statistics to the cmd line or syslog like rate and message depth. + +Then you can also access all thsi information via JMX or WS-DM (work in progress) using QMan. + +h3. What logging tracing and events does Qpid support? + +Qpid support the ability to output events from any the broker or any managed object via QMF, or to do a variety of logging from the broker & clients. for tracing options run qpidd --help. + +Multiple levels of of logging are supported in the C++ broker from debug, warning, error, info, etc -- all of which can be filtered. + +h3. Can I get to all the management data from a client? + +yes, All the management data is just AMQP messages on specially named queues. An API is provided for working with the management data called QMFC + +h3. What is QMF + +QMF is the layered Management protocol used to manage the C++ broker. For details on teh protocol see the Development pages. + +QMF allows you to manage any resource and provides the following infrsstructure: + * Properties + * Statistics + * Commands + * Events + * Schema for resources and versioning + * tools for creating agents and consuming QMF data. + +h3. What are QMF Agents, and what do they do for me? + +An Agent is any client (producer or consumer) that generates a QMF schema and registers itself to be management by QMF. + +A great use case of this is a consumer that is processing order from a queue can reference itself to that queue and for example provide a schema for the number or successful orders process and a method to suspend processing. Now it becomes possible to use qpid-tool to connect to the broker, see which order processors are on queue via the reference and the via the stats of the order processor client. It is also possible to issue a command to the client via qpid-tool to suspend processing. ACL in the broker can be applied to all these actions if desired. + +h3. What is QMFC and what does it do for mr? + +QMFC is the API used to consume QMF data, event and issue commands to QMF agents from an AMQP client. + +h3. What is QMan + +Qman is a tool that dynamically reads the QMF Schema information and creates JMX objects that consumed by any JMX console or application server to manage Qpid. QMan is also adding support for WS-DM management of QMF resources. + + +h2. Clustering, Federation and Disaster Recovery + +h3. Does Qpid provide Fault Tolerance for the cluster? + +The C++ broker has plug-ins for Active-Active clustering which keep all the nodes of the cluster in sync. This means that any action that is performed on one of the brokers on the cluster is performed on all of them at the same time. New nodes can be added to the cluster at any time, and removed at any time with no consequences, exept for the extra multi-cast load created for the sync on joining. + +h3. What does the cluster guarantee? + +Everything! All configuration, all messages and all actions are replicated in a cluster. This means that two consumers can be connected to different nodes in the cluster and they will behave EXACTLY the same as if they where on a single broker. + +h3. Do clients get notified members joining or leaving the cluster? + +yes, All clients are updated with the addresses of node add/removed as supported by the AMQP 0-10 specification. This means that the client can dynamically track the nodes in the cluster and reconnect as required. + +h3. Can I specify more than one host to connect initially to the cluster to avoid single point of failure? + +yes, the AMQP address is multi-honed and more than one IP address can be specified at the initial connection. The client will then iterate through the host until it makes a successful connection. This feature can also be used in none clustered brokers. + +h3. How does Clustering work? + +When C++ brokers are configured into a cluster, the nodes communicate with each other over a mulitcast protocol called AIS, an open Telco multicast protocol that provides all the quorum and group services. + +Every action that is performed on any node of the cluster is then sequenced via totem and then performed on each node of the cluster in sync. As the cluster backbone is multicast, a separate network can be used for cluster communication and there is little impact adding additional nodes to the cluster with-in reason. + +h3. What is Federation? + +Federation provides the ability to create networks of brokers that communicate with each other in all types of typologies. This allows a producer to publish messages to one broker and someone to consume the messages from another broke somewhere on the broker federated network. + +[For more details see|Using Broker Federation] + +h3. Disater recover features are in process, Q&A will be added once they are complete. + +h2. Heartbeats + +Heartbeat can be configured to allow clients to detect when a broker has failed and connect to another broker or cluster member.Heartbeats are sent by the broker at a client specified, per-connection frequency. If the client does not receive a heartbeat or any other traffic for two heartbeat intervals, the connection will be made to fail. + +h3. What would happen when there is a no heartbeat within a predefined interval? + +If there is no traffic for two heartbeat intervals, the client will fail the connection. The application will see the exact same response as when the connection is killed. + +h3. What happens if the broker is unable to send heartbeat? + +As above, if there is no other traffic the client will eventually kill the connection. + +h3. Does the client retry? + +You can control the heartbeat interval on the client through the heartbeat member of ConnectionSettings (it is measured in seconds). Some of the options on policies do vary for different clients. + +h3. Failover taking too long... + +First check to make sure a heartbeat has been specified in the connection properties for the connection. + +Then make sure that the interfaces on each broker are reachable from the host you run my clients, else it will take a long time for the socket to timeout until it gets to one that can be reached. + +Make sure the list of URL's on the cient are the ones you want tht client to try + +Make sure that the broker is only exporting URL's that the client can connect to, use the --cluster-url option on the broker to specify this. + +h2. Threading + +h3. Could someone provide a brief description of the worker thread duties in the current Qpid release? + +The broker uses IO threads for all the work it does. This means that when work is signalled via an event (socket, RDMA, timer) an IO thread is scheduled and it runs until it completes the work and then returns back to the IO thread pool. This allows the CPUs to be utilized efficiently. The general rule is that we allocate 1 thread per core +1. So on a 8 way machine you see worker-threads default to 9. On a 4 way it will be 5. Sometimes it if work changing the default allocation if: + +a.) you run on high core count machine >8 to a lower number +b.) if you taskset, then set to the cores allocated +1 + +h3. Why was the number X chosen as the default number of worker threads? + +Qpidd defaults to cores + 1 + +h3. What happens in parallel? + +Concurrency in the broker is at the session level. So yes. If you want more concurrency, create another session on the same connection. + +h3. How are worker threads allocated to individual client sessions if there are more clients than threads in the pool? + +They are not allocated to a specific client + +h2. Persistence + +h3. Does Qpid support persistence (durability)? + +Yes, there are third-party (non-Apache) modules for both C++ and Java. Historically, BDB has been used to provide persistence for both C++ and Java. However, this has created a licensing conflict with Apache, and thus the store modules are maintained off-site. + +The Java broker includes a fully Apache licensed persistent store that uses Derby DB. + +The terms _durable_ and _persistent_ are used interchangeably in this FAQ. + +h3. Where do I get the 3rd party persistence store modules? + +The 3rd party persistence store modules may be obtained through anonymous subversion at the following locations: + +C++: http://anonsvn.jboss.org/repos/rhmessaging/store/trunk/cpp +Java: http://anonsvn.jboss.org/repos/rhmessaging/store/trunk/java/bdbstore + +For further details see [3rd Party Libraries] + +h3. How do I build the persistence store module from subversion checkouts? + ++*C\+\+*+ +The README file contains detailed instructions, but here is a summary: + # Make sure that both the db4-devel and libaio-devel packages are installed prior to building. + # Make sure that qpid is built and you know the location of the qpid directory (ie the top-level directory containing the python and cpp sub-directories). + # In the store directory, run: +{code} +./bootstrap +./configure --with-qpid-checkout=/abs/path/to/qpid/dir +make +{code} +# When built, the store library *msgstore.so* will be located in the *lib/.libs* directory. + ++*Java*+ +TODO + +h3. How do I use the persistence store module? + ++*C\+\+*+ + # Start the broker making sure that the store module is loaded, ie +{code} +qpidd --load-module=/path/to/msgstore.so --data-dir=/path/to/store-files ... +{code} + # Make sure that queues that will handle persistent messages are set durable. +{note:title=Note: Existing non-persistent queues cannot be made persistent} +If a queue has been declared without persistence, doing so again with persistence enabled while the old queue still exists in the broker will be ignored. Make sure that when a queue is declared persistent, there is no non-persistent queue of the same name in existence. +{note} + # For each message sent to a durable queue, make sure that it is set durable. + ++*Java*+ +TODO + +h3. How do I configure the persistence store? + ++*C\+\+*+ + +The broker loads help information from each module. To see the help options for the store, load the store module and specify help: +{code} +qpidd --load-module /abs/path/to/store/lib/.libs/msgstore.so --help +{code} + +Note that a set of journal files will be created for each queue declared and marked persistent. Each persistent queue has its own private journal. These are stored in the data directory by default (ie it uses the broker's *\-\-data\-dir* setting) or can be overridden with the *\-\-store\-dir* option. Note that if the broker is started with the *\-\-no\-data\-dir* option, then no store default exists, and the *\-\-store\-dir* option MUST be specified. + +The store file details - or "store geometry" - can be set with command-line options. These include the size and number of files that make up the journal for each queue. The *\-\-num\-jfiles* options sets the number of files to use (between 4 and 64) and the *\-\-jfile\-size\-pgs* sets the size of the file in 64kiB blocks. + +The size of the pages in the write page cache is set with the *\-\-wcache\-page-size* option, and sets a size in KiB. (Legal values are powers of 2, ie: 1, 2, 4, 8, 16, 32, 64, 128). Typically small page sizes give improved latency (especially for small messages), but are bad for message throughput, while large page sizes improve throughput but may cause some messages to have higher latencies. + ++*Java*+ ++Derby Store+ +For details of configuring the Derby Store see [here|Derby Store Plugin] + ++3rd Party Stores+ + +For details of using the 3rd party persistent modules see [here|3rd Party Libraries] + + +h3. \[C++ store\] What is a RHM_IORES_ENQCAPTHRESH error? + +The journal ran out of space (ENQueue CAPacity THRESHold). The journal is a circular file buffer of fixed capacity set by the journal file size and number of files. When an attempt to write a record causes the journal to exceed an approx. 80% threshold, then the enqueue is rejected with this error code. Dequeues (a written record of a consumed message) may continue, however, as these free up space in the journal. Once space has been freed up, enqueues may continue as normal. + +This error may be caused by: +# The journal is too small for the size and number of messages being stored. The journal must be made large enough to hold all of the messages you expect to be on the queue at any one moment (a worst-case scenario). Make the journal capacity larger through the use of the *\-\-num\-jfiles* and *\-\-jfile\-size\-pgs* parameters. +{info:title=Rule of thumb for sizing the journal} +Make the journal twice the size of all the messages you need to store at any one moment in time. +{info} +# Messages are not being dequeued (consumed) as expected. Since the store is a circular file buffer, if one un-dequeued (not consumed) message remains, it can eventually "block" the storage of new messages as the buffer gets overwritten. + +h3. \[C++ store\] What is the TPL? What are the \-\-tpl\-\* options for? +The TPL stands for *Transaction Prepared List*. The store creates a single instance of a store for storing transaction boundaries called the Transaction Prepared List. Because the TPL is frequently flushed and has very different usage patterns to a normal store, it has been provided with its own set of configuration parameters: +* *\-\-tpl\-num\-jfiles:* The number of files in the TPL journal +* *\-\-tpl\-jfile\-size-pgs:* The file size in 64kiB blocks of the TPL journal. +* *\-\-tpl\-wcache\-page-size:* The size of the write cache in the TPL in KiB, which is typically set a lot smaller than the average message store. + +h1. How To + +h2. C++ + +h3. How to use RDMA with Qpid + +The RDMA plugin uses native OFED1.3 and puts AMQP directly onto the DMA. When using the RDMA plug-in for Qpid note the following +* IP over IB or Fibre needs to be setup for the initial negociation +* You need to make sure you have enough memory to pin for DMA use ulimit \-l something large +* you might need to edit /etc/security/limits.conf first then log in again + +Once you have it up and running, use latencytest to make sure it is working. You should see latencies between 50 and 80us round trip. + +h3. Message TTL, auto expire + +I need to be able to set time for a message that I send to be removed from the queue if it is not read by my subscriber. For example: I enqueue a message and I want it to be automatically dequeued after a certain amount of time has passed.Is there a feature like this in qpid? + +yes, the TTL can be set in the message headers and the messages get dequeued if TTL expires + +E.g. from c++: +{code} +Message m("Hello World!"); + m.getDeliveryProperties().setTtl(500); +{code} +Sets a 500 millisecond timeout. + +h3. How to install the qpid-tools for c++ broker? + +I see +{code} +[commands]$ ./qpid-queue-stats + Traceback (most recent call last): + File "./qpid-queue-stats", line 29, in + from qmf.console import Session, Console + ImportError: No module named qmf.console +{code} + +This problem occurs because the PYTHONPATH environment variable does not include the location of the qpid python files. If you are running from the SVN checkout, add <path>/qpid/python to PYTHONPATH (where <path> is the location of your SVN tree). If you are installing from source, make sure you configure with the same prefix where Python is installed. This is most likely: + +{code} +# configure --prefix=/usr +# make +# make install +{code} +If you are running from vendor RPMs, this should work automatically. + + +h2. Java +{children:page=Qpid Java How To}
Added: qpid/trunk/qpid/doc/book/src/LVQ.xml URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/doc/book/src/LVQ.xml?rev=898329&view=auto ============================================================================== --- qpid/trunk/qpid/doc/book/src/LVQ.xml (added) +++ qpid/trunk/qpid/doc/book/src/LVQ.xml Tue Jan 12 14:05:42 2010 @@ -0,0 +1,341 @@ +<?xml version="1.0" encoding="utf-8"?> +<chapter xmlns:html="http://www.w3.org/1999/xhtml";><title> + Apache Qpid : LVQ + </title> + + <section role="h2" id="LVQ-UnderstandingLVQ"><title> + Understanding LVQ + </title> + <para> + Last Value Queues are useful youUser Documentation are only + interested in the latest value entered into a queue. LVQ + semantics are typically used for things like stock symbol updates + when all you care about is the latest value for example. + </para><para> + Qpid C++ M4 or later supports two types of LVQ semantics: + </para><itemizedlist> + <listitem><para>LVQ + </para></listitem> + <listitem><para>LVQ_NO_BROWSE + </para></listitem> + </itemizedlist> + <!--h2--></section> + + + <section role="h2" id="LVQ-LVQsemantics-3A"><title> + LVQ semantics: + </title> + <para> + LVQ uses a header for a key, if the key matches it replaces the + message in-place in the queue except + a.) if the message with the matching key has been acquired + b.) if the message with the matching key has been browsed + In these two cases the message is placed into the queue in FIFO, + if another message with the same key is received it will the + 'un-accessed' message with the same key will be replaced + </para><para> + These two exceptions protect the consumer from missing the last + update where a consumer or browser accesses a message and an + update comes with the same key. + </para><para> + An example + </para> + <programlisting> +[localhost tests]$ ./lvqtest --mode create_lvq +[localhost tests]$ ./lvqtest --mode write +Sending Data: key1=key1.0x7fffdf3f3180 +Sending Data: key2=key2.0x7fffdf3f3180 +Sending Data: key3=key3.0x7fffdf3f3180 +Sending Data: key1=key1.0x7fffdf3f3180 +Sending Data: last=last +[localhost tests]$ ./lvqtest --mode browse +Receiving Data:key1.0x7fffdf3f3180 +Receiving Data:key2.0x7fffdf3f3180 +Receiving Data:key3.0x7fffdf3f3180 +Receiving Data:last +[localhost tests]$ ./lvqtest --mode write +Sending Data: key1=key1.0x7fffe4c7fa10 +Sending Data: key2=key2.0x7fffe4c7fa10 +Sending Data: key3=key3.0x7fffe4c7fa10 +Sending Data: key1=key1.0x7fffe4c7fa10 +Sending Data: last=last +[localhost tests]$ ./lvqtest --mode browse +Receiving Data:key1.0x7fffe4c7fa10 +Receiving Data:key2.0x7fffe4c7fa10 +Receiving Data:key3.0x7fffe4c7fa10 +Receiving Data:last +[localhost tests]$ ./lvqtest --mode consume +Receiving Data:key1.0x7fffdf3f3180 +Receiving Data:key2.0x7fffdf3f3180 +Receiving Data:key3.0x7fffdf3f3180 +Receiving Data:last +Receiving Data:key1.0x7fffe4c7fa10 +Receiving Data:key2.0x7fffe4c7fa10 +Receiving Data:key3.0x7fffe4c7fa10 +Receiving Data:last +</programlisting> +<!--h2--></section> + <section role="h2" id="LVQ-LVQNOBROWSEsemantics-3A"><title> + LVQ_NO_BROWSE + semantics: + </title> + <para> + LVQ uses a header for a key, if the key matches it replaces the + message in-place in the queue except + a.) if the message with the matching key has been acquired + In these two cases the message is placed into the queue in FIFO, + if another message with the same key is received it will the + 'un-accessed' message with the same key will be replaced + </para><para> + Note, in this case browsed messaged are not invalidated, so + updates can be missed. + </para><para> + An example + </para> + <programlisting> +[localhost tests]$ ./lvqtest --mode create_lvq_no_browse +[localhost tests]$ ./lvqtest --mode write +Sending Data: key1=key1.0x7fffce5fb390 +Sending Data: key2=key2.0x7fffce5fb390 +Sending Data: key3=key3.0x7fffce5fb390 +Sending Data: key1=key1.0x7fffce5fb390 +Sending Data: last=last +[localhost tests]$ ./lvqtest --mode write +Sending Data: key1=key1.0x7fff346ae440 +Sending Data: key2=key2.0x7fff346ae440 +Sending Data: key3=key3.0x7fff346ae440 +Sending Data: key1=key1.0x7fff346ae440 +Sending Data: last=last +[localhost tests]$ ./lvqtest --mode browse +Receiving Data:key1.0x7fff346ae440 +Receiving Data:key2.0x7fff346ae440 +Receiving Data:key3.0x7fff346ae440 +Receiving Data:last +[localhost tests]$ ./lvqtest --mode browse +Receiving Data:key1.0x7fff346ae440 +Receiving Data:key2.0x7fff346ae440 +Receiving Data:key3.0x7fff346ae440 +Receiving Data:last +[localhost tests]$ ./lvqtest --mode write +Sending Data: key1=key1.0x7fff606583e0 +Sending Data: key2=key2.0x7fff606583e0 +Sending Data: key3=key3.0x7fff606583e0 +Sending Data: key1=key1.0x7fff606583e0 +Sending Data: last=last +[localhost tests]$ ./lvqtest --mode consume +Receiving Data:key1.0x7fff606583e0 +Receiving Data:key2.0x7fff606583e0 +Receiving Data:key3.0x7fff606583e0 +Receiving Data:last +[localhost tests]$ + +</programlisting> + <!--h2--></section> + <section role="h2" id="LVQ-Examplesource"><title> + Example source + </title> + + <programlisting> + +/* + * + * 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 <qpid/client/AsyncSession.h> +#include <qpid/client/Connection.h> +#include <qpid/client/SubscriptionManager.h> +#include <qpid/client/Session.h> +#include <qpid/client/Message.h> +#include <qpid/client/MessageListener.h> +#include <qpid/client/QueueOptions.h> + +#include <iostream> + +using namespace qpid::client; +using namespace qpid::framing; +using namespace qpid::sys; +using namespace qpid; +using namespace std; + + +enum Mode { CREATE_LVQ, CREATE_LVQ_NO_BROWSE, WRITE, BROWSE, CONSUME}; +const char* modeNames[] = { "create_lvq","create_lvq_no_browse","write","browse","consume" }; + +// istream/ostream ops so Options can read/display Mode. +istream& operator>>(istream& in, Mode& mode) { + string s; + in >> s; + int i = find(modeNames, modeNames+5, s) - modeNames; + if (i >= 5) throw Exception("Invalid mode: "+s); + mode = Mode(i); + return in; +} + +ostream& operator<<(ostream& out, Mode mode) { + return out << modeNames[mode]; +} + +struct Args : public qpid::Options, + public qpid::client::ConnectionSettings +{ + bool help; + Mode mode; + + Args() : qpid::Options("Simple latency test optins"), help(false), mode(BROWSE) + { + using namespace qpid; + addOptions() + ("help", optValue(help), "Print this usage statement") + ("broker,b", optValue(host, "HOST"), "Broker host to connect to") + ("port,p", optValue(port, "PORT"), "Broker port to connect to") + ("username", optValue(username, "USER"), "user name for broker log in.") + ("password", optValue(password, "PASSWORD"), "password for broker log in.") + ("mechanism", optValue(mechanism, "MECH"), "SASL mechanism to use when authenticating.") + ("tcp-nodelay", optValue(tcpNoDelay), "Turn on tcp-nodelay") + ("mode", optValue(mode, "'see below'"), "Action mode." + "\ncreate_lvq: create a new queue of type lvq.\n" + "\ncreate_lvq_no_browse: create a new queue of type lvq with no lvq on browse.\n" + "\nwrite: write a bunch of data & keys.\n" + "\nbrowse: browse the queue.\n" + "\nconsume: consume from the queue.\n"); + } +}; + +class Listener : public MessageListener +{ + private: + Session session; + SubscriptionManager subscriptions; + std::string queue; + Message request; + QueueOptions args; + public: + Listener(Session& session); + void setup(bool browse); + void send(std::string kv); + void received(Message& message); + void browse(); + void consume(); +}; + +Listener::Listener(Session& s) : + session(s), subscriptions(s), + queue("LVQtester") +{} + +void Listener::setup(bool browse) +{ + // set queue mode + args.setOrdering(browse?LVQ_NO_BROWSE:LVQ); + + session.queueDeclare(arg::queue=queue, arg::exclusive=false, arg::autoDelete=false, arg::arguments=args); + +} + +void Listener::browse() +{ + subscriptions.subscribe(*this, queue, SubscriptionSettings(FlowControl::unlimited(), ACCEPT_MODE_NONE, ACQUIRE_MODE_NOT_ACQUIRED)); + subscriptions.run(); +} + +void Listener::consume() +{ + subscriptions.subscribe(*this, queue, SubscriptionSettings(FlowControl::unlimited(), ACCEPT_MODE_NONE, ACQUIRE_MODE_PRE_ACQUIRED)); + subscriptions.run(); +} + +void Listener::send(std::string kv) +{ + request.getDeliveryProperties().setRoutingKey(queue); + + std::string key; + args.getLVQKey(key); + request.getHeaders().setString(key, kv); + + std::ostringstream data; + data << kv; + if (kv != "last") data << "." << hex << this; + request.setData(data.str()); + + cout << "Sending Data: " << kv << "=" << data.str() << std::endl; + async(session).messageTransfer(arg::content=request); + +} + +void Listener::received(Message& response) +{ + + cout << "Receiving Data:" << response.getData() << std::endl; +/* if (response.getData() == "last"){ + subscriptions.cancel(queue); + } +*/ +} + +int main(int argc, char** argv) +{ + Args opts; + opts.parse(argc, argv); + + if (opts.help) { + std::cout << opts << std::endl; + return 0; + } + + Connection connection; + try { + connection.open(opts); + Session session = connection.newSession(); + Listener listener(session); + + switch (opts.mode) + { + case CONSUME: + listener.consume(); + break; + case BROWSE: + listener.browse(); + break; + case CREATE_LVQ: + listener.setup(false); + break; + case CREATE_LVQ_NO_BROWSE: + listener.setup(true); + break; + case WRITE: + listener.send("key1"); + listener.send("key2"); + listener.send("key3"); + listener.send("key1"); + listener.send("last"); + break; + } + connection.close(); + return 0; + } catch(const std::exception& error) { + std::cout << error.what() << std::endl; + } + return 1; +} + +</programlisting> +<!--h2--></section> +</chapter> --------------------------------------------------------------------- Apache Qpid - AMQP Messaging Implementation Project: http://qpid.apache.org Use/Interact: mailto:[email protected]
