AMQP Messaging Broker (Implemented in Java)

The java implementation of Qpid is a pure Java message broker that implements the AMQP protocol. Essentially, Qpid is a robust, performant middleware component that can handle your messaging traffic.

It currently supports the following features:

The broker comes with a script for unix/linux/cygwin called qpid-server, which can be found in the bin directory of the installed package. This command can be executed without any paramters and will then use the default configuration file provided on install.

For the Windows OS, please use qpid-server.bat.

There's no need to set your classpath for QPID as the scripts take care of that by adding jar's with classpath defining manifest files to your classpath.

For more information on running the broker please see our ??? page.

Please see the ??? documentation.

see ???

Here is a donated Spring configuration file appContext.zip which shows the config for Qpid side by side with Weblogic. HtH !

The system property

amqj.logging.level

can be used to configure the logging level. For the broker, you can use the environment variable AMQJ_LOGGING_LEVEL which is picked up by the qpid-run script (called by qpid-server to start the broker) at runtime.

For client code that you've written, simply pass in a system property to your command line to set it to the level you'd like i.e.

-Damqj.logging.level=INFO

The log level for the broker defaults to INFO if the env variable is not set, but you may find that your log4j properties affect this. Setting the property noted above should address this.

If you don't already have a logging implementation in your classpath you should add slf4-log4j12-1.4.0.jar and log4j-1.2.12.jar.

The broker configuration is contained in the <installed-dir>/etc/config.xml file. You can copy and edit this file and then specify your own configuration file as a parameter to the startup script using the -c flag i.e. qpid-server -c <your_config_file's_path>

For more detailed information on configuration, please see ???

The broker defaults to use port 5672 at startup for AMQP traffic. If the management interface is enabled it starts on port 8999 by default.

The JMX management interface actually requires 2 ports to operate, the second of which is indicated to the client application during connection initiation to the main (default: 8999) port. Previously this second port has been chosen at random during broker startup, however since Qpid 0.5 this has been fixed to a port 100 higher than the main port(ie Default:9099) in order to ease firewall navigation.

The broker defaults to use port 5672 at startup for AMQP traffic. The broker also uses port 8999 for the JMX Management interface.

To change the AMQP traffic port use the -p flag at startup. To change the management port use -m i.e. qpid-server -p <port_number_to_use> -m <port_number_to_use>

Use this to get round any issues on your host server with port 5672/8999 being in use/unavailable.

For additional details on what ports the broker uses see Section 1.2.1.11, “ What ports does the broker use? ” FAQ entry. For more detailed information on configuration, please see ???

The following command line options are available:

The following options are available:

You should login as user guest with password guest

You can configure queues which will be created at broker startup by tailoring a copy of the virtualhosts.xml file provided in the installed qpid-version/etc directory.

So, if you're using a queue called 'devqueue' you can ensure that it is created at startup by using an entry something like this:

<virtualhosts>
  <default>test</default>
  <virtualhost>
   <name>test</name>
   <test>
   <queue>
      <name>devqueue</name>
      <devqueue>
        <exchange>amq.direct</exchange>
        <maximumQueueDepth>4235264</maximumQueueDepth>  <!-- 4Mb -->
        <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
        <maximumMessageAge>600000</maximumMessageAge>  <!-- 10 mins -->
      </devqueue>
   </queue>
   </test>
  </virtualhost>
</virtualhosts>

Note that the name (in thie example above the name is 'test') element should match the virtualhost that you're using to create connections to the broker. This is effectively a namespace used to prevent queue name clashes etc. You can also see that we've set the 'test' virtual host to be the default for any connections which do not specify a virtual host (in the <default> tag).

You can amend the config.xml to point at a different virtualhosts.xml file by editing the <virtualhosts/> element.

So, for example, you could tell the broker to use a file in your home directory by creating a new config.xml file with the following entry:

<virtualhosts>/home/myhomedir/virtualhosts.xml</virtualhosts>

You can then pass this amended config.xml into the broker at startup using the -c flag i.e. qpid-server -c <path>/config.xml

Queues can be dynamically created at runtime by creating a consumer for them. After they have been created and bound (which happens automatically when a JMS Consumer is created) a publisher can send messages to them.

There are a number of tuning options available, please see the Section 2.8, “ How to Tune M3 Java Broker Performance ” page for more information.

At present, messages with an invalid routing key will be returned to the sender. If you register an exception listener for your publisher (easiest to do by making your publisher implement the ExceptionListener interface and coding the onException method) you'll see that you end up in onException in this case. You can expect to be catching a subclass of org.apache.qpid.AMQUndeliveredException.

If you simply start the Qpid broker using the default configuration, then the log file is written to $QPID_WORK/log/qpid.log

This is not ideal if you want to run several instances from one install, or acrhive logs to a shared drive from several hosts.

To make life easier, there are two optional ways to configure the naming convention used for the broker log.

The client code currently has various timeouts scattered throughout the code. These can cause your client to appear like it has hung when it is actually waiting for the timeout ot compelete. One example is when the broker becomes non-responsive, the client code has a hard coded 2 minute timeout that it will wait when closing a connection. These timeouts need to be consolidated and exposed. see ???

For general questions, please subscribe to the users@qpid.apache.org mailing list.

For development questions, please subscribe to the dev@qpid.apache.org mailing list.

More details on these lists are available on our ??? page.

You can do this via the ???. To do this simply log in to the management console as an admin user (you need to have created an admin account in the jmxremote.access file first) and then select the 'UserManagement' mbean. Select the user in the table and click the Set Password button. Alternatively, update the password file and use the management console to reload the file with the button at the bottom of the 'UserManagement' view. In both cases, this will take effect when the user next logs in i.e. will not cause them to be disconnected if they are already connected.

For more information on the Management Console please see our Section 3.1.1.5, “ Qpid JMX Management Console User Guide ”

Knowing that there is a consumer for a message is quite tricky. That said using the qpid.jms.Session#createProducer with immediate and mandatory set to true will get you part of the way there.

If you are publishing to a well known queue then immediate will let you know if there is any consumer able to pre-fetch that message at the time you send it. If not it will be returned to you on your connection listener.

If you are sending to a queue that the consumer creates then the mandatory flag will let you know if they have not yet created that queue.

These flags will not be able to tell you if the consuming application has received the message and is able to process it.

There are two possibilities here:

1) The management console can be used to interogate an active broker and browse the contents of a queue.See the ??? page for further details.

2) The ??? can be used to inspect the contents of a persistent message store. Note: this can currently only be used when the broker is offline.

You should check that you aren't sending persistent messages, this is the default. If you want to send transient messages you must explicitly set this option when instantiating your MessageProducer or on the send() method.

The Java broker does not currently implement producer flow control. Publishes are currently asynchronous, so there is no ability to rate limit this automatically. While this is something which will be addressed in the future, it is currently up to applications to ensure that they do not publish faster than the messages are being consumed for signifcant periods of time.

The broker can no longer store any more messages in memory. This is particular evident if you are using the MemoryMessageStore. To alleviate this issue you should ensure that your clients are consuming all the messages from the broker.

You may also want to increase the memory allowance to the broker though this will only delay the exception if you are publishing messages faster than you are consuming. See ??? for details of changing the memory settings.

If you get a stack trace like this when you try to publish, then you may have typo'd the exchange type in your queue or topic declaration. Open your virtualhosts.xml and check that the

<exchange>amq.direct</exchange>

2009-01-12 15:26:27,957 ERROR [pool-11-thread-2] protocol.AMQMinaProtocolSession (AMQMinaProtocolSession.java:365) - Unexpected exception while processing frame. Closing connection. 
java.lang.NullPointerException 
        at org.apache.qpid.server.security.access.PrincipalPermissions.authorise(PrincipalPermissions.java:398) 
        at org.apache.qpid.server.security.access.plugins.SimpleXML.authorise(SimpleXML.java:302) 
        at org.apache.qpid.server.handler.QueueBindHandler.methodReceived(QueueBindHandler.java:111) 
        at org.apache.qpid.server.handler.ServerMethodDispatcherImpl.dispatchQueueBind(ServerMethodDispatcherImpl.java:498) 
        at org.apache.qpid.framing.amqp_8_0.QueueBindBodyImpl.execute(QueueBindBodyImpl.java:167) 
        at org.apache.qpid.server.state.AMQStateManager.methodReceived(AMQStateManager.java:204) 
        at org.apache.qpid.server.protocol.AMQMinaProtocolSession.methodFrameReceived(AMQMinaProtocolSession.java:295) 
        at org.apache.qpid.framing.AMQMethodBodyImpl.handle(AMQMethodBodyImpl.java:93) 
        at org.apache.qpid.server.protocol.AMQMinaProtocolSession.frameReceived(AMQMinaProtocolSession.java:235) 
        at org.apache.qpid.server.protocol.AMQMinaProtocolSession.dataBlockReceived(AMQMinaProtocolSession.java:191) 
        at org.apache.qpid.server.protocol.AMQPFastProtocolHandler.messageReceived(AMQPFastProtocolHandler.java:244) 
        at org.apache.mina.common.support.AbstractIoFilterChain$TailFilter.messageReceived(AbstractIoFilterChain.java:703) 
        at org.apache.mina.common.support.AbstractIoFilterChain.callNextMessageReceived(AbstractIoFilterChain.java:362) 
        at org.apache.mina.common.support.AbstractIoFilterChain.access$1200(AbstractIoFilterChain.java:54) 
        at org.apache.mina.common.support.AbstractIoFilterChain$EntryImpl$1.messageReceived(AbstractIoFilterChain.java:800) 
        at org.apache.qpid.pool.PoolingFilter.messageReceived(PoolingFilter.java:371) 
        at org.apache.mina.filter.ReferenceCountingIoFilter.messageReceived(ReferenceCountingIoFilter.java:96) 
        at org.apache.mina.common.support.AbstractIoFilterChain.callNextMessageReceived(AbstractIoFilterChain.java:362) 
        at org.apache.mina.common.support.AbstractIoFilterChain.access$1200(AbstractIoFilterChain.java:54) 
        at org.apache.mina.common.support.AbstractIoFilterChain$EntryImpl$1.messageReceived(AbstractIoFilterChain.java:800) 
        at org.apache.mina.filter.codec.support.SimpleProtocolDecoderOutput.flush(SimpleProtocolDecoderOutput.java:60) 
        at org.apache.mina.filter.codec.QpidProtocolCodecFilter.messageReceived(QpidProtocolCodecFilter.java:174) 
        at org.apache.mina.common.support.AbstractIoFilterChain.callNextMessageReceived(AbstractIoFilterChain.java:362) 
        at org.apache.mina.common.support.AbstractIoFilterChain.access$1200(AbstractIoFilterChain.java:54) 
        at org.apache.mina.common.support.AbstractIoFilterChain$EntryImpl$1.messageReceived(AbstractIoFilterChain.java:800) 
        at org.apache.qpid.pool.Event$ReceivedEvent.process(Event.java:86) 
        at org.apache.qpid.pool.Job.processAll(Job.java:110) 
        at org.apache.qpid.pool.Job.run(Job.java:149) 
        at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:885) 
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907) 
        at java.lang.Thread.run(Thread.java:619) 

These threads are part of the thread pool used by Mina to process the socket. In the future we may provide tuning guidelines but at this point we have seen no performance implications from the current configuration. As the threads are part of a pool they should remain inactive until required.

You have not configured the console's SSL trust store properly, see ??? for more details.

See ???

There are two main Qpid environment variables which are required to be set for Qpid deployments, QPID_HOME and QPID_WORK.

QPID_HOME - This variable is used to tell the Qpid broker where it's installed home is, which is in turn used to find dependency JARs which Qpid uses.

QPID_WORK - This variable is used by Qpid when creating all 'writeable' directories that it uses. This includes the log directory and the storage location for any BDB instances in use by your deployment (if you're using persistence with BDB). If you do not set this variable, then the broker will default (in the qpid-server script) to use the current user's homedir as the root directory for creating the writeable locations that it uses.

If you simply start the Qpid broker, it will default to use a -Xmx setting of 1024M for the broker JVM. However, we would recommend that you make the maximum -Xmx heap size available, if possible, of 3Gb (for 32-bit platforms).

You can control the memory setting for your broker by setting the QPID_JAVA_MEM variable before starting the broker e.g. -Xmx3668m . Enclose your value within quotes if you also specify a -Xms value. The value in use is echo'd by the qpid-server script on startup.

A configuration section has two entries that can be used to identify how the configuration will be applied: 'name' and 'subscriptionName'.

            
                     <topic>
                         <name>stocks.us</name>        
                

                     <topic>
                         <subscriptionName>clientid:mysubscription</subscriptionName>        
            

It is also possible to combine these two identifiers to specify a unique subscription to a given topic.

                    <topic>
                        <name>stocks.us</name>
                        <subscriptionName>clientid:mysubscription</subscriptionName>
            

Currently only one element of the designed configuration is processed, that of the slow consumer detection. This is setup as below using the 'slow-consumer-detection' element. There are two required types of tag, first the trigger, which is one of 'depth', 'messageAge' or 'messageCount' and secondly the 'policy'.

	    <slow-consumer-detection>
                <!-- The maximum depth before which the policy will be applied-->
                <depth>4235264</depth>

                <!-- The maximum message age before which the policy will be applied-->
                <messageAge>600000</messageAge>

                <!-- The maximum number of message before which the policy will be applied-->
                <messageCount>50</messageCount>

                <!-- Policy Selection -->
                <policy name="TopicDelete"/>
            </slow-consumer-detection>
        

The trigger is used to determine when the policy should be applied. Currently we have a simple policy 'topicdelete', this will disconnect consumers of topics where their consumption rate falls sufficiently to hit one of the trigger values.

As of 0.8 the topic configuration is limited to straight string matching. This means that given the following two topic configuring sections for 'stocks.us' and 'stocks.*' a subscription for 'stocks.uk' will not match the expected 'stocks.*'. Nor will any additional configuration listed in 'stocks.*' affect any 'stocks.us' subscriptions.

            <topics>
                <topic>
            	    <name>stocks.us</name>
	                ...
	            </topic>
    	        <topic>
	               <name>stocks.*</name>
	               ...
                </topic>
            </topics>
    

A subscription for 'stocks.us' will only receive configuration settings that are defined in the 'stocks.us' section.

The plain file has the following format:

# Plain password authentication file.
# default name : passwd
# Format <username>:<password>
#e.g.
martin:password

As the contents of the file are plain text and the password is taken to be everything to the right of the ':'(colon). The password, therefore, cannot contain a ':' colon, but this can be used to delimit the password.

Lines starting with a '#' are treated as comments.

The location of the password file in use for your broker is as configured in your config.xml file.

<principal-databases>
            <principal-database>
                <name>passwordfile</name>
                <class>org.apache.qpid.server.security.auth.database.PlainPasswordFilePrincipalDatabase</class>
                <attributes>
                    <attribute>
                        <name>passwordFile</name>
                        <value>${conf}/passwd</value>
                    </attribute>
                </attributes>
            </principal-database>
        </principal-databases>

So in the example config.xml file this password file lives in the directory specified as the conf directory (at the top of your config.xml file).

If you wish to use Base64 encoding for your password file, then in the <class> element above you should specify org.apache.qpid.server.security.auth.database.Base64MD5PasswordFilePrincipalDatabase

The default is:

 <conf>${prefix}/etc</conf>

This format can be used to ensure that SAs cannot read the plain text password values from your password file on disk.

The Base64MD5 file uses the following format:

# Base64MD5 password authentication file
# default name : qpid.passwd
# Format <username>:<Base64 Encoded MD5 hash of the users password>
#e.g.
martin:X03MO1qnZdYdgyfeuILPmQ==

As with the Plain format the line is delimited by a ':'(colon). The password field contains the MD5 Hash of the users password encoded in Base64.

This file is read on broker start-up and is not re-read.

To update the file there are two options:

Suppose you wish to permission two users: a user 'operator' must be able to perform all Management operations, and a user 'readonly' must be enable to perform only read-only functions. Neither 'operator' nor 'readonly' should be allow to connect for messaging.

        # Give operator permission to execute all JMX Methods
        ACL ALLOW operator ALL METHOD
        # Give operator permission to execute only read-only JMX Methods
        ACL ALLOW readonly ACCESS METHOD
        # Deny operator/readonly permission to perform messaging.
        ACL DENY operator ACCESS VIRTUALHOST
        ACL DENY readonly ACCESS VIRTUALHOST
        ...
        ... rules for other users
        ...
        # Explicitly deny all (log) to eveyone 
        ACL DENY-LOG ALL ALL
      

Suppose you wish to restrict User Management operations to users belonging to a group 'usermaint'. No other user is allowed to perform user maintainence This example illustrates the permissioning of a individual component and a group definition.

        # Create a group usermaint with members bob and alice
        GROUP usermaint bob alice
        # Give operator permission to execute all JMX Methods
        ACL ALLOW usermaint ALL METHOD component="UserManagement"
        ACL DENY ALL ALL METHOD component="UserManagement"
        ...
        ... rules for other users
        ...
        ACL DENY-LOG ALL ALL
      

Suppose you wish to permission a system using a request/response paradigm. Two users: 'client' publishes requests; 'server' consumes the requests and generates a response. This example illustrates the permissioning of AMQP exchanges and queues.

        # Allow client and server to connect to the virtual host.
        ACL ALLOW client ACCESS VIRTUALHOST
        ACL ALLOW server ACCESS VIRTUALHOST

        # Client side
        # Allow the 'client' user to publish requests to the request queue. As is the norm for the request/response paradigm, the client
        # is required to create a temporary queue on which the server will response.  Consequently, there are rules to allow the creation
        # of the temporary queues and consumption of messages from it.
        ACL ALLOW client CREATE QUEUE temporary="true"
        ACL ALLOW client CONSUME QUEUE temporary="true"
        ACL ALLOW client DELETE QUEUE temporary="true"
        ACL ALLOW client BIND EXCHANGE name="amq.direct" temporary="true"
        ACL ALLOW client UNBIND EXCHANGE name="amq.direct" temporary="true"
        ACL ALLOW client PUBLISH EXCHANGE name="amq.direct" routingKey="example.RequestQueue"
        
        # Server side
        # Allow the 'server' user to consume from the request queue and publish a response to the temporary response queue created by
        # client.  We also allow the server to create the request queue.
        ACL ALLOW server CREATE QUEUE name="example.RequestQueue"
        ACL ALLOW server CONSUME QUEUE name="example.RequestQueue"
        ACL ALLOW server BIND EXCHANGE
        ACL ALLOW server PUBLISH EXCHANGE name="amq.direct" routingKey="TempQueue*"
        
        ACL DENY-LOG all all
      

The broker configuration file (config.xml) needs to be updated to include the SSL keystore location details.

<!-- Additions required to Connector Section -->

<ssl>
    <enabled>true</enabled>
    <sslOnly>true</sslOnly>
    <keyStorePath>/path/to/keystore.ks</keyStorePath>
    <keyStorePassword>keystorepass</keyStorePassword>
</ssl>

The sslOnly option is included here for completeness however this will disable the unencrypted port and leave only the SSL port listening for connections.

The best place to start looking is class SSLConfiguration this is provided to the connection during creation however there is currently no example that demonstrates its use.

The virtualhosts.xml which currently ships as part of the Qpid distribution is really targeted at development use, and supports various artifacts commonly used by the Qpid development team.

As a result, it is reasonably complex. In the example XML below, I have tried to simplify one example virtual host setup which is possibly more useful for new users of Qpid or development teams looking to simply make use of the Qpid broker in their deployment.

I have also added some inline comments on each section, which should give some extra information on the purpose of the various elements.

<virtualhosts>
    <!-- Sets the default virtual host for connections which do not specify a vh -->
    <default>localhost</default>
    <!-- Define a virtual host and all it's config -->
    <virtualhost>
        <name>localhost</name>
        <localhost>    
            <!-- Define the types of additional AMQP exchange available for this vh -->   
            <!-- Always get amq.direct (for queues) and amq.topic (for topics) by default -->     
            <exchanges>
                <!-- Example of declaring an additional exchanges type for developer use only -->
                <exchange>
                    <type>direct</type>
                    <name>test.direct</name>
                    <durable>true</durable>
                </exchange>
            </exchanges>
             
            <!-- Define the set of queues to be created at broker startup -->
            <queues>
                <!-- The properties configured here will be applied as defaults to all -->
                <!-- queues subsequently defined unless explicitly overridden -->
                <exchange>amq.direct</exchange>
                <!-- Set threshold values for queue monitor alerting to log --> 
                <maximumQueueDepth>4235264</maximumQueueDepth>  <!-- 4Mb -->
                <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
                <maximumMessageAge>600000</maximumMessageAge>  <!-- 10 mins -->

                <!-- Define a queue with all default settings -->   
                <queue>
                    <name>ping</name>
                </queue>
                <!-- Example definitions of queues with overriden settings -->
                <queue>
                    <name>test-queue</name>
                    <test-queue>
                        <exchange>test.direct</exchange>
                        <durable>true</durable>
                    </test-queue>
                </queue>
                <queue>
                    <name>test-ping</name>
                    <test-ping>
                        <exchange>test.direct</exchange>
                    </test-ping>
                </queue>
            </queues>
        </localhost>
    </virtualhost>
</virtualhosts>

Note that the config.xml file shipped as an example (or developer default) in the Qpid distribution contains an element which defines the path to the virtualhosts.xml.

When using your own virtualhosts.xml you must edit this path to point at the location of your file.

deprecation notice Version 0.6 of the Java broker includes ??? functionality which improves upon these messages and as such enabling status logging would be more beneficial. The configuration file has been left here for assistence with broker versions prior to 0.6.

The goals of this configuration are to record:

An additional goal of this configuration is to minimise any impact to the 'message flow' path. So it should not adversely affect production systems.

<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
    <appender name="FileAppender" class="org.apache.log4j.FileAppender">
        <param name="File" value="${QPID_WORK}/log/${logprefix}qpid${logsuffix}.log"/>
        <param name="Append" value="false"/>

        <layout class="org.apache.log4j.PatternLayout">
            <param name="ConversionPattern" value="%d %-5p [%t] %C{2} (%F:%L) - %m%n"/>
        </layout>

    </appender>

    <appender name="STDOUT" class="org.apache.log4j.ConsoleAppender">

        <layout class="org.apache.log4j.PatternLayout">
            <param name="ConversionPattern" value="%d %-5p [%t] %C{2} (%F:%L) - %m%n"/>
        </layout>
    </appender>

    <category name="Qpid.Broker">

        <priority value="debug"/>
    </category>


    <!-- Provide warnings to standard output -->
    <category name="org.apache.qpid">
        <priority value="warn"/>
    </category>


    <!-- Connection Logging -->

    <!-- Log details of client starting connection -->
    <category name="org.apache.qpid.server.handler.ConnectionStartOkMethodHandler">
        <priority value="info"/>
    </category>
    <!-- Log details of client closing connection -->
    <category name="org.apache.qpid.server.handler.ConnectionCloseMethodHandler">
        <priority value="info"/>
    </category>
    <!-- Log details of client responding to be asked to closing connection -->

    <category name="org.apache.qpid.server.handler.ConnectionCloseOkMethodHandler">
        <priority value="info"/>
    </category>


    <!-- Consumer Logging -->
    <!-- Provide details of Consumers connecting-->
    <category name="org.apache.qpid.server.handler.BasicConsumeMethodHandler">
        <priority value="debug"/>
    </category>

    <!-- Provide details of Consumers disconnecting, if the call it-->
    <category name="org.apache.qpid.server.handler.BasicCancelMethodHandler">
        <priority value="debug"/>
    </category>
    <!-- Provide details of when a channel closes to attempt to match to the Consume as a Cancel is not always issued-->
    <category name="org.apache.qpid.server.handler.ChannelCloseHandler">
        <priority value="info"/>
    </category>

    <!-- Provide details of Consumers starting to consume-->
    <category name="org.apache.qpid.server.handler.ChannelFlowHandler">
        <priority value="debug"/>
    </category>
    <!-- Provide details of what consumers are going to be consuming-->
    <category name="org.apache.qpid.server.handler.QueueBindHandler">
        <priority value="info"/>
    </category>

    <!-- No way of determining if publish message is returned, client log should show it.-->

    <root>
        <priority value="debug"/>
        <appender-ref ref="STDOUT"/>      
        <appender-ref ref="FileAppender"/>
    </root>

</log4j:configuration>

This is the most often asked for set of configuration. The goals of this configuration are to record:

NOTE: This configuration enables message logging on the 'message flow' path so should only be used were message volume is low. Every message that is sent to the broker will generate at least four logging statements

<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
    <appender name="FileAppender" class="org.apache.log4j.FileAppender">
        <param name="File" value="${QPID_WORK}/log/${logprefix}qpid${logsuffix}.log"/>
        <param name="Append" value="false"/>

        <layout class="org.apache.log4j.PatternLayout">
            <param name="ConversionPattern" value="%d %-5p [%t] %C{2} (%F:%L) - %m%n"/>
        </layout>

    </appender>

    <appender name="STDOUT" class="org.apache.log4j.ConsoleAppender">

        <layout class="org.apache.log4j.PatternLayout">
            <param name="ConversionPattern" value="%d %-5p [%t] %C{2} (%F:%L) - %m%n"/>
        </layout>
    </appender>

    <category name="Qpid.Broker">

        <priority value="debug"/>
    </category>


    <!-- Provide warnings to standard output -->
    <category name="org.apache.qpid">
        <priority value="warn"/>
    </category>


    <!-- Connection Logging -->

    <!-- Log details of client starting connection -->
    <category name="org.apache.qpid.server.handler.ConnectionStartOkMethodHandler">
        <priority value="info"/>
    </category>
    <!-- Log details of client closing connection -->
    <category name="org.apache.qpid.server.handler.ConnectionCloseMethodHandler">
        <priority value="info"/>
    </category>
    <!-- Log details of client responding to be asked to closing connection -->

    <category name="org.apache.qpid.server.handler.ConnectionCloseOkMethodHandler">
        <priority value="info"/>
    </category>

    <!-- Consumer Logging -->
    <!-- Provide details of Consumers connecting-->
    <category name="org.apache.qpid.server.handler.BasicConsumeMethodHandler">
        <priority value="debug"/>
    </category>

    <!-- Provide details of Consumers disconnecting, if the call it-->
    <category name="org.apache.qpid.server.handler.BasicCancelMethodHandler">
        <priority value="debug"/>
    </category>
    <!-- Provide details of when a channel closes to attempt to match to the Consume as a Cancel is not always issued-->
    <category name="org.apache.qpid.server.handler.ChannelCloseHandler">
        <priority value="info"/>
    </category>

    <!-- Provide details of Consumers starting to consume-->
    <category name="org.apache.qpid.server.handler.ChannelFlowHandler">
        <priority value="debug"/>
    </category>
    <!-- Provide details of what consumers are going to be consuming-->
    <category name="org.apache.qpid.server.handler.QueueBindHandler">
        <priority value="info"/>
    </category>

    <!-- No way of determining if publish message is returned, client log should show it.-->

    <!-- WARNING DO NOT ENABLE THIS IN PRODUCTION -->
    <!-- Will generate minimum one log statements per published message -->
    <!-- Will generate will log receiving of all body frame, count will vary on size of message.-->
    <!-- Empty Message = no body, Body is up to 64kb of data -->
    <!-- Will generate three log statements per recevied message -->

    <!-- Log messages flow-->
    <category name="org.apache.qpid.server.AMQChannel">

        <priority value="debug"/>
    </category>

    <root>
        <priority value="debug"/>
        <appender-ref ref="STDOUT"/>      
        <appender-ref ref="FileAppender"/>
    </root>

</log4j:configuration>

It is worth testing if changing the size of the Qpid read/write thread pool improves performance (eg. by setting JAVA_OPTS="-Damqj.read_write_pool_size=32" before running qpid-server). By default this is equal to the number of CPU cores, but a higher number may show better performance with some work loads.

It is also important to note that you should give the Qpid broker plenty of memory - for any serious application at least a -Xmx of 3Gb. If you are deploying on a 64 bit platform, a larger heap is definitely worth testing with. We will be testing tuning options around a larger heap shortly.

Firstly, check the source for Qpid out of our subversion repository:

???

For the broker code you need JDK 1.5.0_15 or later. You should set JAVA_HOME and include the bin directory in your PATH.

Check it's ok by executing java -v !

If you are wanting to run the python tests against the broker you will of course need a version of python.

Currently the Qpid java project builds using ant.

The ant build system is set up in a modular way, with a top level build script and template for module builds and then a module level build script which inherits from the template.

So, at the top level there are:

Then, in each module subdirectory there is:

The main build targets you are probably interested in are:

So, if you just want to compile everything you should run the build target in the top level build.xml file.

If you want to build an installable version of Qpid, run the archive task from the top level build.xml file.

If you want to compile an individual module, simply run the build target from the appropriate module e.g. to compile the broker source

1. Run the ant build from the root directory of Java trunk. 2. New project -> create from existing file system for broker, common, client, junit-toolkit, perftests, systests and each directory under management 4. Add the contents of lib/ to the build path 5. Setup Generated Code 6. Setup Dependencies

for i in `find . -name build.xml` ; do echo "$i:"; grep module.depends $i ; done

If you want to run your built Qpid package, see our ??? for details of how to do that.

If you want to run our tests, you can use the ant test or testreport (produces a useful report) targets.

Qpid clients receive buffered messages in batches, sized according to the pre-fetch value. The current default is 5000.

However, if you use the default value you will probably not see desirable behaviour with messages of different priority. This is because a message arriving after the pre-fetch buffer has filled will not leap frog messages of lower priority. It will be delivered at the front of the next batch of buffered messages (if that is appropriate), but this is most likely NOT what you need.

So, you need to set the prefetch values for your client (consumer) to make this sensible. To do this set the java system property max_prefetch on the client environment (using -D) before creating your consumer.

Setting the Qpid pre-fetch to 1 for your client means that message priority will be honoured by the Qpid broker as it dispatches messages to your client. A default for all client connections can be set via a system property:

-Dmax_prefetch=1

The prefetch can be also be adjusted on a per connection basis by adding a 'maxprefetch' value to the ???

amqp://guest:guest@client1/development?maxprefetch='1'&brokerlist='tcp://localhost:5672'

There is a slight performance cost here if using the receive() method and you could test with a slightly higher pre-fetch (up to 10) if the trade-off between throughput and prioritisation is weighted towards the former for your application. (If you're using OnMessage() then this is not a concern.)

If you are using the receive() method to consume messages then you should also only use one consumer per session with priority queues. If you're using OnMessage() then this is not a concern.

You can configure the frequency with which the SCD process will check for slow consumers, along with the unit of time used to specify that frequency.

The virtualhosts.virtualhost.hostname.slow-consumer-detection elements delay and timeunit are used to specify the frequency and timeunit respectively in the virtualhosts.xml file e.g.

<virtualhosts>
	<default>test</default>
	<virtualhost>
		<name>test</name>
		<test>
		   <slow-consumer-detection>
			<delay>60<delay/>
			<timeunit>seconds<timeunit/>
		<slow-consumer-detection/>
		</test>
	</virtualhost>
</virtualhosts>

When the SCD component finds a queue with a configured threshold to check, the operational logging component (if enabled) will output the following line:

    SCD-1003 : Checking Status of Queue
    

One key feature of SCD is the disconnection of a consuming client when a specified threshold is exceeded. For a pub-sub model using topics, this means that messages will no longer be delivered to the private queue which was associated with that consuming client, thus reducing any associated backlog in the broker.

For durable subscriptions, simply disconnecting the consuming client will not suffice since the associated queue is by definition durable and messages would continue to flow to it after disconnection, potentially worsening any backing up of data on the broker.

The solution is to configure durable subscriptions to delete the underlying queue on disconnection. This means that messages will no longer be delivered to the private queue associated with the subscription, thus preventing any backlog.

Full details of how to configure the thresholds are provided below.

You can configure SCD to be triggered on a topic or subscription when the oldest message in the associated private queue for the consumer ages beyond the specified value, in milliseconds.

You can opt to use the depth of the queue in bytes as a threshold. SCD will be triggered by a queue depth greater than the threshold specified i.e. when a broker receives a message that takes the queue depth over the threshold.

You can use the message count for the consumer's queue as the trigger, where a count higher than that specified will trigger disconnection.

You can configure the policy you wish to apply in your broker configuration. There are currently 2 policies available:

Delete Temporary Queues Only

If you do not specify a <topicDelete/> element in your configuration, then only temporary queues associated with a topic subscription will be deleted on client disconnect. This is the default behaviour.

Delete Durable Subscription Queues

If you add the <topicDelete/> element with the sub-element <delete-persistent/> to your config, then the persistent queue which is associated with durable subscriptions to a topic will also be deleted. This is an important consideration since without deleting the underlying queue the client's unconsumed data will grow indefinitely while they will be unable to reconnect to that queue due to the SCD threshold configured, potentially having an adverse effect on the application or broker in use.

Example Topic Configuration

The following steps are required to configure SCD:

The example below shows a simple definition, with all three thresholds specified and a simple disconnection, with deletion of any temporary queue, defined.

For a durable subscription to this topic, no queue deletion would be applied on disconnect - which is likely to be undesirable (see section above).

<topics>
	 <topic>
	 <name>stocks.us.*</name>
		 <slow-consumer-detection>
			 <!-- The maximum depth before which -->
			 <!-- the policy will be applied-->
			 <depth>4235264</depth>
			 <!-- The maximum message age before which -->
			 <!-- the policy will be applied-->
			 <messageAge>600000</messageAge>
			 <!-- The maximum number of message before -->
			 <!-- which the policy will be applied-->
			 <messageCount>50</messageCount>
			 <!-- Policy Selection -->
			 <policy name="TopicDelete"/>
		 </slow-consumer-detection>
	 </topic>
</topics>

The Qpid Java broker has a single source of users for the system. So a user can connect to the broker to send messages and via the JMX console to check the state of the broker.

<username>=<privilege>

#Generated by JMX Console : Last edited by user:admin
#Tue Jun 12 16:46:39 BST 2007
admin=admin
guest=readonly
user=readwrite

Management tool: See our ??? for details of how to use various console options with the Qpid management features.

The management of QPID is categorised into following types-

 1) Managing and Monitoring Exchanges: Following is the list of features, which we can have available for managing and monitoring an Exchange running on a Qpid Server Domain-

2) Managing and Monitoring Queues:  Following are the features, which we can have for a Queue on a Qpid Server Domain-

3) Managing and Monitoring Connections: Following are the features, which we can have for a connection on a QPID Server Domain-

4) Managing the Broker: Features for the Broker-

Each of the MessageStore implementations provide different back end storage for their messages and so would need a different tool to be able to interrogate their contents at the back end.

What this tool does is to utilise the Java broker code base to access the contents of the storage providing the user with a consistent means to inspect the storage contents in broker memory. The tool allows the current messages in the store to be inspected and copied/moved between queues. The tool uses the message instance in memory for all its access paths, but changes made will be reflected in the physical store (if one exists).

The tools-distribution currently includes a unix shell command 'msTool.sh' this script will launch the java tool.

The tool loads $QPID_HOME/etc/config.xml by default. If an alternative broker configuration is required this should be provided on the command line as would be done for the broker.

msTool.sh -c <path to different config.xml>

On startup the user is present with a command prompt

$ msTool.sh
MessageStoreTool - for examining Persistent Qpid Broker MessageStore instances
bdb$ 

The available commands in the tool can be seen through the use of the 'help' command.

bdb$ help
+----------------------------------------------------------------+
|                       Available Commands                       |
+----------------------------------------------------------------+
| Command | Description                                          |
+----------------------------------------------------------------+
| quit    | Quit the tool.                                       |
| list    | list available items.                                |
| dump    | Dump selected message content. Default: show=content |
| load    | Loads specified broker configuration file.           |
| clear   | Clears any selection.                                |
| show    | Shows the messages headers.                          |
| select  | Perform a selection.                                 |
| help    | Provides detailed help on commands.                  |
+----------------------------------------------------------------+
bdb$

A brief description is displayed and further usage information is shown with 'help <command>'

bdb$ help list
list availble items.
Usage:list queues [<exchange>] | exchanges | bindings [<exchange>] | all
bdb$

Currently the tool only works whilst the broker is offline i.e. it is up, but not accepting AMQP connections. This requires a stop/start of the broker. If this functionality was incorporated into the broker then a telnet functionality could be provided allowing online management.

At the very beginning please build Apache Qpid by refering this installation guide from here ???.

After successfully build Apache Qpid you'll be able to start Apache Qpid Java broker,then only you are in a position to use Qpid CLI.

First check out the source from subversion repository. Please visit the following link for more information about different versions of Qpid CLI.

???

For the broker code you need JDK 1.5.0_15 or later. You should set JAVA_HOME and include the bin directory in your PATH.

Check it's ok by executing java -v !

This project is currently having only an ant build system.Please install ant build system before trying to install Qpid CLI.

To compile the source please run following command

ant compile 

To compile the test source run the following command

ant compile-tests 

After successful compilation set QPID_CLI environment variable to the main source directory.(set the environment variable to the directory where ant build script stored in the SVN checkout).Please check whether the Qpid Java broker is up an running in the appropriate location and run the following command to start the Qpid CLI by running the qpid-cli script in the bin directory.

$QPID_CLI/bin/qpid-cli -h <hostname of the broker> -p <broker running port> For more details please have a look in to README file which ships with source package of Qpid CLI.

For now we are supporting those ant targets.