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.

I would take a look at the testPassiveTTL in TimeToLiveTest.java

The setUp and tearDown methods show how to correctly start up a broker for InVM testing. If you write your tests using a file for the JNDI you can then very easily swap between running your tests InVM and against a real broker.

See our ??? on how to confgure it

Basically though you just need to set two System Properites:

java.naming.factory.initial = org.apache.qpid.jndi.PropertiesFileInitialContextFactory java.naming.provider.url = <your JNDI file>

and call getInitialContext() in your code.

You will of course need to have the broker libraries on your class path for this to run.

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.

Certain operations wait for a response from the Server. One such operations is commit. If the server does not respond to the commit request within a set time a Request Timeout [error code: 408] exception is thrown (Server did not respond in a timely fashion). This is to ensure that a server that has hung does not cause the client process to be come unresponsive.

However, it is possible that the server just needs a long time to process a give request. For example, sending a large persistent message when using a persistent store will take some time to a) Transfer accross the network and b) to be fully written to disk.

These situations require that the default timeout value be increased. A cilent ??? 'amqj.default_syncwrite_timeout' can be set on the client to increase the wait time. The default in 0.5 is 30000 (30s).

See ???